@Test public void shouldQueryItemsInCacheAndStore() throws Exception { final List<KeyValue<Windowed<String>, Long>> added = addSessionsUntilOverflow("a"); final KeyValueIterator<Windowed<String>, Long> iterator = cachingStore.findSessions("a", 0, added.size() * 10); final List<KeyValue<Windowed<String>, Long>> actual = toList(iterator); assertEquals(added, actual); }
public KeyValueIterator<Windowed<K>, AGG> findSessions(final K key, final long earliestSessionEndTime, final long latestSessionStartTime) { validateStoreOpen(); final Bytes binarySessionId = Bytes.wrap(serdes.rawKey(key)); final Bytes cacheKeyFrom = cacheFunction.cacheKey(keySchema.lowerRangeFixedSize(binarySessionId, earliestSessionEndTime)); final Bytes cacheKeyTo = cacheFunction.cacheKey(keySchema.upperRangeFixedSize(binarySessionId, latestSessionStartTime)); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(cacheName, cacheKeyFrom, cacheKeyTo); final KeyValueIterator<Windowed<Bytes>, byte[]> storeIterator = bytesStore.findSessions( binarySessionId, earliestSessionEndTime, latestSessionStartTime ); final HasNextCondition hasNextCondition = keySchema.hasNextCondition(binarySessionId, binarySessionId, earliestSessionEndTime, latestSessionStartTime); final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator(cacheIterator, hasNextCondition, cacheFunction); return new MergedSortedCacheSessionStoreIterator<>(filteredCacheIterator, storeIterator, serdes, cacheFunction); }
CachingSessionStore extends WrappedStateStore.AbstractStateStore implements SessionStore<K, AGG>, CachedStateStore<Windowed<K>, AGG> { public KeyValueIterator<Windowed<K>, AGG> findSessions(final K key, final long earliestSessionEndTime, final long latestSessionStartTime) { validateStoreOpen(); final Bytes binarySessionId = Bytes.wrap(serdes.rawKey(key)); final Bytes cacheKeyFrom = cacheFunction.cacheKey(keySchema.lowerRangeFixedSize(binarySessionId, earliestSessionEndTime)); final Bytes cacheKeyTo = cacheFunction.cacheKey(keySchema.upperRangeFixedSize(binarySessionId, latestSessionStartTime)); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(cacheName, cacheKeyFrom, cacheKeyTo); final KeyValueIterator<Windowed<Bytes>, byte[]> storeIterator = bytesStore.findSessions( binarySessionId, earliestSessionEndTime, latestSessionStartTime ); final HasNextCondition hasNextCondition = keySchema.hasNextCondition(binarySessionId, binarySessionId, earliestSessionEndTime, latestSessionStartTime); final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator(cacheIterator, hasNextCondition, cacheFunction); return new MergedSortedCacheSessionStoreIterator<>(filteredCacheIterator, storeIterator, serdes, cacheFunction); } }
CachingSessionStore extends WrappedStateStore.AbstractStateStore implements SessionStore<K, AGG>, CachedStateStore<Windowed<K>, AGG> { public KeyValueIterator<Windowed<K>, AGG> findSessions(final K key, final long earliestSessionEndTime, final long latestSessionStartTime) { validateStoreOpen(); final Bytes binarySessionId = Bytes.wrap(serdes.rawKey(key)); final Bytes cacheKeyFrom = cacheFunction.cacheKey(keySchema.lowerRangeFixedSize(binarySessionId, earliestSessionEndTime)); final Bytes cacheKeyTo = cacheFunction.cacheKey(keySchema.upperRangeFixedSize(binarySessionId, latestSessionStartTime)); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(cacheName, cacheKeyFrom, cacheKeyTo); final KeyValueIterator<Windowed<Bytes>, byte[]> storeIterator = bytesStore.findSessions( binarySessionId, earliestSessionEndTime, latestSessionStartTime ); final HasNextCondition hasNextCondition = keySchema.hasNextCondition(binarySessionId, binarySessionId, earliestSessionEndTime, latestSessionStartTime); final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator(cacheIterator, hasNextCondition, cacheFunction); return new MergedSortedCacheSessionStoreIterator<>(filteredCacheIterator, storeIterator, serdes, cacheFunction); } CachingSessionStore(final SessionStore<Bytes, byte[]> bytesStore,
                        final Serde<K> keySerde,
                        final Serde<AGG> aggSerde,
                        final long segmentInterval); }
