Workflow output definition

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

@@ -0,0 +1,58 @@
+/*
+ * 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 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 static nextflow.extension.DataflowHelper.newOperator
+/**
+ * Implements the {@link OperatorImpl#topic} operator
+ *
+ * @author Ben Sherman <[email protected]>
+ */
+@Slf4j
+@CompileStatic
+class IntoTopicOp {
+
+    private DataflowReadChannel source
+
+    private String name
+
+    private List<DataflowWriteChannel> outputs
+
+    IntoTopicOp( DataflowReadChannel source, String name ) {
+        this.source = source
+        this.name = name
+    }
+
+    DataflowWriteChannel apply() {
+        final target = CH.createBy(source)
+        final topicSource = CH.createTopicSource(name)
+        this.outputs = [target, topicSource]
+        newOperator([source], outputs, new ChainWithClosure(new CopyChannelsClosure()))
+        return target
+    }
+
+    List<DataflowWriteChannel> getOutputs() {
+        return outputs
+    }
+
+}

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

@@ -43,6 +43,7 @@ import nextflow.splitter.FastaSplitter
 import nextflow.splitter.FastqSplitter
 import nextflow.splitter.JsonSplitter
 import nextflow.splitter.TextSplitter
+import org.codehaus.groovy.runtime.InvokerHelper
 import org.codehaus.groovy.runtime.callsite.BooleanReturningMethodInvoker
 import org.codehaus.groovy.runtime.typehandling.DefaultTypeTransformation
 /**
@@ -1239,4 +1240,12 @@ class OperatorImpl {
                 .getOutput()
     }
 
+    DataflowWriteChannel topic(DataflowReadChannel source, String name) {
+        if( !NF.topicChannelEnabled ) throw new MissingMethodException('topic', OperatorImpl.class, InvokerHelper.EMPTY_ARGS)
+        final op = new IntoTopicOp(source, name)
+        final target = op.apply()
+        OpCall.current.get().outputs.addAll( op.outputs )
+        return target
+    }
+
 }

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

@@ -40,29 +40,17 @@ class PublishOp {
 
     private Path sourceDir
 
-    private volatile boolean complete
-
     private Session getSession() { Global.session as Session }
 
     PublishOp(DataflowReadChannel source, Map opts) {
         this.source = source
         this.opts = opts ? new LinkedHashMap(opts) : Collections.emptyMap()
-
-        // adapt `to`  option
-        if( this.opts.containsKey('to') ) {
-            this.opts.path = this.opts.to
-            this.opts.remove('to')
-        }
-
         this.publisher = PublishDir.create(this.opts)
     }
 
-    protected boolean getComplete() { complete }
-
     PublishOp apply() {
         final events = new HashMap(2)
         events.onNext = this.&publish0
-        events.onComplete = this.&done0
         DataflowHelper.subscribeImpl(source, events)
         return this
     }
@@ -76,11 +64,6 @@ class PublishOp {
         publisher.apply(result, sourceDir)
     }
 
-    protected void done0(nope) {
-        log.debug "Publish operator complete"
-        this.complete = true
-    }
-
     protected void collectFiles(entry, Collection<Path> result) {
         if( entry instanceof Path ) {
             result.add(entry)
@@ -103,16 +86,18 @@ class PublishOp {
      * @return
      */
     protected Path getTaskDir(Path path) {
-        if( path==null )
+        if( path == null )
             return null
-        def result = getTaskDir0(path, session.workDir)
+        def result = getTaskDir0(path, session.workDir.resolve('tmp'))
+        if( result == null )
+            result = getTaskDir0(path, session.workDir)
         if( result == null )
             result = getTaskDir0(path, session.bucketDir)
         return result
     }
 
     private Path getTaskDir0(Path file, Path base) {
-        if( base==null )
+        if( base == null )
             return null
         if( base.fileSystem != file.fileSystem )
             return null

modules/nextflow/src/main/groovy/nextflow/script/BaseScript.groovy

@@ -113,6 +113,15 @@ abstract class BaseScript extends Script implements ExecutionContext {
         meta.addDefinition(workflow)
     }
 
+    protected output(Closure closure) {
+        if( !entryFlow )
+            throw new IllegalStateException("Publish definition must be defined after the anonymous workflow")
+        if( ExecutionStack.withinWorkflow() )
+            throw new IllegalStateException("Publish definition is not allowed within a workflow")
+
+        entryFlow.publisher = closure
+    }
+
     protected IncludeDef include( IncludeDef include ) {
         if(ExecutionStack.withinWorkflow())
             throw new IllegalStateException("Include statement is not allowed within a workflow definition")

modules/nextflow/src/main/groovy/nextflow/script/WorkflowDef.groovy

@@ -16,14 +16,18 @@
 
 package nextflow.script
 
+import java.nio.file.Path
+
 import groovy.transform.CompileStatic
 import groovy.transform.PackageScope
 import groovy.util.logging.Slf4j
 import groovyx.gpars.dataflow.DataflowWriteChannel
+import nextflow.Channel
 import nextflow.exception.MissingProcessException
 import nextflow.exception.MissingValueException
 import nextflow.exception.ScriptRuntimeException
 import nextflow.extension.CH
+import nextflow.extension.PublishOp
 /**
  * Models a script workflow component
  *
@@ -52,6 +56,8 @@ class WorkflowDef extends BindableDef implements ChainableDef, IterableDef, Exec
 
     private WorkflowBinding binding
 
+    private Closure publisher
+
     WorkflowDef(BaseScript owner, Closure<BodyDef> rawBody, String name=null) {
         this.owner = owner
         this.name = name
@@ -70,6 +76,10 @@ class WorkflowDef extends BindableDef implements ChainableDef, IterableDef, Exec
     /* ONLY FOR TESTING PURPOSE */
     protected WorkflowDef() {}
 
+    void setPublisher(Closure publisher) {
+        this.publisher = publisher
+    }
+
     WorkflowDef clone() {
         final copy = (WorkflowDef)super.clone()
         copy.@body = body.clone()
@@ -204,6 +214,14 @@ class WorkflowDef extends BindableDef implements ChainableDef, IterableDef, Exec
         closure.call()
         // collect the workflow outputs
         output = collectOutputs(declaredOutputs)
+        // publish the workflow outputs
+        if( publisher ) {
+            final dsl = new WorkflowPublishDsl(binding)
+            final cl = (Closure)publisher.clone()
+            cl.setResolveStrategy(Closure.DELEGATE_FIRST)
+            cl.setDelegate(dsl)
+            cl.call()
+        }
         return output
     }
 
@@ -254,3 +272,69 @@ class WorkflowParamsResolver {
         return opts
     }
 }
+
+/**
+ * Implements the DSL for publishing workflow outputs
+ *
+ * @author Ben Sherman <[email protected]>
+ */
+@CompileStatic
+class WorkflowPublishDsl {
+
+    private Binding binding
+
+    WorkflowPublishDsl(Binding binding) {
+        this.binding = binding
+    }
+
+    @Override
+    Object getProperty(String name) {
+        try {
+            return binding.getProperty(name)
+        }
+        catch( MissingPropertyException e ){
+            return super.getProperty(name)
+        }
+    }
+
+    void path(Map opts=[:], String path, Closure closure) {
+        final dsl = new PathDsl(Path.of(path), opts)
+        final cl = (Closure)closure.clone()
+        cl.setResolveStrategy(Closure.DELEGATE_FIRST)
+        cl.setDelegate(dsl)
+        cl.call()
+    }
+
+    class PathDsl {
+
+        private Path path
+        private Map defaults
+
+        PathDsl(Path path, Map defaults) {
+            this.path = path
+            this.defaults = defaults
+        }
+
+        void path(Map opts=[:], String subpath, Closure closure) {
+            final dsl = new PathDsl(path.resolve(subpath), defaults + opts)
+            final cl = (Closure)closure.clone()
+            cl.setResolveStrategy(Closure.DELEGATE_FIRST)
+            cl.setDelegate(dsl)
+            cl.call()
+        }
+
+        void select(Map opts=[:], DataflowWriteChannel source) {
+            new PublishOp(CH.getReadChannel(source), defaults + opts + [path: path]).apply()
+        }
+
+        void select(Map opts=[:], ChannelOut out) {
+            if( out.size() != 1 )
+                throw new IllegalArgumentException("Cannot publish a multi-channel output")
+            select(opts, out[0])
+        }
+
+        void topic(Map opts=[:], String name) {
+            select(opts, Channel.topic(name))
+        }
+    }
+}

tests/publish-dsl.nf

@@ -0,0 +1,82 @@
+#!/usr/bin/env nextflow
+/*
+ * 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.
+ */
+
+
+process align {
+    input:
+    val(x)
+
+    output:
+    path("*.bam")
+    path("${x}.bai")
+
+    """
+    echo ${x} > ${x}.bam
+    echo ${x} | rev > ${x}.bai
+    """
+}
+
+process my_combine {
+    input:
+    path(bamfile)
+    path(baifile)
+
+    output:
+    path 'result.txt'
+
+    """
+    cat $bamfile > result.txt
+    cat $baifile >> result.txt
+    """
+}
+
+process foo {
+  output:
+  file 'xxx'
+
+  '''
+  mkdir xxx
+  touch xxx/A
+  touch xxx/B
+  touch xxx/C
+  '''
+}
+
+workflow {
+  def input = Channel.of('alpha','beta','delta')
+  align(input)
+
+  def bam = align.out[0].toSortedList { it.name }
+  def bai = align.out[1].toSortedList { it.name }
+  my_combine( bam, bai )
+  my_combine.out.view{ it.text }
+
+  foo()
+}
+
+output {
+  path('data') {
+    select align.out[0], mode: 'copy'
+    select align.out[1], mode: 'copy'
+    select my_combine.out
+    select foo.out, mode: 'link'
+  }
+
+  path('more/data') {
+    select my_combine.out, mode: 'copy'
+  }
+}