Add take and last operator

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

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2013-2024, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package nextflow.extension
+
+
+import groovy.transform.CompileStatic
+import groovyx.gpars.dataflow.DataflowReadChannel
+import groovyx.gpars.dataflow.DataflowVariable
+import groovyx.gpars.dataflow.operator.DataflowProcessor
+import nextflow.extension.op.Op
+
+/**
+ * Implements last operator
+ *
+ * @author Paolo Di Tommaso <[email protected]>
+ */
+@CompileStatic
+class LastOp {
+
+    private DataflowReadChannel source
+    private DataflowVariable target
+
+    LastOp withSource(DataflowReadChannel source) {
+        assert source!=null
+        this.source = source
+        return this
+    }
+
+    LastOp withTarget(DataflowVariable target) {
+        assert target!=null
+        this.target = target
+        return this
+    }
+
+    DataflowVariable apply() {
+        assert source!=null
+        if( target==null )
+            target = new DataflowVariable()
+
+        def last = null
+        final next = { last = it }
+        final done = { DataflowProcessor proc -> Op.bind(proc, target, last) }
+        DataflowHelper.subscribeImpl( source, [onNext:next, onComplete: done] )
+        return target
+    }
+
+}

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

@@ -384,11 +384,7 @@ class OperatorImpl {
      * @return A {@code DataflowVariable} emitting the `last` item in the channel
      */
     DataflowWriteChannel last( final DataflowReadChannel source ) {
-
-        def target = new DataflowVariable()
-        def last = null
-        subscribeImpl( source, [onNext: { last = it }, onComplete: {  target.bind(last) }] )
-        return target
+        new LastOp().withSource(source).apply()
     }
 
     DataflowWriteChannel collect(final DataflowReadChannel source, Closure action=null) {
@@ -399,7 +395,6 @@ class OperatorImpl {
         return new CollectOp(source,action,opts).apply()
     }
 
-
     /**
      * Convert a {@code DataflowQueue} alias *channel* to a Java {@code List}
      *

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

@@ -23,15 +23,15 @@ import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import groovyx.gpars.dataflow.DataflowReadChannel
 import groovyx.gpars.dataflow.DataflowWriteChannel
-import groovyx.gpars.dataflow.operator.ChainWithClosure
-import groovyx.gpars.dataflow.operator.CopyChannelsClosure
 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
 
 /**
+ * Implement "take" operator
  *
  * @author Paolo Di Tommaso <[email protected]>
  */
@@ -73,11 +73,16 @@ class TakeOp {
             }
         }
 
-        newOperator(
-                inputs: [source],
-                outputs: [target],
-                listeners: (length > 0 ? [listener] : []),
-                new ChainWithClosure(new CopyChannelsClosure()))
+        final params = new OpParams()
+            .withInput(source)
+            .withOutput(target)
+        if( length>0 )
+            params.withListener(listener)
+
+        newOperator(params) {
+            final proc = getDelegate() as DataflowProcessor
+            Op.bind(proc, target, it)
+        }
 
         return target
     }

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

@@ -368,7 +368,6 @@ class ProvTest extends Dsl2Spec {
     }
 
     def 'should track provenance with first operator'() {
-
         when:
         dsl_eval(globalConfig(), '''
             workflow {
@@ -394,4 +393,64 @@ class ProvTest extends Dsl2Spec {
         upstream1.size() == 1
         upstream1.first.name == 'p1 (1)'
     }
+
+    def 'should track provenance with take operator'() {
+        when:
+        dsl_eval(globalConfig(), '''
+            workflow {
+                channel.of(1,2,3,4,5) | p1 | take(2) | p2 
+            }
+            
+            process p1 { 
+              input: val(x)
+              output: val(y) 
+              exec: 
+                y = x
+            }
+            
+            process p2 {
+              input: val(x)
+              exec: 
+                println x
+            }
+        ''')
+
+        then:
+        def upstream1 = upstreamTasksOf('p2 (1)')
+        upstream1.size() == 1
+        upstream1.first.name == 'p1 (1)'
+        then:
+        def upstream2 = upstreamTasksOf('p2 (2)')
+        upstream2.size() == 1
+        upstream2.first.name == 'p1 (2)'
+
+    }
+
+    def 'should track provenance with last operator'() {
+        when:
+        dsl_eval(globalConfig(), '''
+            workflow {
+                channel.of(1,2,3,4,5) | p1 | last | p2 
+            }
+            
+            process p1 { 
+              input: val(x)
+              output: val(y) 
+              exec: 
+                y = x
+            }
+            
+            process p2 {
+              input: val(x)
+              exec: 
+                println x
+            }
+        ''')
+
+        then:
+        def upstream1 = upstreamTasksOf('p2')
+        upstream1.size() == 1
+        upstream1.first.name == 'p1 (5)'
+
+    }
 }