How fo force nextflow to repeat a process until all values in a particular channels are used up BUT a single value from another channel is needed

Question

I am trying to find a hack for getting a process to run until all emissions in a channel are used up. The problem is that the process also takes as input a second channel that only has a single emission. Therefore, the process only runs once, but I need it to repeatedly use the single emission channel for all the emissions in the other channel. I imagine this is not too uncommon a problem, but I have not found a solution.

To be more clear, here is the actual bioinformatics I am trying to do: I have split fastqs into files for multiple samples and emit these files into a channel. I copy this channel because I want to use the files again later. I then build an assembly from the combined files which I index with bwa. I next pass the indexed assembly (index channel) into a new process that also takes as input ALL of the original fastq files (the ones I put into a duplicate channel earlier). The goal is to align each of the fastq files against the genome. However, there are only a few emissions in the index channel so the process doesn't repeat for all the items in my fastq channel.

To aid you all here is a simple example that may help, but take note that my situation doesn't use value channels so is a bit trickier to deal with.

exampleChannel_multiplethings = Channel.from( [1, 3, 5, 7, 9] )
exampleChannel_onething = Channel.from( [1] )

process foo {
  input:
  val x from exampleChannel_multiplethings
  val y from exampleChannel_onething
  
  output:
  file 'x.txt' into foo_out

  """
  echo $y $x > x.txt
  """
}

This outputs "1 1" into the x.txt file. This is expected behavior, but I need to circumvent it somehow so that the process is repeated for all the items in the exampleChannel_multiplethings channel. Important note, if I was dealing with single values I could probably use one of the combine operators, but in reality I am trying to pass a handful of files for an indexed genome into the process. I haven't found a combine operator that works, but hopefully I have missed something!

Thanks for any assistance

Actual code (overly lengthy to paste here) can be found at: https://github.com/JHarrisonEcoEvo/nextflow_gbs/blob/main/gbs_nextflow

Answer 1

I discovered a solution using a combination of .collect() and each.

If one collects from a path the result is a value channel of paths. Value channels can be used multiple times, whereas queue channels cannot.

ch_collected = Channel.fromPath( "somepath" ).collect()

ch_collected can be used multiple times but if one tries to input it to a process using the usual file..from syntax one will get all the files input to the process at once. Instead, one must use the each option during input. E.g.,

  input:
    each fastq_file from ch_collected

I hope this helps someone.

Answer 2

Use a value channel:

exampleChannel_onething = Channel.value(1)

On existing channels, you can use https://www.nextflow.io/docs/latest/operator.html#first (or similar) to convert to a value channel.