Multiple subworkflow executions

[PeterP-GH]

I would like to execute a subworkflow, that bundles multiple processes, for each element in an input channel, as with a process. However I can only get all elments from that channel as a list into the subworkflow.

The general structure looks something like this:

workflow subw {
take:
input

main:
input.collect().view() // Here is the problem: All files from "channel" are shown. But i would like "subw" to be executed seperately for each element in "channel"

process1(input)
process2(process1.out)
process3(input.filter { criteria })
...
file = processX.out

emit:
file
}

workflow {
channel = Channel.fromPath("path/*", type: "dir") // This can be some number of directory paths, not known before execution
subw(channel)
out = subw.file.collect()
}

My problem is, that I take as "input" all of the elements in "channel", but I would like to execute "subw" with every element seperately.
Is there some workaround for that problem? I thought about moving the subworkflow to a file and executing it with nextflow from a process, but I would like to avoid this if possible.

Thanks for your help!

[Subject-13]

Have you tried out the flatten() operator on your channel?

https://www.nextflow.io/docs/latest/operator.html#flatten

[PeterP-GH]

Yes, I tried:
subw(channel.flatten())
, but it did not work for me. I think it does not work for the input of a subworkflow in general.

[Subject-13]

What would be the difference between passing the elements from the input channel within the subw workflow to the processes individually? Wouldn't you achieve the same?

[PeterP-GH]

The files in each directory get filtered and processed in subw based on file extensions and funneled into new channels used by the processes. When I have multiple directories in input the files from different directories would get mixed up and I would only get a single output channel for all directories instead of one for each.
For example:

process1(input)
process1.out.collect().filter { it.endsWith(".txt") }.set { channel_for_process2 }

would mix the files from every directory.

[Subject-13]

I see. It is difficult to find a solution without precise background information, but I have an idea that might help you.

After filtering, you could group the files according to their path so that files from the same directory are in the same group.

You could therefore expand your channel beforehand and write your files in tuples.

input = input.map { it -> [it.getParent(), it] } would add tuples to your channel elements and extend the file path. In the next step, you could now group the same file paths using groupTuple.

input = input.groupTuple(by: 0).

Now you would have your files sorted by their path. For further steps you could make use of the file path. For example, in a process directive publishDir you could further manipulate the file path and use it to create and select your specific output folders.

I hope it might help you.

[bentsherman]

It is not possible to execute a subworkflow "for ecah value" in a channel. A workflow can only consume the entire channel and delegate to processes and operators.

Although I'm not 100% grokking what you're trying to do, it should be possible using operators, so I would study the operators to see how you can use them. Aaron's suggestion with groupTuple seems like a good start

[PeterP-GH]

Thanks both of you for your answers!
I have managed to implement the whole thing using groupTuple().
Here is the core of my solution:

process unzip {
    input:
    tuple val(name), path(file)
    
    output:
    tuple val(name), path("${file.baseName}")
    
    """
    gunzip -c ${file} > ${file.baseName}
    """
}

workflow subw {
    take:
    input

    main:
    // Here we get all the files with the .txt extension, while keeping the original tuple structure
    // Transpose is needed if the following process can only handle one input per call
    input.map { it -> tuple(it[0], it[1].findAll { (it.name.endsWith("gz")) }) }.filter{ !it[1].isEmpty() }.transpose().set{ zipped }
    unzip(zipped)
    // Make sure, that the second position in the tuple is a list
    unzip.out.map { it -> tuple(it[0], [it[1]].flatten()) }.set { unzip_out }
    // ...
    // Concat two channels
    txt_channel.concat(unzip_out.map { it -> tuple(it[0], it[1].findAll { (it.name.endsWith(".txt")) }) }).filter { !it[1].isEmpty() }.groupTuple().map { it -> tuple(it[0], it[1].flatten()) }.set { txt_all }
    // ...
}

workflow {
    // This can be done for multiple directories, then leading to [["dirName1", ["file1", "file2", ...]], ["dirName2", [...]], ...]
    channel = Channel.of([dirName, files(dirPath + "/*")])
    // Filter directories without files present
    channel.filter { it -> !it[1].isEmpty() }.set { channel }
    subw(channel)
}