Nextflow：如何将元组列表转换为元组通道？

我有一个Process返回元组列表，想要创建一个能逐个输出元组的Channel，这样就能把元组里的元素作为multiply Process的输入，让每个元组对应的multiply调用并行执行。但直接用my_process.out.my_returned_list只能得到1个元素[[1, 2], [3, 4], [5, 6]]；用my_process.out.my_returned_list.flatten()又会把元组也展开，得到6个元素1, 2, 3, 4, 5, 6。请问怎么定义一个包含3个元素[1, 2]、[3, 4]、[5, 6]的Channel？

原代码如下：

nextflow.enable.dsl=2

process my_process {
  output:
    val myList, emit: my_returned_list

  exec:
    myList = []
    myList.add(tuple(1, 2))
    myList.add(tuple(3, 4))
    myList.add(tuple(5, 6))
}

process multiply {
    input:
        val left
        val right

    output:
        val left_times_right

    exec:
        left_times_rigth = left * right
}

workflow {
    my_process()
    my_process.out.my_returned_list.view()
    my_process.out.my_returned_list.flatten().view()

    // <channel_to_be_defined>.map { i, j ->
    //    multiply(i, j)
    // }
}

使用flatMap操作符即可实现需求——它会遍历列表中的每个元组，将每个元组作为独立元素发送到新Channel中，同时不会展开元组内部的元素。

修改后的完整代码如下：

nextflow.enable.dsl=2

process my_process {
  output:
    val myList, emit: my_returned_list

  exec:
    myList = []
    myList.add(tuple(1, 2))
    myList.add(tuple(3, 4))
    myList.add(tuple(5, 6))
}

process multiply {
    input:
        val left
        val right

    output:
        val left_times_right

    exec:
        left_times_right = left * right  // 修正原代码中的拼写错误：left_times_rigth → left_times_right
}

workflow {
    def my_result = my_process()
    
    // 用flatMap拆分列表，生成包含单个元组的Channel
    def tuple_channel = my_result.out.my_returned_list.flatMap { list -> list }
    
    tuple_channel.view()  // 输出结果：[1,2], [3,4], [5,6]
    
    // 将元组元素传入multiply实现并行调用
    tuple_channel.map { i, j -> multiply(i, j) }
}

说明：

• flatMap接收原始列表[[1,2], [3,4], [5,6]]后，会逐个取出内部的元组并发送到新Channel，最终得到包含3个元组元素的目标Channel。
• 注意修正了原代码中multiply进程里的变量拼写错误，避免运行时报错。

内容的提问来源于stack exchange，提问作者ebosi