MessageFile implements Closeable { public long append(long timestamp, String routingKey, ReadableByteChannel payload, int payloadSize) throws IOException { int n = routingKey.length(); if (n > 255) throw new IllegalArgumentException("Routing key length " + n + " > 255 characters"); byte[] routingKeyBytes = routingKey.getBytes(UTF8); synchronized (channel) { if (length + MESSAGE_HEADER_SIZE + routingKeyBytes.length + payloadSize > maxFileSize) return -1; header.clear(); channel.position(length); header.put(TYPE_MESSAGE); header.putLong(timestamp); header.putShort((short)routingKeyBytes.length); header.putInt(payloadSize); header.put(routingKeyBytes); header.flip(); int id = length - FILE_HEADER_SIZE; channel.write(header); long sz = channel.transferFrom(payload, channel.position(), payloadSize); if (sz != payloadSize) { throw new IOException("Only read " + sz + " bytes from payload channel instead of " + payloadSize); } length = (int)channel.position() + payloadSize; if (bucketIndex < 0 || ((id - bucketMessageId >= bytesPerBucket) && bucketIndex < MAX_BUCKETS - 1)) { if (bucketIndex >= 0) { putBucketDataInFileHeader(); ++bucketIndex; } else { bucketIndex = 0; } bucketMessageId = id; bucketTimestamp = timestamp; bucketCount = 1; } else { ++bucketCount; } mostRecentTimestamp = timestamp; return firstMessageId + id; } } @SuppressWarnings("StatementWithEmptyBody")  MessageFile(File file, long firstMessageId); @SuppressWarnings("StatementWithEmptyBody")  MessageFile(File file, long firstMessageId, int maxFileSize); File getFile(); long getFirstMessageId(); long getNextMessageId(); long append(long timestamp, String routingKey, ReadableByteChannel payload, int payloadSize); static int getMessageSize(String routingKey, int payloadSize); int length(); void checkpoint(boolean force); void use(); void closeIfUnused(); @Override void close(); boolean isOpen(); @Override String toString(); long getMostRecentTimestamp(); int getMessageCount(); Timeline getTimeline(); int getBucketCount(); Bucket getBucket(int i); int findBucket(long messageId); int findBucketByTimestamp(long timestamp); @SuppressWarnings("StatementWithEmptyBody") MessageCursor cursor(long messageId); @SuppressWarnings("StatementWithEmptyBody") MessageCursor cursorByTimestamp(long timestamp); static final int FILE_HEADER_SIZE; }
@Test public void testAppend() throws IOException { File file = new File(dir, "append-from-channel.qdb"); file.delete(); MessageFile mf = new MessageFile(file, 1000L, 1000000); long ts0 = System.currentTimeMillis(); String key0 = "foo"; byte[] payload0 = "piggy".getBytes("UTF8"); int length0 = 1 + 8 + 2 + 4 + key0.length() + payload0.length; long ts1 = ts0 + 1; String key1 = "foobar"; byte[] payload1 = "oink".getBytes("UTF8"); int length1 = 1 + 8 + 2 + 4 + key1.length() + payload1.length; assertEquals(1000L, mf.append(ts0, key0, toChannel(payload0), payload0.length)); assertEquals(1000L + length0, mf.append(ts1, key1, toChannel(payload1), payload1.length)); int expectedLength = 4096 + length0 + length1; assertEquals(expectedLength, mf.length()); mf.close(); assertEquals(expectedLength, file.length()); DataInputStream ins = new DataInputStream(new FileInputStream(file)); assertEquals((short)0xBE01, ins.readShort()); assertEquals((short)0, ins.readShort()); assertEquals(1000000, ins.readInt()); assertEquals(expectedLength, ins.readInt()); assertEquals(0, ins.readInt()); assertEquals(0, ins.readInt()); assertEquals(ts0, ins.readLong()); assertEquals(2, ins.readInt()); for (int i = 16 + 16; i < 4096; i += 16) { assertEquals(-1, ins.readInt()); assertEquals(0L, ins.readLong()); assertEquals(0, ins.readInt()); } assertEquals((byte)0xA1, ins.readByte()); assertEquals(ts0, ins.readLong()); assertEquals(key0.length(), (int)ins.readShort()); assertEquals(payload0.length, ins.readInt()); assertEquals(key0, readUTF8(ins, key0.length())); assertEquals(new String(payload0, "UTF8"), readUTF8(ins, payload0.length)); assertEquals((byte)0xA1, ins.readByte()); assertEquals(ts1, ins.readLong()); assertEquals(key1.length(), (int)ins.readShort()); assertEquals(payload1.length, ins.readInt()); assertEquals(key1, readUTF8(ins, key1.length())); assertEquals(new String(payload1, "UTF8"), readUTF8(ins, payload1.length)); ins.close(); }