@Test public void testCheckpoint() throws IOException { File file = new File(dir, "checkpoint.qdb"); file.delete(); MessageFile mf = new MessageFile(file, 0, 1000000); append(mf, System.currentTimeMillis(), "", "oink".getBytes("UTF8")); mf.checkpoint(false); mf.close(); DataInputStream ins = new DataInputStream(new FileInputStream(file)); int expectedLength = (int) file.length(); ins.skip(2 + 2 + 4); assertEquals(expectedLength, ins.readInt()); ins.close(); FileOutputStream out = new FileOutputStream(file, true); out.write("junk".getBytes("UTF8")); out.close(); assertEquals(expectedLength + 4, file.length()); new MessageFile(file, 0, 1000000).close(); assertEquals(expectedLength, file.length()); }
public void checkpoint(boolean force) throws IOException { synchronized (channel) { channel.force(true); if (length != lastCheckpointLength) { fileHeader.putInt(8, length); if (bucketIndex >= 0) putBucketDataInFileHeader(); fileHeader.position(0); channel.position(0).write(fileHeader); lastCheckpointLength = length; if (force) channel.force(true); } } }
MessageFile implements Closeable { public void checkpoint(boolean force) throws IOException { synchronized (channel) { channel.force(true); if (length != lastCheckpointLength) { fileHeader.putInt(8, length); if (bucketIndex >= 0) putBucketDataInFileHeader(); fileHeader.position(0); channel.position(0).write(fileHeader); lastCheckpointLength = length; if (force) channel.force(true); } } } }
MessageFile implements Closeable { public void checkpoint(boolean force) throws IOException { synchronized (channel) { channel.force(true); if (length != lastCheckpointLength) { fileHeader.putInt(8, length); if (bucketIndex >= 0) putBucketDataInFileHeader(); fileHeader.position(0); channel.position(0).write(fileHeader); lastCheckpointLength = length; if (force) channel.force(true); } } } @SuppressWarnings("StatementWithEmptyBody")  MessageFile(File file, long firstMessageId); @SuppressWarnings("StatementWithEmptyBody")  MessageFile(File file, long firstMessageId, int maxFileSize); }
MessageFile implements Closeable { public void checkpoint(boolean force) throws IOException { synchronized (channel) { channel.force(true); if (length != lastCheckpointLength) { fileHeader.putInt(8, length); if (bucketIndex >= 0) putBucketDataInFileHeader(); fileHeader.position(0); channel.position(0).write(fileHeader); lastCheckpointLength = length; if (force) channel.force(true); } } } @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); }
MessageFile implements Closeable { public void checkpoint(boolean force) throws IOException { synchronized (channel) { channel.force(true); if (length != lastCheckpointLength) { fileHeader.putInt(8, length); if (bucketIndex >= 0) putBucketDataInFileHeader(); fileHeader.position(0); channel.position(0).write(fileHeader); lastCheckpointLength = length; if (force) channel.force(true); } } } @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; }