DatastoreStore extends StoreBase { @Override public void processExpires() { log.debug("Processing expired sessions"); Query<Key> query = Query.newKeyQueryBuilder().setKind(sessionKind) .setFilter(PropertyFilter.le(SessionMetadata.EXPIRATION_TIME, clock.millis())) .build(); QueryResults<Key> keys = datastore.run(query); Stream<Key> toDelete = Streams.stream(keys) .parallel() .flatMap(key -> Streams.stream(datastore.run(Query.newKeyQueryBuilder() .setKind(sessionKind) .setFilter(PropertyFilter.hasAncestor(newKey(key.getName()))) .build()))); datastore.delete(toDelete.toArray(Key[]::new)); }  @Override int getSize(); @Override String[] keys(); @Override Session load(String id); @Override void remove(String id); @Override void clear(); @Override void save(Session session); @Override void processExpires(); void setNamespace(String namespace); void setSessionKind(String sessionKind); void setTraceRequest(boolean traceRequest);  }
@Test public void testSessionExpiration() throws Exception { when(datastore.run(any(KeyQuery.class))).thenReturn( new IteratorQueryResults<>(Collections.singletonList(key).iterator()), new IteratorQueryResults<>(Arrays.asList(key, attributeKey).iterator()) ); store.processExpires(); verify(datastore).delete(Arrays.asList(key, attributeKey).toArray(new Key[0])); }