NextFlow：如何在 DSL2 中使用 inputStream

问题描述

使用 NextFlow (DSL=2)，我想使用文件的每一行作为我的工作流的流输入。

nextflow.enable.dsl=2

process bar {
    input: val data
    output: val result
    exec:
    result = data.toupperCase()
}

workflow {
    myFile = file('input_test.txt')
    myReader = myFile.newReader()
    myFile.withInputStream {
        String line
        while( line = myReader.readLine() ) {
            channel.from(line) | bar | view
        }
    }
}

我面临的问题是我只能使用“bar”进程一次： Process 'bar' has been already used -- If you need to reuse the same component include it with a different name or include in a different workflow context

我还尝试创建一个从线路和呼叫栏中获取通道的子工作流。

有没有办法使用 Nextflow 将流数据用作输入？

注意：我的最终目标不仅仅是应用大写函数。我想在数据流上链接几个复杂的过程。

谢谢！

解决方法

您的示例代码看起来像一个反模式 - 它会尝试为输入文件中的每一行创建一个新通道。相反，请查看 splitting operators，尤其是 splitText operator：

workflow {
    Channel.fromPath('input_test.txt')
        | splitText { it.trim() } \
        | bar \
        | view()
}

如果以上没有帮助，请准确描述您想对输入文件中的每一行做什么。