@Test public void shouldFetchExactKeySkippingShorterKeys() throws Exception { final Bytes key = Bytes.wrap(new byte[]{0, 0}); final HasNextCondition hasNextCondition = sessionKeySchema.hasNextCondition(key, key, 0, Long.MAX_VALUE); final List<Integer> results = getValues(hasNextCondition); assertThat(results, equalTo(Arrays.asList(1, 5))); }
@Override public HasNextCondition hasNextCondition(final Bytes binaryKeyFrom, final Bytes binaryKeyTo, final long from, final long to) { return new HasNextCondition() { @Override public boolean hasNext(final KeyValueIterator<Bytes, ?> iterator) { while (iterator.hasNext()) { final Bytes bytes = iterator.peekNextKey(); final Windowed<Bytes> windowedKey = SessionKeySerde.fromBytes(bytes); if (windowedKey.key().compareTo(binaryKeyFrom) >= 0 && windowedKey.key().compareTo(binaryKeyTo) <= 0 && windowedKey.window().end() >= from && windowedKey.window().start() <= to) { return true; } iterator.next(); } return false; } }; }
SessionKeySchema implements SegmentedBytesStore.KeySchema { @Override public HasNextCondition hasNextCondition(final Bytes binaryKeyFrom, final Bytes binaryKeyTo, final long from, final long to) { return new HasNextCondition() { @Override public boolean hasNext(final KeyValueIterator<Bytes, ?> iterator) { while (iterator.hasNext()) { final Bytes bytes = iterator.peekNextKey(); final Windowed<Bytes> windowedKey = SessionKeySerde.fromBytes(bytes); if (windowedKey.key().compareTo(binaryKeyFrom) >= 0 && windowedKey.key().compareTo(binaryKeyTo) <= 0 && windowedKey.window().end() >= from && windowedKey.window().start() <= to) { return true; } iterator.next(); } return false; } }; } }
SessionKeySchema implements SegmentedBytesStore.KeySchema { @Override public HasNextCondition hasNextCondition(final Bytes binaryKeyFrom, final Bytes binaryKeyTo, final long from, final long to) { return new HasNextCondition() { @Override public boolean hasNext(final KeyValueIterator<Bytes, ?> iterator) { while (iterator.hasNext()) { final Bytes bytes = iterator.peekNextKey(); final Windowed<Bytes> windowedKey = SessionKeySerde.fromBytes(bytes); if (windowedKey.key().compareTo(binaryKeyFrom) >= 0 && windowedKey.key().compareTo(binaryKeyTo) <= 0 && windowedKey.window().end() >= from && windowedKey.window().start() <= to) { return true; } iterator.next(); } return false; } }; }  }
SessionKeySchema implements SegmentedBytesStore.KeySchema { @Override public HasNextCondition hasNextCondition(final Bytes binaryKeyFrom, final Bytes binaryKeyTo, final long from, final long to) { return new HasNextCondition() { @Override public boolean hasNext(final KeyValueIterator<Bytes, ?> iterator) { while (iterator.hasNext()) { final Bytes bytes = iterator.peekNextKey(); final Windowed<Bytes> windowedKey = SessionKeySerde.fromBytes(bytes); if (windowedKey.key().compareTo(binaryKeyFrom) >= 0 && windowedKey.key().compareTo(binaryKeyTo) <= 0 && windowedKey.window().end() >= from && windowedKey.window().start() <= to) { return true; } iterator.next(); } return false; } }; }  @Override void init(final String topic); @Override Bytes upperRangeFixedSize(final Bytes key, final long to); @Override Bytes lowerRangeFixedSize(final Bytes key, final long from); @Override Bytes upperRange(Bytes key, long to); @Override Bytes lowerRange(Bytes key, long from); @Override long segmentTimestamp(final Bytes key); @Override HasNextCondition hasNextCondition(final Bytes binaryKeyFrom, final Bytes binaryKeyTo, final long from, final long to); @Override List<Segment> segmentsToSearch(final Segments segments, final long from, final long to); }
SessionKeySchema implements SegmentedBytesStore.KeySchema { @Override public HasNextCondition hasNextCondition(final Bytes binaryKeyFrom, final Bytes binaryKeyTo, final long from, final long to) { return new HasNextCondition() { @Override public boolean hasNext(final KeyValueIterator<Bytes, ?> iterator) { while (iterator.hasNext()) { final Bytes bytes = iterator.peekNextKey(); final Windowed<Bytes> windowedKey = SessionKeySerde.fromBytes(bytes); if (windowedKey.key().compareTo(binaryKeyFrom) >= 0 && windowedKey.key().compareTo(binaryKeyTo) <= 0 && windowedKey.window().end() >= from && windowedKey.window().start() <= to) { return true; } iterator.next(); } return false; } }; }  @Override void init(final String topic); @Override Bytes upperRangeFixedSize(final Bytes key, final long to); @Override Bytes lowerRangeFixedSize(final Bytes key, final long from); @Override Bytes upperRange(Bytes key, long to); @Override Bytes lowerRange(Bytes key, long from); @Override long segmentTimestamp(final Bytes key); @Override HasNextCondition hasNextCondition(final Bytes binaryKeyFrom, final Bytes binaryKeyTo, final long from, final long to); @Override List<Segment> segmentsToSearch(final Segments segments, final long from, final long to);  }