Move v1 process output resolution to TaskOutputResolverV1

Signed-off-by: Ben Sherman <bentshermann@gmail.com>

Author: bentsherman

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

@@ -77,8 +77,6 @@ import nextflow.script.ScriptMeta
 import nextflow.script.ScriptType
 import nextflow.script.bundle.ResourcesBundle
 import nextflow.script.dsl.Types
-import nextflow.script.params.BaseOutParam
-import nextflow.script.params.CmdEvalParam
 import nextflow.script.params.DefaultOutParam
 import nextflow.script.params.EachInParam
 import nextflow.script.params.EnvInParam
@@ -1421,7 +1419,7 @@ class TaskProcessor {
         if( config instanceof ProcessConfigV2 )
             collectOutputsV2( task )
         else if( config instanceof ProcessConfigV1 )
-            collectOutputsV1( task, task.getTargetDir() )
+            collectOutputsV1( task )
     }
 
     @CompileStatic
@@ -1442,138 +1440,19 @@ class TaskProcessor {
         task.canBind = true
     }
 
-    final protected void collectOutputsV1( TaskRun task, Path workDir ) {
+    @CompileStatic
+    protected void collectOutputsV1( TaskRun task ) {
         log.trace "<$name> collecting output: ${task.outputs}"
 
-        for( OutParam param : task.outputs.keySet() ) {
-
-            switch( param ) {
-                case StdOutParam:
-                    collectStdOut(task, (StdOutParam)param, task.@stdout)
-                    break
-
-                case FileOutParam:
-                    collectOutFiles(task, (FileOutParam)param, workDir)
-                    break
-
-                case ValueOutParam:
-                    collectOutValues(task, (ValueOutParam)param, task.context)
-                    break
-
-                case EnvOutParam:
-                    collectOutEnvParam(task, (EnvOutParam)param, workDir)
-                    break
+        final resolver = new TaskOutputResolverV1(task)
 
-                case CmdEvalParam:
-                    collectOutEnvParam(task, (CmdEvalParam)param, workDir)
-                    break
-
-                case DefaultOutParam:
-                    task.setOutput(param, DefaultOutParam.Completion.DONE)
-                    break
-
-                default:
-                    throw new IllegalArgumentException("Illegal output parameter: ${param.class.simpleName}")
-
-            }
-        }
+        for( OutParam param : task.outputs.keySet() )
+            resolver.resolve(param)
 
         // mark ready for output binding
         task.canBind = true
     }
 
-    protected void collectOutEnvParam(TaskRun task, BaseOutParam param, Path workDir) {
-
-        // fetch the output value
-        final outCmds =  param instanceof CmdEvalParam ? task.getOutputEvals() : null
-        final val = collectOutEnvMap(workDir,outCmds).get(param.name)
-        if( val == null && !param.optional )
-            throw new MissingValueException("Missing environment variable: $param.name")
-        // set into the output set
-        task.setOutput(param,val)
-        // trace the result
-        log.trace "Collecting param: ${param.name}; value: ${val}"
-
-    }
-
-    /**
-     * Parse the `.command.env` file which holds the value for `env` and `cmd`
-     * output types
-     *
-     * @param workDir
-     *      The task work directory that contains the `.command.env` file
-     * @param outEvals
-     *      A {@link Map} instance containing key-value pairs
-     * @return
-     */
-    @CompileStatic
-    @Memoized(maxCacheSize = 10_000)
-    protected Map collectOutEnvMap(Path workDir, Map<String,String> outEvals) {
-        return new TaskEnvCollector(workDir, outEvals).collect()
-    }
-
-    /**
-     * Collects the process 'std output'
-     *
-     * @param task The executed process instance
-     * @param param The declared {@link StdOutParam} object
-     * @param stdout The object holding the task produced std out object
-     */
-    protected void collectStdOut( TaskRun task, StdOutParam param, def stdout ) {
-
-        if( stdout == null && task.type == ScriptType.SCRIPTLET ) {
-            throw new IllegalArgumentException("Missing 'stdout' for process > ${safeTaskName(task)}")
-        }
-
-        if( stdout instanceof Path && !stdout.exists() ) {
-            throw new MissingFileException("Missing 'stdout' file: ${stdout.toUriString()} for process > ${safeTaskName(task)}")
-        }
-
-        task.setOutput(param, stdout)
-    }
-
-    protected void collectOutFiles( TaskRun task, FileOutParam param, Path workDir ) {
-
-        // type file parameter can contain a multiple files pattern separating them with a special character
-        final filePatterns = param.getFilePatterns(task.context, task.workDir)
-        final opts = [
-            followLinks: param.followLinks,
-            glob: param.glob,
-            hidden: param.hidden,
-            includeInputs: param.includeInputs,
-            maxDepth: param.maxDepth,
-            optional: param.optional || param.arity?.min == 0,
-            type: param.type,
-        ]
-        final allFiles = collectOutFiles0(task, filePatterns, opts)
-
-        if( !param.isValidArity(allFiles.size()) )
-            throw new IllegalArityException("Incorrect number of output files for process `${safeTaskName(task)}` -- expected ${param.arity}, found ${allFiles.size()}")
-
-        task.setOutput( param, allFiles.size()==1 && param.isSingle() ? allFiles[0] : allFiles )
-
-    }
-
-    protected List<Path> collectOutFiles0(TaskRun task, List<String> filePatterns, Map opts) {
-        return new TaskFileCollector(filePatterns, opts, task).collect()
-    }
-
-    protected void collectOutValues( TaskRun task, ValueOutParam param, Map ctx ) {
-
-        try {
-            // fetch the output value
-            final val = param.resolve(ctx)
-            // set into the output set
-            task.setOutput(param,val)
-            // trace the result
-            log.trace "Collecting param: ${param.name}; value: ${val}"
-        }
-        catch( MissingPropertyException e ) {
-            throw new MissingValueException("Missing value declared as output parameter: ${e.property}")
-        }
-
-    }
-
     @Memoized
     ResourcesBundle getModuleBundle() {
         final script = this.getOwnerScript()

modules/nextflow/src/test/groovy/nextflow/processor/TaskProcessorTest.groovy

@@ -39,7 +39,6 @@ import nextflow.script.ScriptMeta
 import nextflow.script.ScriptType
 import nextflow.script.bundle.ResourcesBundle
 import nextflow.script.params.FileInParam
-import nextflow.script.params.FileOutParam
 import nextflow.util.CacheHelper
 import nextflow.util.MemoryUnit
 import spock.lang.Specification
@@ -538,90 +537,6 @@ class TaskProcessorTest extends Specification {
         'f*'            | ['/a','/b']                               | '3'       | 'Incorrect number of input files for process `foo` -- expected 3, found 2'
     }
 
-    def 'should collect output files' () {
-        given:
-        def executor = Mock(Executor)
-        def session = Mock(Session) {getFilePorter()>>Mock(FilePorter) }
-        def processor = Spy(new TaskProcessor(session:session, executor:executor))
-        and:
-        def context = new TaskContext(holder: new HashMap<String, Object>())
-        def task = new TaskRun(
-                name: 'foo',
-                type: ScriptType.SCRIPTLET,
-                context: context,
-                config: new TaskConfig())
-        and:
-        def workDir = Path.of('/work')
-
-        when:
-        def param = new FileOutParam(new Binding(), [])
-                .setPathQualifier(true)
-                .optional(OPTIONAL)
-                .bind(FILE_NAME) as FileOutParam
-        if( ARITY )
-            param.setArity(ARITY)
-        and:
-        processor.collectOutFiles(task, param, workDir)
-        then:
-        processor.collectOutFiles0(_,_,_) >> RESULTS
-        and:
-        task.getOutputs().get(param) == EXPECTED
-
-        where:
-        FILE_NAME       | RESULTS                                   | OPTIONAL  | ARITY         | EXPECTED
-        'file.txt'      | [Path.of('/work/file.txt')]               | false     | null          | Path.of('/work/file.txt')
-        '*'             | [Path.of('/work/file.txt')]               | false     | null          | Path.of('/work/file.txt')
-        '*'             | [Path.of('/work/A'), Path.of('/work/B')]  | false     | null          | [Path.of('/work/A'), Path.of('/work/B')]
-        '*'             | []                                        | true      | null          | []
-        and:
-        'file.txt'      | [Path.of('/work/file.txt')]               | false     | '1'           | Path.of('/work/file.txt')
-        '*'             | [Path.of('/work/file.txt')]               | false     | '1'           | Path.of('/work/file.txt')
-        '*'             | [Path.of('/work/file.txt')]               | false     | '1..*'        | [Path.of('/work/file.txt')]
-        '*'             | [Path.of('/work/A'), Path.of('/work/B')]  | false     | '2'           | [Path.of('/work/A'), Path.of('/work/B')]
-        '*'             | [Path.of('/work/A'), Path.of('/work/B')]  | false     | '1..*'        | [Path.of('/work/A'), Path.of('/work/B')]
-        '*'             | []                                        | false     | '0..*'        | []
-    }
-
-    @Unroll
-    def 'should report output file arity error' () {
-        given:
-        def executor = Mock(Executor)
-        def session = Mock(Session)
-        def processor = Spy(new TaskProcessor(session:session, executor:executor))
-        and:
-        def context = new TaskContext(holder: new HashMap<String, Object>())
-        def task = new TaskRun(
-                name: 'foo',
-                type: ScriptType.SCRIPTLET,
-                context: context,
-                config: new TaskConfig())
-        and:
-        def workDir = Path.of('/work')
-
-        when:
-        def param = new FileOutParam(new Binding(), [])
-                .setPathQualifier(true)
-                .optional(OPTIONAL)
-                .bind(FILE_NAME) as FileOutParam
-        if( ARITY )
-            param.setArity(ARITY)
-        and:
-        processor.collectOutFiles(task, param, workDir)
-        then:
-        processor.collectOutFiles0(_,_,_) >> RESULTS
-        and:
-        def e = thrown(EXCEPTION)
-        e.message == ERROR
-
-        where:
-        FILE_NAME       | RESULTS                                   | OPTIONAL  | ARITY         | EXCEPTION             | ERROR
-        'file.txt'      | [Path.of('/work/file.txt')]               | false     | '2'           | IllegalArityException | "Incorrect number of output files for process `foo` -- expected 2, found 1"
-        '*'             | [Path.of('/work/file.txt')]               | false     | '2'           | IllegalArityException | "Incorrect number of output files for process `foo` -- expected 2, found 1"
-        '*'             | [Path.of('/work/file.txt')]               | false     | '2..*'        | IllegalArityException | "Incorrect number of output files for process `foo` -- expected 2..*, found 1"
-        '*'             | []                                        | true      | '1..*'        | IllegalArityException | "Incorrect number of output files for process `foo` -- expected 1..*, found 0"
-
-    }
-
     def 'should submit a task' () {
         given:
         def exec = Mock(Executor)