Add rewindable stream as a separate interface

Author: Burak Serdar

crud/src/main/java/com/redhat/lightblue/assoc/CompositeFindImpl.java

@@ -22,6 +22,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.ArrayList;
+import java.util.Arrays;
 
 import java.util.concurrent.Executors;
 
@@ -144,9 +145,7 @@ public void explain(OperationContext ctx,
         initialize(ctx,req);
         ExecutionContext executionContext = new ExecutionContext(ctx,null);
         JsonDoc doc=new JsonDoc(executionPlan.explain(executionContext));
-        ArrayList<DocCtx> l=new ArrayList<>(1);
-        l.add(new DocCtx(doc));
-        ctx.setDocumentStream(new ListDocumentStream<DocCtx>(l));
+        ctx.setDocumentStream(new ListDocumentStream<DocCtx>(Arrays.asList(new DocCtx(doc))));
     }
 
     @Override

crud/src/main/java/com/redhat/lightblue/crud/DocumentStream.java

@@ -38,24 +38,17 @@ public interface DocumentStream<T> extends Iterator<T>{
      */
     void tee(Consumer<T> dest);
 
-    /**
-     * returns true if rewind works
-     */
-    default boolean canRewind() { return false; }
-
-    /**
-     * returns a new stream that starts the same resultset from the beginning. Only works if canRewind() is true
-     */
-    default DocumentStream<T> rewind() { throw new UnsupportedOperationException(); }
-
     public static <S,D> DocumentStream<D> map(final DocumentStream<S> source,final Function<S,D> map) {
-        return new DocumentStreamMapper<S,D>(source,map);
+        if(source instanceof RewindableDocumentStream)
+            return new RewindableDocumentStream.RewindableDocumentStreamMapper<S,D>((RewindableDocumentStream<S>)source,map);
+        else
+            return new DocumentStreamMapper<S,D>(source,map);
     }
 
-    static final class DocumentStreamMapper<S,D> implements DocumentStream<D> {
-        private final ArrayList<Consumer<D>> listeners=new ArrayList<>();
-        private final DocumentStream<S> source;
-        private final Function<S,D> map;
+    static class DocumentStreamMapper<S,D> implements DocumentStream<D> {
+        final ArrayList<Consumer<D>> listeners=new ArrayList<>();
+        final DocumentStream<S> source;
+        final Function<S,D> map;
         DocumentStreamMapper(DocumentStream<S> source,Function<S,D> map) {
             this.source=source;
             this.map=map;
@@ -76,14 +69,6 @@ public D next() {
             return d;
         }
         @Override
-        public boolean canRewind() {
-            return source.canRewind();
-        }
-        @Override
-        public DocumentStream<D> rewind() {
-            return new DocumentStreamMapper<S,D>(source.rewind(),map);
-        }
-        @Override
         public void tee(Consumer<D> t) {
             listeners.add(t);
         }

crud/src/main/java/com/redhat/lightblue/crud/ListDocumentStream.java

@@ -26,7 +26,7 @@
 /**
  * Default trivial implementation of document stream that uses a list to hold its elements
  */
-public class ListDocumentStream<T> implements DocumentStream<T> {
+public class ListDocumentStream<T> implements RewindableDocumentStream<T> {
 
     private final List<T> documents;
     private Iterator<T> itr;
@@ -37,12 +37,7 @@ public ListDocumentStream(List<T> list) {
     }
 
     @Override
-    public boolean canRewind() {
-        return true;
-    }
-
-    @Override
-    public DocumentStream<T> rewind() {
+    public RewindableDocumentStream<T> rewind() {
         return new ListDocumentStream<T>(documents);
     }
 

crud/src/main/java/com/redhat/lightblue/crud/RewindableDocumentStream.java

@@ -0,0 +1,47 @@
+/*
+ Copyright 2013 Red Hat, Inc. and/or its affiliates.
+
+ This file is part of lightblue.
+
+ This program is free software: you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation, either version 3 of the License, or
+ (at your option) any later version.
+
+ This program is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package com.redhat.lightblue.crud;
+
+import java.util.Iterator;
+import java.util.ArrayList;
+import java.util.function.Function;
+import java.util.function.Consumer;
+
+/**
+ * A document stream than can be rewound
+ */
+public interface RewindableDocumentStream<T> extends DocumentStream<T> {
+    
+    /**
+     * returns a new stream that starts the same resultset from the beginning. Only works if canRewind() is true
+     */
+    RewindableDocumentStream<T> rewind();
+
+    static class RewindableDocumentStreamMapper<S,D> extends DocumentStream.DocumentStreamMapper<S,D>
+        implements RewindableDocumentStream<D> {
+        RewindableDocumentStreamMapper(RewindableDocumentStream<S> source,Function<S,D> map) {
+            super(source,map);
+        }
+        @Override
+        public RewindableDocumentStream<D> rewind() {
+            return new RewindableDocumentStreamMapper<S,D>(((RewindableDocumentStream<S>)source).rewind(),map);
+        }
+    }
+
+}

crud/src/main/java/com/redhat/lightblue/hooks/HookManager.java

@@ -33,6 +33,7 @@
 import com.redhat.lightblue.crud.CrudConstants;
 import com.redhat.lightblue.crud.DocCtx;
 import com.redhat.lightblue.crud.DocumentStream;
+import com.redhat.lightblue.crud.RewindableDocumentStream;
 import com.redhat.lightblue.eval.Projector;
 import com.redhat.lightblue.mediator.OperationContext;
 import com.redhat.lightblue.metadata.EntityMetadata;
@@ -272,8 +273,8 @@ private void queueHooks(CRUDOperationContext ctx, boolean mediatorHooks) {
             }
 
             DocumentStream<DocCtx> documents=ctx.getDocumentStream();
-            if(documents.canRewind()) {
-                DocumentStream<DocCtx> stream=documents.rewind();
+            if(documents instanceof RewindableDocumentStream) {
+                RewindableDocumentStream<DocCtx> stream=((RewindableDocumentStream<DocCtx>)documents).rewind();
                 while(stream.hasNext()) {
                     addDocument(hookList,stream.next());
                 }

crud/src/test/java/com/redhat/lightblue/hooks/HookManagerTest.java

@@ -21,6 +21,8 @@
 import java.util.List;
 import java.util.ArrayList;
 import java.util.Map;
+import java.util.Iterator;
+import java.util.function.Consumer;
 
 import javax.management.RuntimeErrorException;
 
@@ -52,6 +54,7 @@
 import com.redhat.lightblue.crud.Factory;
 import com.redhat.lightblue.crud.DocCtx;
 import com.redhat.lightblue.crud.ListDocumentStream;
+import com.redhat.lightblue.crud.DocumentStream;
 
 import com.redhat.lightblue.util.test.AbstractJsonNodeTest;
 import com.redhat.lightblue.util.JsonDoc;
@@ -332,15 +335,48 @@ public void crudFindQueueTest() throws Exception {
     }
 
 
+    private static class NonRewindableDocumentStream<T> implements DocumentStream<T> {
+
+        private final List<T> documents;
+        private Iterator<T> itr;
+        private final ArrayList<Consumer<T>> listeners=new ArrayList<>();
+        
+        public NonRewindableDocumentStream(List<T> list) {
+            this.documents=list;
+        }
+        
+        @Override
+        public boolean hasNext() {
+            if(itr==null)
+                itr=documents.iterator();
+            return itr.hasNext();
+        }
+        
+        @Override
+        public T next() {
+            if(itr==null)
+                itr=documents.iterator();
+            T doc=itr.next();
+            for(Consumer<T> c:listeners)
+                c.accept(doc);
+            return doc;
+        }
+        
+        @Override
+        public void close() {}
+        
+        @Override
+        public void tee(Consumer<T> listener) {
+            listeners.add(listener);
+        }
+    }
+
     @Test
     public void crudFindQueueTest_deferredProcessing() throws Exception {
         HookManager hooks = new HookManager(resolver, nodeFactory);
         TestOperationContext ctx = setupContext(CRUDOperation.FIND);
 
-        ctx.setDocumentStream(new ListDocumentStream<DocCtx>(ctx.getInputDocuments()) {
-                @Override
-                public boolean canRewind() {return false;}
-            });
+        ctx.setDocumentStream(new NonRewindableDocumentStream<DocCtx>(ctx.getInputDocuments()));
 
         //  hook2 should be called
         hooks.queueHooks(ctx);