CachingSessionStore extends WrappedStateStore.AbstractStateStore implements SessionStore<K, AGG>, CachedStateStore<Windowed<K>, AGG> { public KeyValueIterator<Windowed<K>, AGG> findSessions(final K key, final long earliestSessionEndTime, final long latestSessionStartTime) { validateStoreOpen(); final Bytes binarySessionId = Bytes.wrap(serdes.rawKey(key)); final Bytes cacheKeyFrom = cacheFunction.cacheKey(keySchema.lowerRangeFixedSize(binarySessionId, earliestSessionEndTime)); final Bytes cacheKeyTo = cacheFunction.cacheKey(keySchema.upperRangeFixedSize(binarySessionId, latestSessionStartTime)); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(cacheName, cacheKeyFrom, cacheKeyTo); final KeyValueIterator<Windowed<Bytes>, byte[]> storeIterator = bytesStore.findSessions( binarySessionId, earliestSessionEndTime, latestSessionStartTime ); final HasNextCondition hasNextCondition = keySchema.hasNextCondition(binarySessionId, binarySessionId, earliestSessionEndTime, latestSessionStartTime); final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator(cacheIterator, hasNextCondition, cacheFunction); return new MergedSortedCacheSessionStoreIterator<>(filteredCacheIterator, storeIterator, serdes, cacheFunction); } CachingSessionStore(final SessionStore<Bytes, byte[]> bytesStore,
                        final Serde<K> keySerde,
                        final Serde<AGG> aggSerde,
                        final long segmentInterval); void init(final ProcessorContext context, final StateStore root); KeyValueIterator<Windowed<K>, AGG> findSessions(final K key,
                                                           final long earliestSessionEndTime,
                                                           final long latestSessionStartTime); @Override KeyValueIterator<Windowed<K>, AGG> findSessions(K keyFrom,
                                                           K keyTo,
                                                           long earliestSessionEndTime,
                                                           long latestSessionStartTime); @Override void remove(final Windowed<K> sessionKey); @Override void put(final Windowed<K> key, AGG value); @Override KeyValueIterator<Windowed<K>, AGG> fetch(final K key); @Override KeyValueIterator<Windowed<K>, AGG> fetch(K from, K to); void flush(); void close(); void setFlushListener(CacheFlushListener<Windowed<K>, AGG> flushListener); }
CachingSessionStore extends WrappedStateStore.AbstractStateStore implements SessionStore<K, AGG>, CachedStateStore<Windowed<K>, AGG> { public KeyValueIterator<Windowed<K>, AGG> findSessions(final K key, final long earliestSessionEndTime, final long latestSessionStartTime) { validateStoreOpen(); final Bytes binarySessionId = Bytes.wrap(serdes.rawKey(key)); final Bytes cacheKeyFrom = cacheFunction.cacheKey(keySchema.lowerRangeFixedSize(binarySessionId, earliestSessionEndTime)); final Bytes cacheKeyTo = cacheFunction.cacheKey(keySchema.upperRangeFixedSize(binarySessionId, latestSessionStartTime)); final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(cacheName, cacheKeyFrom, cacheKeyTo); final KeyValueIterator<Windowed<Bytes>, byte[]> storeIterator = bytesStore.findSessions( binarySessionId, earliestSessionEndTime, latestSessionStartTime ); final HasNextCondition hasNextCondition = keySchema.hasNextCondition(binarySessionId, binarySessionId, earliestSessionEndTime, latestSessionStartTime); final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator(cacheIterator, hasNextCondition, cacheFunction); return new MergedSortedCacheSessionStoreIterator<>(filteredCacheIterator, storeIterator, serdes, cacheFunction); } CachingSessionStore(final SessionStore<Bytes, byte[]> bytesStore,
                        final Serde<K> keySerde,
                        final Serde<AGG> aggSerde,
                        final long segmentInterval); void init(final ProcessorContext context, final StateStore root); KeyValueIterator<Windowed<K>, AGG> findSessions(final K key,
                                                           final long earliestSessionEndTime,
                                                           final long latestSessionStartTime); @Override KeyValueIterator<Windowed<K>, AGG> findSessions(K keyFrom,
                                                           K keyTo,
                                                           long earliestSessionEndTime,
                                                           long latestSessionStartTime); @Override void remove(final Windowed<K> sessionKey); @Override void put(final Windowed<K> key, AGG value); @Override KeyValueIterator<Windowed<K>, AGG> fetch(final K key); @Override KeyValueIterator<Windowed<K>, AGG> fetch(K from, K to); void flush(); void close(); void setFlushListener(CacheFlushListener<Windowed<K>, AGG> flushListener);  }