@Test public void returnsAStreamPositionedAtTheLastEntry() throws Exception { try (InputStream input = navigator.getStreamAtLastEntry()) { assertThat(navigatorsCodec.decode(input).getSeqNum(), is(equalTo((long) LAST_SEQ_NUM))); } }
@Override public InputStream getStreamAtLastEntry() throws IOException { long lastEntrySeqNum = lastIndexedSeqNum(); long lastEntryAddress = index.get(lastEntrySeqNum); PersistenceReader reader = persistence.getReader(); reader.position(lastEntryAddress); InputStream inputStream = Channels.newInputStream(reader); try { while (true) { long entryStartAddress = reader.position(); lastEntrySeqNum = codec.skipEntryAndReturnSeqNum(inputStream); lastEntryAddress = entryStartAddress; } } catch (EOFException ignore) { } reader.position(lastEntryAddress); addToIndex(lastEntrySeqNum, lastEntryAddress); return inputStream; }
InMemoryPersistenceNavigator implements PersistenceNavigator { @Override public InputStream getStreamAtLastEntry() throws IOException { long lastEntrySeqNum = lastIndexedSeqNum(); long lastEntryAddress = index.get(lastEntrySeqNum); PersistenceReader reader = persistence.getReader(); reader.position(lastEntryAddress); InputStream inputStream = Channels.newInputStream(reader); try { while (true) { long entryStartAddress = reader.position(); lastEntrySeqNum = codec.skipEntryAndReturnSeqNum(inputStream); lastEntryAddress = entryStartAddress; } } catch (EOFException ignore) { } reader.position(lastEntryAddress); addToIndex(lastEntrySeqNum, lastEntryAddress); return inputStream; } }
InMemoryPersistenceNavigator implements PersistenceNavigator { @Override public InputStream getStreamAtLastEntry() throws IOException { long lastEntrySeqNum = lastIndexedSeqNum(); long lastEntryAddress = index.get(lastEntrySeqNum); PersistenceReader reader = persistence.getReader(); reader.position(lastEntryAddress); InputStream inputStream = Channels.newInputStream(reader); try { while (true) { long entryStartAddress = reader.position(); lastEntrySeqNum = codec.skipEntryAndReturnSeqNum(inputStream); lastEntryAddress = entryStartAddress; } } catch (EOFException ignore) { } reader.position(lastEntryAddress); addToIndex(lastEntrySeqNum, lastEntryAddress); return inputStream; } InMemoryPersistenceNavigator(BytePersistence persistence, SequentialEntryCodec<E> codec);  InMemoryPersistenceNavigator(BytePersistence persistence, SequentialEntryCodec<E> codec, long offset); }
InMemoryPersistenceNavigator implements PersistenceNavigator { @Override public InputStream getStreamAtLastEntry() throws IOException { long lastEntrySeqNum = lastIndexedSeqNum(); long lastEntryAddress = index.get(lastEntrySeqNum); PersistenceReader reader = persistence.getReader(); reader.position(lastEntryAddress); InputStream inputStream = Channels.newInputStream(reader); try { while (true) { long entryStartAddress = reader.position(); lastEntrySeqNum = codec.skipEntryAndReturnSeqNum(inputStream); lastEntryAddress = entryStartAddress; } } catch (EOFException ignore) { } reader.position(lastEntryAddress); addToIndex(lastEntrySeqNum, lastEntryAddress); return inputStream; } InMemoryPersistenceNavigator(BytePersistence persistence, SequentialEntryCodec<E> codec);  InMemoryPersistenceNavigator(BytePersistence persistence, SequentialEntryCodec<E> codec, long offset); void setMaxEntrySeek(int numberOfEntries); @Override void notifyLogging(long seqNum, long byteAddress); @Override void addToIndex(long seqNum, long address); @Override void notifyTruncation(long seqNum); @Override long getAddressOfEntry(long seqNum); @Override InputStream getStreamAtSeqNum(long seqNum); @Override InputStream getStreamAtFirstEntry(); @Override InputStream getStreamAtLastEntry(); }
InMemoryPersistenceNavigator implements PersistenceNavigator { @Override public InputStream getStreamAtLastEntry() throws IOException { long lastEntrySeqNum = lastIndexedSeqNum(); long lastEntryAddress = index.get(lastEntrySeqNum); PersistenceReader reader = persistence.getReader(); reader.position(lastEntryAddress); InputStream inputStream = Channels.newInputStream(reader); try { while (true) { long entryStartAddress = reader.position(); lastEntrySeqNum = codec.skipEntryAndReturnSeqNum(inputStream); lastEntryAddress = entryStartAddress; } } catch (EOFException ignore) { } reader.position(lastEntryAddress); addToIndex(lastEntrySeqNum, lastEntryAddress); return inputStream; } InMemoryPersistenceNavigator(BytePersistence persistence, SequentialEntryCodec<E> codec);  InMemoryPersistenceNavigator(BytePersistence persistence, SequentialEntryCodec<E> codec, long offset); void setMaxEntrySeek(int numberOfEntries); @Override void notifyLogging(long seqNum, long byteAddress); @Override void addToIndex(long seqNum, long address); @Override void notifyTruncation(long seqNum); @Override long getAddressOfEntry(long seqNum); @Override InputStream getStreamAtSeqNum(long seqNum); @Override InputStream getStreamAtFirstEntry(); @Override InputStream getStreamAtLastEntry();  }