Add listener support to stream

Author: Burak Serdar

crud/src/main/java/com/redhat/lightblue/assoc/ep/JoinSearch.java

@@ -22,6 +22,7 @@
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.NoSuchElementException;
+import java.util.function.Consumer;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -71,6 +72,7 @@ private class BatchQueryIterator implements DocumentStream<ResultDocument> {
         private final int batchSize;
         private final ExecutionContext ctx;
         private final Iterator<JoinTuple> sourceStream;
+        private final ArrayList<Consumer<ResultDocument>> listeners=new ArrayList<>();
 
         private DocumentStream<DocCtx> currentIterator;
         private boolean done=false; // Are we still iterating, or are we done?
@@ -100,8 +102,12 @@ public ResultDocument next() {
             if(!done) {
                 if(currentIterator==null||!currentIterator.hasNext())
                     retrieveNextBatch();
-                if(currentIterator!=null)
-                    return new ResultDocument(block,currentIterator.next().getOutputDocument());
+                if(currentIterator!=null) {
+                    ResultDocument doc=new ResultDocument(block,currentIterator.next().getOutputDocument());
+                    for(Consumer<ResultDocument> l:listeners)
+                        l.accept(doc);
+                    return doc;
+                }
             }
             throw new NoSuchElementException();
         }
@@ -110,7 +116,12 @@ public ResultDocument next() {
         public void close() {
             if(currentIterator!=null)
                 currentIterator.close();
-        }            
+        }
+
+        @Override
+        public void tee(Consumer<ResultDocument> listener) {
+            listeners.add(listener);
+        }
 
         private void retrieveNextBatch() {
             do {

crud/src/main/java/com/redhat/lightblue/assoc/ep/StepResultDocumentStream.java

@@ -1,6 +1,8 @@
 package com.redhat.lightblue.assoc.ep;
 
 import java.util.Iterator;
+import java.util.ArrayList;
+import java.util.function.Consumer;
 import java.util.stream.Stream;
 
 import com.redhat.lightblue.crud.DocCtx;
@@ -12,6 +14,7 @@
 public class StepResultDocumentStream implements DocumentStream<DocCtx> {
 
     private final Iterator<DocCtx> itr;
+    private final ArrayList<Consumer<DocCtx>> listeners=new ArrayList<>();
 
     public StepResultDocumentStream(StepResult<DocCtx> result) {
         this.itr=result.stream().iterator();
@@ -24,9 +27,17 @@ public boolean hasNext() {
 
     @Override
     public DocCtx next() {
-        return itr.next();
+        DocCtx doc=itr.next();
+        for(Consumer<DocCtx> c:listeners)
+            c.accept(doc);
+        return doc;
     }
 
     @Override
     public void close() {}
+
+    @Override
+    public void tee(Consumer<DocCtx> listener) {
+        listeners.add(listener);
+    }
 }

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

@@ -19,7 +19,9 @@
 package com.redhat.lightblue.crud;
 
 import java.util.Iterator;
+import java.util.ArrayList;
 import java.util.function.Function;
+import java.util.function.Consumer;
 
 /**
  * This interface is used to stream documents. A call to getDocuments
@@ -31,6 +33,11 @@ public interface DocumentStream<T> extends Iterator<T>{
      */
     void close();
 
+    /**
+     * Adds a listener that will be called for each document of the stream when next() is called
+     */
+    void tee(Consumer<T> dest);
+
     /**
      * returns true if rewind works
      */
@@ -42,19 +49,44 @@ public interface DocumentStream<T> extends Iterator<T>{
     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 DocumentStream<D>() {
-            @Override
-            public boolean hasNext() {
-                return source.hasNext();
-            }
-            @Override
-            public void close() {
-                source.close();
-            }
-            @Override
-            public D next() {
-                return map.apply(source.next());
-            }
-        };
+        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;
+        DocumentStreamMapper(DocumentStream<S> source,Function<S,D> map) {
+            this.source=source;
+            this.map=map;
+        }
+        @Override
+        public boolean hasNext() {
+            return source.hasNext();
+        }
+        @Override
+        public void close() {
+            source.close();
+        }
+        @Override
+        public D next() {
+            D d=map.apply(source.next());
+            for(Consumer<D> c:listeners)
+                c.accept(d);
+            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

@@ -20,6 +20,8 @@
 
 import java.util.Iterator;
 import java.util.List;
+import java.util.ArrayList;
+import java.util.function.Consumer;
 
 /**
  * Default trivial implementation of document stream that uses a list to hold its elements
@@ -28,6 +30,7 @@ public class ListDocumentStream<T> implements DocumentStream<T> {
 
     private final List<T> documents;
     private Iterator<T> itr;
+    private final ArrayList<Consumer<T>> listeners=new ArrayList<>();
 
     public ListDocumentStream(List<T> list) {
         this.documents=list;
@@ -58,9 +61,17 @@ public boolean hasNext() {
     public T next() {
         if(itr==null)
             itr=documents.iterator();
-        return itr.next();
+        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);
+    }
 }

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

@@ -309,7 +309,7 @@ private void queueHooks(CRUDOperationContext ctx, boolean mediatorHooks) {
                 // Queue the hooks
                 queuedHooks.addAll(hookCache.values());
             } else {
-                // TODO: Use a closure to tap into the stream
+                // Stream is one-way, and cannot be rewound
             }
         }
     }

metadata/src/main/java/com/redhat/lightblue/hooks/CRUDHook.java

@@ -36,8 +36,7 @@ public interface CRUDHook {
     /**
      * Process the hook
      *
-     * @param ctx The operation context. This contains the operation performed
-     * on the entity, and the complete list of documents
+     * @param md Entity metadata
      * @param cfg The hook configuration as specified in the metadata
      * @param processedDocuments Contains the list of documents that are
      * operated on. If the hook specifies a projection in metadata, the input