@Test public void testAppend() throws IOException { PersistentMessageBuffer b = new PersistentMessageBuffer(mkdir("append")); assertTrue(b.toString().contains("append")); assertTrue(b.getCreationTime() >= System.currentTimeMillis()); b.setSegmentLength(10000 + MessageFile.FILE_HEADER_SIZE); assertEquals(0, b.getFileCount()); assertEquals(0L, b.getSize()); assertEquals(10000 + MessageFile.FILE_HEADER_SIZE, b.getSegmentLength()); long ts = System.currentTimeMillis(); assertEquals(0L, append(b, ts, "", 5000)); assertEquals(5000L, append(b, ts, "", 5000)); assertEquals(1, b.getFileCount()); assertEquals(10000L + MessageFile.FILE_HEADER_SIZE, b.getSize()); assertEquals(10000L, append(b, ts, "", 5000)); assertEquals(2, b.getFileCount()); assertEquals(15000L + MessageFile.FILE_HEADER_SIZE * 2, b.getSize()); assertEquals(15000L, append(b, ts, "", 5000)); assertEquals(2, b.getFileCount()); assertEquals(20000L + MessageFile.FILE_HEADER_SIZE * 2, b.getSize()); b.close(); }
@Override public long append(long timestamp, String routingKey, byte[] payload) throws IOException { return append(timestamp, routingKey, Channels.newChannel(new ByteArrayInputStream(payload)), payload.length); }
PersistentMessageBuffer implements MessageBuffer { @Override public long append(long timestamp, String routingKey, byte[] payload) throws IOException { return append(timestamp, routingKey, Channels.newChannel(new ByteArrayInputStream(payload)), payload.length); } }
PersistentMessageBuffer implements MessageBuffer { @Override public long append(long timestamp, String routingKey, byte[] payload) throws IOException { return append(timestamp, routingKey, Channels.newChannel(new ByteArrayInputStream(payload)), payload.length); } PersistentMessageBuffer(File dir); }
PersistentMessageBuffer implements MessageBuffer { @Override public long append(long timestamp, String routingKey, byte[] payload) throws IOException { return append(timestamp, routingKey, Channels.newChannel(new ByteArrayInputStream(payload)), payload.length); } 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); }
PersistentMessageBuffer implements MessageBuffer { @Override public long append(long timestamp, String routingKey, byte[] payload) throws IOException { return append(timestamp, routingKey, Channels.newChannel(new ByteArrayInputStream(payload)), payload.length); } 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);  }