PersistentMessageBuffer implements MessageBuffer { public void cleanup() throws IOException { for (;;) { File doomed; synchronized (this) { if (maxSize == 0 || getSize() <= maxSize || firstFile >= lastFile - 1) return; doomed = getFile(firstFile); ++firstFile; } if (!doomed.delete()) { throw new IOException("Unable to delete [" + doomed + "]"); } } } PersistentMessageBuffer(File dir); @Override synchronized void setFirstId(long firstMessageId); @Override void setMaxSize(long bytes); @Override long getMaxSize(); void setSegmentCount(int segmentCount); int getSegmentCount(); @Override void setMaxPayloadSize(int maxPayloadSize); @Override int getMaxPayloadSize(); void setSegmentLength(int segmentLength); int getSegmentLength(); @Override void setExecutor(Executor executor); @Override synchronized boolean isEmpty(); @Override synchronized long getSize(); synchronized int getFileCount(); @Override long append(long timestamp, String routingKey, byte[] payload); @SuppressWarnings({"ConstantConditions", "SynchronizationOnLocalVariableOrMethodParameter"}) @Override long append(long timestamp, String routingKey, ReadableByteChannel payload, int payloadSize); @Override int getMessageSize(String routingKey, int payloadSize); @Override void setAutoSyncInterval(int ms); @Override int getAutoSyncInterval(); @Override void setTimer(Timer timer); void cleanup(); @Override synchronized void sync(); @Override synchronized boolean isOpen(); @Override void close(); @Override synchronized long getNextId(); @Override synchronized long getMessageCount(); @Override synchronized Date getOldestTimestamp(); @Override synchronized long getOldestId(); synchronized Date getMostRecentTimestamp(); @Override long getCreationTime(); @Override synchronized Timeline getTimeline(); @Override synchronized Timeline getTimeline(long messageId); @Override String toString(); @Override synchronized MessageCursor cursor(long messageId); @Override synchronized MessageCursor cursorByTimestamp(long timestamp);  }
@Test public void testCleanup() throws IOException { File bd = mkdir("cleanup"); PersistentMessageBuffer b = new PersistentMessageBuffer(bd); b.setSegmentLength(8192 + MessageFile.FILE_HEADER_SIZE); append(b, 0, "", 8192); append(b, 0, "", 8192); append(b, 0, "", 8192); append(b, 0, "", 8192); expect(bd.list(), "0000000000000000-0000000000000000-1.qdb", "0000000000002000-0000000000000000-1.qdb", "0000000000004000-0000000000000000-1.qdb", "0000000000006000-0000000000000000-0.qdb"); b.setMaxSize((8192 + MessageFile.FILE_HEADER_SIZE) * 2); b.cleanup(); assertEquals(0x4000, b.getOldestId()); expect(bd.list(), "0000000000004000-0000000000000000-1.qdb", "0000000000006000-0000000000000000-0.qdb"); b.setMaxSize(1); b.cleanup(); expect(bd.list(), "0000000000006000-0000000000000000-0.qdb"); b.close(); }