Add count operator

Signed-off-by: Paolo Di Tommaso <[email protected]>

Author: pditommaso

modules/nextflow/src/main/groovy/nextflow/extension/CollectOp.groovy

@@ -62,7 +62,7 @@ class CollectOp {
             Op.bind(processor, target, msg)
         }
 
-        DataflowHelper.subscribeImpl(source, events)
+        DataflowHelper.subscribeImpl(source, true, events)
         return target
     }
 

modules/nextflow/src/main/groovy/nextflow/extension/OperatorImpl.groovy

@@ -91,6 +91,7 @@ class OperatorImpl {
      * @param closure
      * @return
      */
+    @Deprecated
     DataflowWriteChannel chain(final DataflowReadChannel<?> source, final Closure closure) {
         final target = CH.createBy(source)
         newOperator(source, target, stopErrorListener(source,target), new ChainWithClosure(closure))
@@ -104,6 +105,7 @@ class OperatorImpl {
      * @param closure
      * @return
      */
+    @Deprecated
     DataflowWriteChannel chain(final DataflowReadChannel<?> source, final Map<String, Object> params, final Closure closure) {
         return ChainOp.create()
                 .withSource(source)
@@ -423,8 +425,7 @@ class OperatorImpl {
      * @return
      */
     DataflowWriteChannel count(final DataflowReadChannel source ) {
-        final target = count0(source, null)
-        return target
+        return count0(source, null)
     }
 
     /**
@@ -435,33 +436,25 @@ class OperatorImpl {
      * @return
      */
     DataflowWriteChannel count(final DataflowReadChannel source, final Object criteria ) {
-        final target = count0(source, criteria)
-        return target
+        return count0(source, criteria)
     }
 
     private static DataflowVariable count0(DataflowReadChannel<?> source, Object criteria) {
 
         final target = new DataflowVariable()
         final discriminator = criteria != null ? new BooleanReturningMethodInvoker("isCase") : null
 
-        if( source instanceof DataflowExpression) {
-            source.whenBound { item ->
-                discriminator == null || discriminator.invoke(criteria, item) ? target.bind(1) : target.bind(0)
-            }
-        }
-        else {
-            final action = { current, item ->
-                discriminator == null || discriminator.invoke(criteria, item) ? current+1 : current
-            }
-
-            ReduceOp .create()
-                .withSource(source)
-                .withTarget(target)
-                .withSeed(0)
-                .withAction(action)
-                .apply()
+        final action = { current, item ->
+            discriminator == null || discriminator.invoke(criteria, item) ? current+1 : current
         }
 
+        ReduceOp .create()
+            .withSource(source)
+            .withTarget(target)
+            .withSeed(0)
+            .withAction(action)
+            .apply()
+        
         return target
     }
 

modules/nextflow/src/main/groovy/nextflow/extension/ReduceOp.groovy

@@ -17,17 +17,19 @@
 
 package nextflow.extension
 
+import static nextflow.extension.DataflowHelper.*
+
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import groovyx.gpars.dataflow.DataflowReadChannel
 import groovyx.gpars.dataflow.DataflowVariable
+import groovyx.gpars.dataflow.expression.DataflowExpression
 import groovyx.gpars.dataflow.operator.DataflowEventAdapter
 import groovyx.gpars.dataflow.operator.DataflowProcessor
 import nextflow.Channel
 import nextflow.Global
 import nextflow.Session
 import nextflow.extension.op.Op
-
 /**
  * Implements reduce operator logic
  * 
@@ -88,30 +90,12 @@ class ReduceOp {
             throw new IllegalArgumentException("Missing reduce operator source channel")
         if( target==null )
             target = new DataflowVariable()
-
+        final stopOnFirst = source instanceof DataflowExpression
         // the *accumulator* value
         def accum = this.seed
 
         // intercepts operator events
         final listener = new DataflowEventAdapter() {
-            /*
-             * call the passed closure each time
-             */
-            void afterRun(final DataflowProcessor processor, final List<Object> messages) {
-                final item = Op.unwrap(messages).get(0)
-                final value = accum == null ? item : action.call(accum, item)
-
-                if( value == Channel.VOID ) {
-                    // do nothing
-                }
-                else if( value == Channel.STOP ) {
-                    processor.terminate()
-                }
-                else {
-                    accum = value
-                }
-            }
-
             /*
              * when terminates bind the result value
              */
@@ -129,13 +113,20 @@ class ReduceOp {
             }
         }
 
-        ChainOp.create()
-            .withSource(source)
-            .withTarget(CH.create())
-            .withListener(listener)
+        final parameters = new OpParams()
+            .withInput(source)
             .withAccumulator(true)
-            .withAction({true})
-            .apply()
+            .withListener(listener)
+
+        newOperator(parameters) {
+            final value = accum == null ? it : action.call(accum, it)
+            final proc = getDelegate() as DataflowProcessor
+            if( value!=Channel.VOID && value!=Channel.STOP ) {
+                accum = value
+            }
+            if( stopOnFirst || value==Channel.STOP )
+                proc.terminate()
+        }
 
         return target
     }

modules/nextflow/src/test/groovy/nextflow/prov/ProvTest.groovy

@@ -451,6 +451,84 @@ class ProvTest extends Dsl2Spec {
         def upstream1 = upstreamTasksOf('p2')
         upstream1.size() == 1
         upstream1.first.name == 'p1 (5)'
+    }
+
+    def 'should track provenance with collect operator'() {
+        when:
+        dsl_eval(globalConfig(), '''
+            workflow {
+                channel.of(1,2,3) | p1 | collect | p2 
+            }
+            
+            process p1 { 
+              input: val(x)
+              output: val(y) 
+              exec: 
+                y = x
+            }
+            
+            process p2 {
+              input: val(x)
+              exec: 
+                println x
+            }
+        ''')
 
+        then:
+        def t1 = upstreamTasksOf('p2')
+        t1.name == ['p1 (1)', 'p1 (2)', 'p1 (3)']
     }
+
+    def 'should track provenance with count value operator'() {
+        when:
+        dsl_eval(globalConfig(), '''
+            workflow {
+                channel.value(1) | p1 | count | p2 
+            }
+            
+            process p1 { 
+              input: val(x)
+              output: val(y) 
+              exec: 
+                y = x
+            }
+            
+            process p2 {
+              input: val(x)
+              exec: 
+                println x
+            }
+        ''')
+
+        then:
+        def t1 = upstreamTasksOf('p2')
+        t1.name == ['p1']
+    }
+
+    def 'should track provenance with count many operator'() {
+        when:
+        dsl_eval(globalConfig(), '''
+            workflow {
+                channel.of(1,2,3) | p1 | count | p2 
+            }
+            
+            process p1 { 
+              input: val(x)
+              output: val(y) 
+              exec: 
+                y = x
+            }
+            
+            process p2 {
+              input: val(x)
+              exec: 
+                println x
+            }
+        ''')
+
+        then:
+        def t1 = upstreamTasksOf('p2')
+        t1.name == ['p1 (1)', 'p1 (2)', 'p1 (3)']
+    }
+
 }