@SuppressWarnings("unchecked") @Test public void shouldSupportRangeAcrossMultipleKVStores() throws Exception { final KeyValueStore<String, String> cache = newStoreInstance(); stubProviderTwo.addStore(storeName, cache); stubOneUnderlying.put("a", "a"); stubOneUnderlying.put("b", "b"); stubOneUnderlying.put("z", "z"); cache.put("c", "c"); cache.put("d", "d"); cache.put("x", "x"); final List<KeyValue<String, String>> results = toList(theStore.range("a", "e")); assertTrue(results.contains(new KeyValue<>("a", "a"))); assertTrue(results.contains(new KeyValue<>("b", "b"))); assertTrue(results.contains(new KeyValue<>("c", "c"))); assertTrue(results.contains(new KeyValue<>("d", "d"))); assertEquals(4, results.size()); }
@Override public KeyValueIterator<K, V> range(final K from, final K to) { final NextIteratorFunction<K, V> nextIteratorFunction = new NextIteratorFunction<K, V>() { @Override public KeyValueIterator<K, V> apply(final ReadOnlyKeyValueStore<K, V> store) { try { return store.range(from, to); } catch (InvalidStateStoreException e) { throw new InvalidStateStoreException("State store is not available anymore and may have been migrated to another instance; please re-discover its location from the state metadata."); } } }; final List<ReadOnlyKeyValueStore<K, V>> stores = storeProvider.stores(storeName, storeType); return new DelegatingPeekingKeyValueIterator<>(storeName, new CompositeKeyValueIterator(stores.iterator(), nextIteratorFunction)); }
CompositeReadOnlyKeyValueStore implements ReadOnlyKeyValueStore<K, V> { @Override public KeyValueIterator<K, V> range(final K from, final K to) { final NextIteratorFunction<K, V> nextIteratorFunction = new NextIteratorFunction<K, V>() { @Override public KeyValueIterator<K, V> apply(final ReadOnlyKeyValueStore<K, V> store) { try { return store.range(from, to); } catch (InvalidStateStoreException e) { throw new InvalidStateStoreException("State store is not available anymore and may have been migrated to another instance; please re-discover its location from the state metadata."); } } }; final List<ReadOnlyKeyValueStore<K, V>> stores = storeProvider.stores(storeName, storeType); return new DelegatingPeekingKeyValueIterator<>(storeName, new CompositeKeyValueIterator(stores.iterator(), nextIteratorFunction)); } }
CompositeReadOnlyKeyValueStore implements ReadOnlyKeyValueStore<K, V> { @Override public KeyValueIterator<K, V> range(final K from, final K to) { final NextIteratorFunction<K, V> nextIteratorFunction = new NextIteratorFunction<K, V>() { @Override public KeyValueIterator<K, V> apply(final ReadOnlyKeyValueStore<K, V> store) { try { return store.range(from, to); } catch (InvalidStateStoreException e) { throw new InvalidStateStoreException("State store is not available anymore and may have been migrated to another instance; please re-discover its location from the state metadata."); } } }; final List<ReadOnlyKeyValueStore<K, V>> stores = storeProvider.stores(storeName, storeType); return new DelegatingPeekingKeyValueIterator<>(storeName, new CompositeKeyValueIterator(stores.iterator(), nextIteratorFunction)); } CompositeReadOnlyKeyValueStore(final StateStoreProvider storeProvider,
                                          final QueryableStoreType<ReadOnlyKeyValueStore<K, V>> storeType,
                                          final String storeName); }
CompositeReadOnlyKeyValueStore implements ReadOnlyKeyValueStore<K, V> { @Override public KeyValueIterator<K, V> range(final K from, final K to) { final NextIteratorFunction<K, V> nextIteratorFunction = new NextIteratorFunction<K, V>() { @Override public KeyValueIterator<K, V> apply(final ReadOnlyKeyValueStore<K, V> store) { try { return store.range(from, to); } catch (InvalidStateStoreException e) { throw new InvalidStateStoreException("State store is not available anymore and may have been migrated to another instance; please re-discover its location from the state metadata."); } } }; final List<ReadOnlyKeyValueStore<K, V>> stores = storeProvider.stores(storeName, storeType); return new DelegatingPeekingKeyValueIterator<>(storeName, new CompositeKeyValueIterator(stores.iterator(), nextIteratorFunction)); } CompositeReadOnlyKeyValueStore(final StateStoreProvider storeProvider,
                                          final QueryableStoreType<ReadOnlyKeyValueStore<K, V>> storeType,
                                          final String storeName); @Override V get(final K key); @Override KeyValueIterator<K, V> range(final K from, final K to); @Override KeyValueIterator<K, V> all(); @Override long approximateNumEntries(); }
CompositeReadOnlyKeyValueStore implements ReadOnlyKeyValueStore<K, V> { @Override public KeyValueIterator<K, V> range(final K from, final K to) { final NextIteratorFunction<K, V> nextIteratorFunction = new NextIteratorFunction<K, V>() { @Override public KeyValueIterator<K, V> apply(final ReadOnlyKeyValueStore<K, V> store) { try { return store.range(from, to); } catch (InvalidStateStoreException e) { throw new InvalidStateStoreException("State store is not available anymore and may have been migrated to another instance; please re-discover its location from the state metadata."); } } }; final List<ReadOnlyKeyValueStore<K, V>> stores = storeProvider.stores(storeName, storeType); return new DelegatingPeekingKeyValueIterator<>(storeName, new CompositeKeyValueIterator(stores.iterator(), nextIteratorFunction)); } CompositeReadOnlyKeyValueStore(final StateStoreProvider storeProvider,
                                          final QueryableStoreType<ReadOnlyKeyValueStore<K, V>> storeType,
                                          final String storeName); @Override V get(final K key); @Override KeyValueIterator<K, V> range(final K from, final K to); @Override KeyValueIterator<K, V> all(); @Override long approximateNumEntries();  }