@Test public void testCursor() throws IOException { File bd = mkdir("cursor"); Random rnd = new Random(123); PersistentMessageBuffer b = new PersistentMessageBuffer(bd); b.setFirstId(1000); b.setSegmentLength(8192 + MessageFile.FILE_HEADER_SIZE); MessageCursor c = b.cursor(0); assertFalse(c.next()); Msg m0 = appendFixedSizeMsg(b, 100, 4096, rnd); assertNextMsg(m0, c); assertFalse(c.next()); c.close(); c = b.cursor(0); assertNextMsg(m0, c); assertFalse(c.next()); Msg m1 = appendFixedSizeMsg(b, 200, 4096, rnd); assertNextMsg(m1, c); assertFalse(c.next()); Msg m2 = appendFixedSizeMsg(b, 300, 4096, rnd); Msg m3 = appendFixedSizeMsg(b, 400, 4096, rnd); Msg m4 = appendFixedSizeMsg(b, 500, 4096, rnd); assertNextMsg(m2, c); assertNextMsg(m3, c); assertNextMsg(m4, c); assertFalse(c.next()); c.close(); c = b.cursor(0); MessageCursor c2 = b.cursor(0); assertNextMsg(m0, c); assertNextMsg(m0, c2); c.close(); c2.close(); seekByIdCheck(b, m0); seekByIdCheck(b, m1); seekByIdCheck(b, m2); seekByIdCheck(b, m3); seekByIdCheck(b, m4); seekByTimestampCheck(b, m0); seekByTimestampCheck(b, m1); seekByTimestampCheck(b, m2); seekByTimestampCheck(b, m3); seekByTimestampCheck(b, m4); b.close(); b = new PersistentMessageBuffer(bd); seekByTimestampCheck(b, m0); b.close(); }
@Override public synchronized MessageCursor cursor(long messageId) throws IOException { int i = findFileIndex(messageId); if (i < 0) return new EmptyCursor(); MessageFile mf = getMessageFileForCursor(i); long first = mf.getFirstMessageId(); if (messageId < first) messageId = first; return new Cursor(i, mf, mf.cursor(messageId)); }
PersistentMessageBuffer implements MessageBuffer { @Override public synchronized MessageCursor cursor(long messageId) throws IOException { int i = findFileIndex(messageId); if (i < 0) return new EmptyCursor(); MessageFile mf = getMessageFileForCursor(i); long first = mf.getFirstMessageId(); if (messageId < first) messageId = first; return new Cursor(i, mf, mf.cursor(messageId)); } }
PersistentMessageBuffer implements MessageBuffer { @Override public synchronized MessageCursor cursor(long messageId) throws IOException { int i = findFileIndex(messageId); if (i < 0) return new EmptyCursor(); MessageFile mf = getMessageFileForCursor(i); long first = mf.getFirstMessageId(); if (messageId < first) messageId = first; return new Cursor(i, mf, mf.cursor(messageId)); } PersistentMessageBuffer(File dir); }
PersistentMessageBuffer implements MessageBuffer { @Override public synchronized MessageCursor cursor(long messageId) throws IOException { int i = findFileIndex(messageId); if (i < 0) return new EmptyCursor(); MessageFile mf = getMessageFileForCursor(i); long first = mf.getFirstMessageId(); if (messageId < first) messageId = first; return new Cursor(i, mf, mf.cursor(messageId)); } 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 synchronized MessageCursor cursor(long messageId) throws IOException { int i = findFileIndex(messageId); if (i < 0) return new EmptyCursor(); MessageFile mf = getMessageFileForCursor(i); long first = mf.getFirstMessageId(); if (messageId < first) messageId = first; return new Cursor(i, mf, mf.cursor(messageId)); } 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);  }