Add mix operator

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

Author: pditommaso

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

@@ -17,16 +17,16 @@
 
 package nextflow.extension
 
-import static nextflow.extension.DataflowHelper.*
-
 import java.util.concurrent.atomic.AtomicInteger
 
 import groovy.transform.CompileStatic
 import groovyx.gpars.dataflow.DataflowReadChannel
 import groovyx.gpars.dataflow.DataflowWriteChannel
 import groovyx.gpars.dataflow.operator.DataflowProcessor
 import nextflow.Channel
+import nextflow.extension.op.ContextRunPerThread
 import nextflow.extension.op.Op
+import nextflow.extension.op.OpContext
 
 /**
  * Implements Nextflow Mix operator
@@ -39,6 +39,7 @@ class MixOp {
     private DataflowReadChannel source
     private List<DataflowReadChannel> others
     private DataflowWriteChannel target
+    private OpContext context = new ContextRunPerThread()
 
     MixOp(DataflowReadChannel source, DataflowReadChannel other) {
         this.source = source
@@ -71,14 +72,22 @@ class MixOp {
                 onComplete: { DataflowProcessor proc -> if(count.decrementAndGet()==0) { Op.bind(proc, target, Channel.STOP) } }
         ]
 
-        subscribeImpl(source, handlers)
+        subscribe0(source, handlers)
         for( def it : others ) {
-            subscribeImpl(it, handlers)
+            subscribe0(it, handlers)
         }
 
         final allSources = [source]
         allSources.addAll(others)
         return target
     }
 
+    private void subscribe0(final DataflowReadChannel source, final Map<String,Closure> events ) {
+        new SubscribeOp()
+            .withSource(source)
+            .withContext(context)
+            .withEvents(events)
+            .apply()
+    }
+
 }

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

@@ -574,7 +574,6 @@ class ProvTest extends Dsl2Spec {
                 .name == ['p1 (5)']
     }
 
-    @Ignore
     def 'should track provenance with mix operator'() {
         when:
         dsl_eval(globalConfig(), '''
@@ -608,6 +607,9 @@ class ProvTest extends Dsl2Spec {
         then:
         upstreamTasksOf('p3 (1)')
                 .name == ['p1 (1)']
+        and:
+        upstreamTasksOf('p3 (2)')
+            .name == ['p2 (1)']
 
     }