@Test public void shouldFlushItemsToStoreOnEviction() throws Exception { final List<KeyValue<Windowed<String>, Long>> added = addSessionsUntilOverflow("a", "b", "c", "d"); assertEquals(added.size() - 1, cache.size()); final KeyValueIterator<Bytes, byte[]> iterator = underlying.fetch(Bytes.wrap(added.get(0).key.key().getBytes()), 0, 0); final KeyValue<Bytes, byte[]> next = iterator.next(); assertEquals(added.get(0).key, SessionKeySerde.from(next.key.get(), Serdes.String().deserializer(), "dummy")); assertArrayEquals(serdes.rawValue(added.get(0).value), next.value); }
@Override public KeyValueIterator<Windowed<K>, AGG> fetch(final K key) { return findSessions(key, 0, Long.MAX_VALUE); }
CachingSessionStore extends WrappedStateStore.AbstractStateStore implements SessionStore<K, AGG>, CachedStateStore<Windowed<K>, AGG> { @Override public KeyValueIterator<Windowed<K>, AGG> fetch(final K key) { return findSessions(key, 0, Long.MAX_VALUE); } }
CachingSessionStore extends WrappedStateStore.AbstractStateStore implements SessionStore<K, AGG>, CachedStateStore<Windowed<K>, AGG> { @Override public KeyValueIterator<Windowed<K>, AGG> fetch(final K key) { return findSessions(key, 0, Long.MAX_VALUE); } 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> { @Override public KeyValueIterator<Windowed<K>, AGG> fetch(final K key) { return findSessions(key, 0, Long.MAX_VALUE); } 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> { @Override public KeyValueIterator<Windowed<K>, AGG> fetch(final K key) { return findSessions(key, 0, Long.MAX_VALUE); } 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);  }