Input from multiple channel, merge and use in nextflow

Question

I need to merge multiple channel to reuse them in two processes. So far I tried as follows:

channel.from([15,20,21]).into {k1;k2}
channel.from(['a','b','c']).into {m1;m2}
channel.from([1,2,3]).into {l1;l2}
para1=m1.merge(k1).merge(l1)
para2=m2.merge(k2).merge(l2)
process p1{
input:
each m, k, l from para1
"""
echo $m, $k, $l
"""
}
process p2{
input:
each m, k, l from para2
"""
echo $m, $k, $l
"""
}

The error I am getting:

Unknown process directive: '_in_each'

Did you mean of these?
    each

If I use para1.view() I can see the merged elements as expected. Any help

Answer 1

I don't think you can unpack tuples using the each qualifier like that. You can, however, unpack them in the script block:

process test {
    
    input:
    each triple from para1
    
    script: 
    def (m, k, l) = triple
    
    """
    echo "${m} ${k} ${l}"
    """
}

Consider also that the merge operator is deprecated and will be removed from future Nextflow releases. Instead, you may need to join by index, for example:

def indexedChannel( items ) {
    return Channel.from( items.withIndex() ).map { item, idx -> tuple( idx, item ) }
}

k_ch = indexedChannel( [ 15, 20, 21 ] )
m_ch = indexedChannel( [ 'a', 'b', 'c' ] )
l_ch = indexedChannel( [ 1, 2, 3 ] )

m_ch
    .join( k_ch )
    .join( l_ch )
    .map { idx, m, k, l -> tuple( m, k, l ) }
    .into { para1; para2 }