I’m building a ChIP-Seq pipeline in Nextflow, and I’m running into this persistent error when trying to connect two processes that utilize HOMER's makeTagDirectory and findPeaks utilities.

The error that I'm getting is as follows:

ERROR ~ Invalid method invocation `call` with arguments: [rep1, IP_rep1_subset, GRCh38, /path/IP_rep1_subset_tags, INPUT_rep1_subset, GRCh38, /path/INPUT_rep1_subset_tags] (java.util.ArrayList) on _closure15 type
 -- Check '.nextflow.log' file for details

It seems to indicate that I’m passing a list instead of a channel to the process, but I can’t figure out why. The channel chain looks correct.

My preceding process looks like this, and it took aligned BAM files from Bowtie2:

process TAGDIR {

    input:
    tuple val(sample), val(name), path(bam)

    output:
    tuple val(sample), val(name), path("${sample}_tags"), emit: tags

    script:
    """
    mkdir -p ${sample}_tags
    makeTagDirectory ${sample}_tags ${bam}
    """
}

The process that I'm trying to feed into looks like this:

process FINDPEAKS {

    input:
    tuple(
        val(rep),
        val(ip_sample), val(name), path(ip_tags),
        val(control_sample), val(name2), path(control_tags)
    )

    output:
    tuple val(rep), val(ip_sample), path("${rep}_peaks.txt"), emit: peaks

    script:
    """
    findPeaks ${ip_tags} -style chipseq -i ${control_tags} -o ${rep}_peaks.txt
    """
}

My workflow section looks like this:

TAGDIR(BOWTIE2_ALIGN.out.bam)

ip_ch = TAGDIR.out.tags
    .filter { it[0].startsWith("IP") }
    .map { sample, name, path ->
        def rep = sample.find(/rep\d+/)
        [rep, sample, name, path]
    }

input_ch = TAGDIR.out.tags
    .filter { it[0].startsWith("INPUT") }
    .map { sample, name, path ->
        def rep = sample.find(/rep\d+/)
        [rep, sample, name, path]
    }

find_peaks_input = ip_ch
    .join(input_ch)
    .map { ip, ctrl ->
        def rep = ip[0]
        [rep, ip[1], ip[2], ip[3], ctrl[1], ctrl[2], ctrl[3]]
    }

FINDPEAKS(find_peaks_input)

I am trying to match up input and IP samples for the same replicate to then use findPeaks. I am working with 2 replicates for now.

Why does Nextflow think I’m invoking FINDPEAKS with a list instead of a channel? Is there something subtle about how .join() and .map() interact with workflow scope or channel emission?