Merge branch 'master' into cid-store

Signed-off-by: Jorge Ejarque <[email protected]>

Author: jorgee

docs/channel.md

@@ -20,7 +20,7 @@ In Nextflow there are two kinds of channels: *queue channels* and *value channel
 ### Queue channel
 
 A *queue channel* is a non-blocking unidirectional FIFO queue connecting a *producer* process (i.e. outputting a value)
-to a consumer process, or an operators.
+to a consumer process or an operator.
 
 A queue channel can be created by factory methods ({ref}`channel-of`, {ref}`channel-path`, etc), operators ({ref}`operator-map`, {ref}`operator-flatmap`, etc), and processes (see {ref}`Process outputs <process-output>`).
 

modules/nextflow/src/main/groovy/nextflow/ast/GStringToLazyVisitor.groovy

@@ -36,6 +36,8 @@ import nextflow.script.TokenStdoutCall
 import nextflow.script.TokenValCall
 import nextflow.script.TokenValRef
 import nextflow.script.TokenVar
+import nextflow.script.control.GStringToLazyVisitor
+import nextflow.script.control.TaskCmdXformVisitor
 import org.codehaus.groovy.ast.ASTNode
 import org.codehaus.groovy.ast.ClassCodeVisitorSupport
 import org.codehaus.groovy.ast.ClassNode

modules/nextflow/src/main/groovy/nextflow/ast/NextflowDSLImpl.groovy

@@ -23,6 +23,7 @@ import java.lang.annotation.Target
 
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
+import nextflow.script.control.TaskCmdXformVisitor
 import org.codehaus.groovy.ast.ASTNode
 import org.codehaus.groovy.ast.ClassNode
 import org.codehaus.groovy.control.CompilePhase

modules/nextflow/src/main/groovy/nextflow/ast/TaskCmdXform.groovy

@@ -22,6 +22,7 @@ import java.lang.annotation.RetentionPolicy
 import java.lang.annotation.Target
 
 import groovy.transform.CompileStatic
+import nextflow.config.control.ConfigToGroovyVisitor
 import org.codehaus.groovy.ast.ASTNode
 import org.codehaus.groovy.control.CompilePhase
 import org.codehaus.groovy.control.SourceUnit

modules/nextflow/src/main/groovy/nextflow/ast/TaskCmdXformVisitor.groovy

