@Test public void shouldIterateOverRange() throws Exception { int items = addItemsToCache(); final KeyValueIterator<String, String> range = store.range(String.valueOf(0), String.valueOf(items)); final List<String> results = new ArrayList<>(); while (range.hasNext()) { results.add(range.next().key); } assertEquals(items, results.size()); }
@Override public KeyValueIterator<K, V> range(final K from, final K to) { validateStoreOpen(); final Bytes origFrom = Bytes.wrap(serdes.rawKey(from)); final Bytes origTo = Bytes.wrap(serdes.rawKey(to)); final KeyValueIterator<Bytes, byte[]> storeIterator = underlying.range(origFrom, origTo); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(cacheName, origFrom, origTo); return new MergedSortedCacheKeyValueStoreIterator<>(cacheIterator, storeIterator, serdes); }
CachingKeyValueStore extends WrappedStateStore.AbstractStateStore implements KeyValueStore<K, V>, CachedStateStore<K, V> { @Override public KeyValueIterator<K, V> range(final K from, final K to) { validateStoreOpen(); final Bytes origFrom = Bytes.wrap(serdes.rawKey(from)); final Bytes origTo = Bytes.wrap(serdes.rawKey(to)); final KeyValueIterator<Bytes, byte[]> storeIterator = underlying.range(origFrom, origTo); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(cacheName, origFrom, origTo); return new MergedSortedCacheKeyValueStoreIterator<>(cacheIterator, storeIterator, serdes); } }
CachingKeyValueStore extends WrappedStateStore.AbstractStateStore implements KeyValueStore<K, V>, CachedStateStore<K, V> { @Override public KeyValueIterator<K, V> range(final K from, final K to) { validateStoreOpen(); final Bytes origFrom = Bytes.wrap(serdes.rawKey(from)); final Bytes origTo = Bytes.wrap(serdes.rawKey(to)); final KeyValueIterator<Bytes, byte[]> storeIterator = underlying.range(origFrom, origTo); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(cacheName, origFrom, origTo); return new MergedSortedCacheKeyValueStoreIterator<>(cacheIterator, storeIterator, serdes); } CachingKeyValueStore(final KeyValueStore<Bytes, byte[]> underlying,
                         final Serde<K> keySerde,
                         final Serde<V> valueSerde); }
CachingKeyValueStore extends WrappedStateStore.AbstractStateStore implements KeyValueStore<K, V>, CachedStateStore<K, V> { @Override public KeyValueIterator<K, V> range(final K from, final K to) { validateStoreOpen(); final Bytes origFrom = Bytes.wrap(serdes.rawKey(from)); final Bytes origTo = Bytes.wrap(serdes.rawKey(to)); final KeyValueIterator<Bytes, byte[]> storeIterator = underlying.range(origFrom, origTo); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(cacheName, origFrom, origTo); return new MergedSortedCacheKeyValueStoreIterator<>(cacheIterator, storeIterator, serdes); } CachingKeyValueStore(final KeyValueStore<Bytes, byte[]> underlying,
                         final Serde<K> keySerde,
                         final Serde<V> valueSerde); @SuppressWarnings("unchecked") @Override void init(final ProcessorContext context, final StateStore root); void setFlushListener(final CacheFlushListener<K, V> flushListener); @Override synchronized void flush(); @Override void close(); @Override boolean persistent(); @Override boolean isOpen(); @Override synchronized V get(final K key); @Override KeyValueIterator<K, V> range(final K from, final K to); @Override KeyValueIterator<K, V> all(); @Override synchronized long approximateNumEntries(); @Override synchronized void put(final K key, final V value); @Override synchronized V putIfAbsent(final K key, final V value); @Override synchronized void putAll(final List<KeyValue<K, V>> entries); @Override synchronized V delete(final K key); @Override StateStore inner(); }
CachingKeyValueStore extends WrappedStateStore.AbstractStateStore implements KeyValueStore<K, V>, CachedStateStore<K, V> { @Override public KeyValueIterator<K, V> range(final K from, final K to) { validateStoreOpen(); final Bytes origFrom = Bytes.wrap(serdes.rawKey(from)); final Bytes origTo = Bytes.wrap(serdes.rawKey(to)); final KeyValueIterator<Bytes, byte[]> storeIterator = underlying.range(origFrom, origTo); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(cacheName, origFrom, origTo); return new MergedSortedCacheKeyValueStoreIterator<>(cacheIterator, storeIterator, serdes); } CachingKeyValueStore(final KeyValueStore<Bytes, byte[]> underlying,
                         final Serde<K> keySerde,
                         final Serde<V> valueSerde); @SuppressWarnings("unchecked") @Override void init(final ProcessorContext context, final StateStore root); void setFlushListener(final CacheFlushListener<K, V> flushListener); @Override synchronized void flush(); @Override void close(); @Override boolean persistent(); @Override boolean isOpen(); @Override synchronized V get(final K key); @Override KeyValueIterator<K, V> range(final K from, final K to); @Override KeyValueIterator<K, V> all(); @Override synchronized long approximateNumEntries(); @Override synchronized void put(final K key, final V value); @Override synchronized V putIfAbsent(final K key, final V value); @Override synchronized void putAll(final List<KeyValue<K, V>> entries); @Override synchronized V delete(final K key); @Override StateStore inner();  }