Start a sub-workflow for each file of a flatten channel

[Subject-13]

Hello there,

I have the following scenario: I have written a workflow that calls a process that emits 10 files. This workflow finally emits these results as a channel (as far as I understand it).

Now I want to start a separate sub-workflow for each file from this channel, whereby the results of the sub-workflows are to be saved in their own folders whose names are based on the file names of the 10 original files.

To do this, I am currently passing a .flatten channel to the sub-workflow call (like: subworkflow(start.out.flatten()) so that each item is passed individually. This also seems to work the way I want it to. But I still need dynamic output folders based on the file names. Can I have each sub-workflow-call create its own folder where the results are stored?

Thanks in advance!

[mribeirodantas]

You should set it per process with the publishDir process directive. There are multiple examples here. You can read more about the publishDir directive in the official docs here. Does that solve your issue?

[Subject-13]

Thanks for replying! I already solved that problem weeks ago by myself, thats why I closed this thread. But I'll publish my solution for the record.

workflow {
   A(some_inputs)
   
   def file_name_tuple = A.out.files
                             .flatten
                             .map { it -> [it, it.getSampleName().toString()] }

   B(file_name_tuple)
}

process A {
   input: 
   something

   output:
   path "*/path/to/files/*.txt", emit: files

   script:
   """
   some script
   """
}

process B {
   publishDir "${params.outdir}/${file_name}/optional_subfolder"

   input:
   tuple file(file), val(file_name)

   output:
   tuple path(*_outfile.txt), val(file_name), emit: something_tuple

   script:
   """
   some script
   """
}

That's how I'm passing the name of the desired subfolder for all future process calls, so every generated file of process A finds the right subfolder.

Edit: Typos