Sorted cache objects (#25)

* Ensure cache is in a deterministic order

Sorts cache by checksum.

* Fixed spec.

* Added streamSliceSince

src/main/java/no/fint/cache/CacheService.java

@@ -102,6 +102,15 @@ public Stream<T> streamSince(String orgId, long sinceTimestamp) {
         return getCache(orgId).orElseThrow(() -> new CacheNotFoundException(orgId)).streamSince(sinceTimestamp).map(CacheObject::getObject);
     }
 
+    public Stream<T> streamSliceSince(String orgId, long sinceTimestamp, int skip, int limit) {
+        return getCache(orgId)
+                .orElseThrow(() -> new CacheNotFoundException(orgId))
+                .streamSince(sinceTimestamp)
+                .skip(skip)
+                .limit(limit)
+                .map(CacheObject::getObject);
+    }
+
     public Optional<T> getOne(String orgId, Predicate<T> idFunction) {
         Cache<T> cache = getCache(orgId).orElseThrow(() -> new CacheNotFoundException(orgId));
         return cache.filter(idFunction).max(Comparator.comparingLong(CacheObject::getLastUpdated)).map(CacheObject::getObject);

src/main/java/no/fint/cache/FintCache.java

@@ -58,7 +58,7 @@ private void updateInternal(Map<String, CacheObject<T>> cacheObjectMap) {
             });
 
             cacheObjectsCopy.addAll(cacheObjectMap.values());
-            cacheObjects = ImmutableList.copyOf(cacheObjectsCopy);
+            cacheObjects = ImmutableList.sortedCopyOf(Comparator.comparing(CacheObject::getChecksum), cacheObjectsCopy);
         }
 
         updateMetaData();
@@ -82,7 +82,7 @@ public void add(List<T> objects) {
     private void addInternal(Map<String, CacheObject<T>> newObjects) {
         List<CacheObject<T>> cacheObjectsCopy = new ArrayList<>(cacheObjects);
         cacheObjectsCopy.addAll(newObjects.values());
-        cacheObjects = ImmutableList.copyOf(cacheObjectsCopy);
+        cacheObjects = ImmutableList.sortedCopyOf(Comparator.comparing(CacheObject::getChecksum), cacheObjectsCopy);
         updateMetaData();
     }
 
@@ -133,7 +133,7 @@ private void updateMetaData() {
             int i = iterator.nextIndex();
             CacheObject<T> it = iterator.next();
             digest.update(it.rawChecksum());
-            IntStream.of(it.getHashCodes()).forEach(key -> newIndex.compute(key, (k,v) -> {
+            IntStream.of(it.getHashCodes()).forEach(key -> newIndex.compute(key, (k, v) -> {
                 if (v == null) {
                     return new SingleIndex(i);
                 }
@@ -171,7 +171,9 @@ public int size() {
     }
 
     @Override
-    public long volume() { return cacheMetaData.getSize(); }
+    public long volume() {
+        return cacheMetaData.getSize();
+    }
 
     @Override
     public Stream<CacheObject<T>> filter(Predicate<T> predicate) {

src/test/groovy/no/fint/cache/FintCacheIntegrationSpec.groovy

@@ -174,14 +174,14 @@ class FintCacheIntegrationSpec extends Specification {
 
     def "On received event"() {
         given:
-        def event = new Event(orgId: 'rogfk.no', data: ['test1', 'test2', 'test3'])
+        def event = new Event(orgId: 'rogfk.no', data: ['test1', 'test2', 'test3', 'test4'])
 
         when:
         testCacheService.onAction(event)
         def values = testCacheService.getAll('rogfk.no')
 
         then:
-        values.size() == 3
+        values.size() == 4
     }
 
     def 'Update cache with hashcodes'() {
@@ -216,6 +216,17 @@ class FintCacheIntegrationSpec extends Specification {
         def result = testCacheService.streamSlice('rogfk.no', 1, 1).collect(Collectors.toList())
 
         then:
-        result == ['test2']
+        result == ['test4']
+    }
+
+    def 'Stream cache slice since'() {
+        given:
+        testCacheService.update('rogfk.no', ['test1', 'test2', 'test3', 'test4'])
+
+        when:
+        def result = testCacheService.streamSliceSince('rogfk.no', System.currentTimeMillis() - 500, 1, 1).collect(Collectors.toList())
+
+        then:
+        result == ['test4']
     }
 }