@@ -639,7 +639,7 @@ class TaskProcessor {
         // -- map the inputs to a map and use to delegate closure values interpolation
         final secondPass = [:]
         int count = makeTaskContextStage1(task, secondPass, values)
-        makeTaskContextStage2(task, secondPass, count)
+        final foreignFiles = makeTaskContextStage2(task, secondPass, count)
 
         // verify that `when` guard, when specified, is satisfied
         if( !checkWhenGuard(task) )
@@ -653,6 +653,9 @@ class TaskProcessor {
         if( checkStoredOutput(task) )
             return
 
+        // -- download foreign files
+        session.filePorter.transfer(foreignFiles)
+
         def hash = createTaskHashKey(task)
         checkCachedOrLaunchTask(task, hash, resumable)
     }
@@ -1928,7 +1931,7 @@ class TaskProcessor {
         throw new ProcessUnrecoverableException("Not a valid path value: '$str'")
     }
 
-    protected List<FileHolder> normalizeInputToFiles( Object obj, int count, boolean coerceToPath, FilePorter.Batch batch ) {
+    protected List<FileHolder> normalizeInputToFiles( Object obj, int count, boolean coerceToPath, FilePorter.Batch foreignFiles ) {
 
         Collection allItems = obj instanceof Collection ? obj : [obj]
         def len = allItems.size()
@@ -1939,10 +1942,11 @@ class TaskProcessor {
 
             if( item instanceof Path || coerceToPath ) {
                 def path = normalizeToPath(item)
+
                 if (path instanceof RealPathAware){
                     path = path.toRealPath()
                 }
-                def target = executor.isForeignFile(path) ? batch.addToForeign(path) : path
+                def target = executor.isForeignFile(path) ? foreignFiles.addToForeign(path) : path
                 def holder = new FileHolder(target)
                 files << holder
             }
@@ -2142,20 +2146,20 @@ class TaskProcessor {
         return count
     }
 
-    final protected void makeTaskContextStage2( TaskRun task, Map secondPass, int count ) {
+    final protected FilePorter.Batch makeTaskContextStage2( TaskRun task, Map secondPass, int count ) {
 
         final ctx = task.context
         final allNames = new HashMap<String,Integer>()
 
-        final FilePorter.Batch batch = session.filePorter.newBatch(executor.getStageDir())
+        final FilePorter.Batch foreignFiles = session.filePorter.newBatch(executor.getStageDir())
 
         // -- all file parameters are processed in a second pass
         //    so that we can use resolve the variables that eventually are in the file name
         for( Map.Entry<FileInParam,?> entry : secondPass.entrySet() ) {
             final param = entry.getKey()
             final val = entry.getValue()
             final fileParam = param as FileInParam
-            final normalized = normalizeInputToFiles(val, count, fileParam.isPathQualifier(), batch)
+            final normalized = normalizeInputToFiles(val, count, fileParam.isPathQualifier(), foreignFiles)
             final resolved = expandWildcards( fileParam.getFilePattern(ctx), normalized )
 
             if( !param.isValidArity(resolved.size()) )
@@ -2183,9 +2187,7 @@ class TaskProcessor {
             def message = "Process `$name` input file name collision -- There are multiple input files for each of the following file names: ${conflicts.keySet().join(', ')}"
             throw new ProcessUnrecoverableException(message)
         }
-
-        // -- download foreign files
-        session.filePorter.transfer(batch)
+        return foreignFiles
     }
 
     protected void makeTaskContextStage3( TaskRun task, HashCode hash, Path folder ) {

modules/nextflow/src/main/groovy/nextflow/config/parser/v2/ConfigToGroovyXform.groovy

@@ -36,6 +36,7 @@
 import nextflow.script.control.ModuleResolver;
 import nextflow.script.control.ResolveIncludeVisitor;
 import nextflow.script.control.ScriptResolveVisitor;
+import nextflow.script.control.ScriptToGroovyVisitor;
 import nextflow.script.control.TypeCheckingVisitor;
 import nextflow.script.parser.ScriptParserPluginFactory;
 import org.codehaus.groovy.ast.ASTNode;

modules/nextflow/src/main/groovy/nextflow/processor/TaskProcessor.groovy

@@ -376,7 +376,7 @@ class PluginsFacade implements PluginStateListener {
         }
 
         // add tower plugin when config contains tower options
-        if( (Bolts.navigate(config,'tower.enabled') || env.TOWER_ACCESS_TOKEN ) && !specs.find {it.id == 'nf-tower' } ) {
+        if( (Bolts.navigate(config,'tower.enabled') || Bolts.navigate(config,'fusion.enabled') || env.TOWER_ACCESS_TOKEN ) && !specs.find {it.id == 'nf-tower' } ) {
             specs << defaultPlugins.getPlugin('nf-tower')
         }
         if( (Bolts.navigate(config,'wave.enabled') || Bolts.navigate(config,'fusion.enabled')) && !specs.find {it.id == 'nf-wave' } ) {

modules/nextflow/src/main/groovy/nextflow/script/parser/v2/ScriptCompiler.java

@@ -124,6 +124,13 @@ class PluginsFacadeTest extends Specification {
         then:
         result == [ new PluginSpec('nf-tower', '0.1.0') ]
 
+        // fusion requires both nf-tower and nf-wave
+        when:
+        handler = new PluginsFacade(defaultPlugins: defaults, env: [NXF_PLUGINS_DEFAULT:'true'])
+        result = handler.pluginsRequirement([fusion:[enabled:true]])
+        then:
+        result == [ new PluginSpec('nf-tower', '0.1.0'), new PluginSpec('nf-wave', '0.1.0')  ]
+
         when:
         handler = new PluginsFacade(defaultPlugins: defaults, env: [:])
         result = handler.pluginsRequirement([wave:[enabled:true]])