Closure-Based Channel Operators

nf-bids extends Nextflow with three closure-driven operators — groupTupleBy, joinBy, and combineBy — that work with maps and arbitrary Groovy objects instead of requiring fixed-position tuples.

Why closure-based operators?

Nextflow’s built-in groupTuple, join, and combine operators require tuple index keys:

// Built-in: tuple structure required, index is opaque
channel.of(['sub-01', 'file.nii'], ['sub-01', 'other.nii'])
       .groupTuple(by: 0)

With nf-bids flat output, items are maps — not tuples. Closure-based operators read any field without restructuring:

// Closure-based: works with maps, self-documenting
channel.of([subject: 'sub-01', file: 'file.nii'],
           [subject: 'sub-01', file: 'other.nii'])
       .groupTupleBy { it.subject }

Importing the operators

include { groupTupleBy } from 'plugin/nf-bids'
include { joinBy }       from 'plugin/nf-bids'
include { combineBy }    from 'plugin/nf-bids'

groupTupleBy

Signature

channel.groupTupleBy(Closure keyExtractor)
channel.groupTupleBy(Closure keyExtractor, Map opts)

Output

Emits [key, [items]] tuples.

Options

Option Type Description

size

Integer

Emit a group as soon as it reaches this many items (streaming groups).

sort

Boolean, Closure, or Comparator

true for natural order; a closure such as { it.run } or a comparator for custom ordering.

remainder

Boolean

When true (default), emit incomplete groups when the channel closes.

Examples

Group by single field
Channel
    .of([subject: 'sub-01', file: 'a.nii'],
        [subject: 'sub-01', file: 'b.nii'],
        [subject: 'sub-02', file: 'c.nii'])
    .groupTupleBy { it.subject }
// → ['sub-01', [[subject:'sub-01', file:'a.nii'],
//               [subject:'sub-01', file:'b.nii']]]
//   ['sub-02', [[subject:'sub-02', file:'c.nii']]]
Composite key (subject + session)
channel.groupTupleBy { [it.subject, it.session] }
// Closure returns a List → wrapped automatically in CompositeKey
// → [[sub-01, ses-01], [ [item1, item2] ]]
Sort items within each group
channel.groupTupleBy({ it.subject }, [sort: { it.run }])
Streaming groups of 3
channel.groupTupleBy({ it.batch }, [size: 3, remainder: false])

joinBy

Signature

left.joinBy(DataflowReadChannel right, Closure keyExtractor)
left.joinBy(DataflowReadChannel right, Closure keyExtractor, Map opts)
left.joinBy(DataflowReadChannel right, Closure leftKey, Closure rightKey)
left.joinBy(DataflowReadChannel right, Closure leftKey, Closure rightKey, Map opts)

When a single keyExtractor is supplied it is applied to items from both channels. Each join closure is a key extractor: nf-bids calls it with one item from its respective channel and matches the returned keys.

Output

Emits one fused item for every matching pair. No separate join-key tuple is emitted.

Options

Option Type Description

remainder

Boolean

When true, emit unmatched items with the available side preserved as-is (outer-join semantics).

Examples

Join anatomical and functional by subject
anatomical = Channel.of([subject: 'sub-01', t1: 't1.nii'],
                        [subject: 'sub-02', t1: 't1.nii'])
functional = Channel.of([subject: 'sub-01', bold: 'bold.nii'],
                        [subject: 'sub-02', bold: 'bold.nii'])

anatomical.joinBy(functional) { it.subject }
// → [subject:'sub-01', t1:'t1.nii', bold:'bold.nii']
//   [subject:'sub-02', ...]
Different key fields on each channel
scans.joinBy(metadata,
             { it.scan_id },        // left key
             { it.participant_id }) // right key
Outer join (include unmatched)
left.joinBy(right, { it.id }, [remainder: true])
// Unmatched left items stay as [id:'C', val:3]
// Unmatched right items stay as [id:'D', data:'z']

combineBy

Signature

left.combineBy(DataflowReadChannel right, Closure keyExtractor)
left.combineBy(DataflowReadChannel right, Closure keyExtractor, Map opts)
left.combineBy(DataflowReadChannel right, Closure leftKey, Closure rightKey)
left.combineBy(DataflowReadChannel right, Closure leftKey, Closure rightKey, Map opts)

Like joinBy, the closures passed to combineBy are key extractors invoked with one item at a time from the left or right channel.

Output

Emits fused combined items — the full cartesian product of matching items within each key group. No separate key tuple is emitted.

Examples

Match subjects with sessions
subjects = Channel.of([id: 'sub-01', age: 25], [id: 'sub-02', age: 30])
sessions = Channel.of([id: 'sub-01', session: 'ses-01'],
                      [id: 'sub-01', session: 'ses-02'],
                      [id: 'sub-02', session: 'ses-01'])

subjects.combineBy(sessions, { it.id })
// → [id:'sub-01', age:25, session:'ses-01']
//   [id:'sub-01', age:25, session:'ses-02']
//   [id:'sub-02', age:30, session:'ses-01']
Filter after combining
subjects
    .combineBy(sessions, { it.id })
    .filter { fused -> fused.age >= 18 }

CompositeKey and KeyExtractor

When a closure returns a List, KeyExtractor.extractKey() wraps it in CompositeKey — an @EqualsAndHashCode-annotated class — before it is used as a HashMap key. This ensures [sub-01, ses-01] == [sub-01, ses-01] even across different List instances.

channel.groupTupleBy { [it.subject, it.session] }
// Returns List → becomes CompositeKey([sub-01, ses-01]) internally

KeyExtractor.validateKeyExtractor(closure, operatorName) checks that the closure accepts at least one parameter. Zero-parameter closures raise IllegalArgumentException at operator construction time.

Performance notes

  • groupTupleBy buffers all items in memory until the channel closes (or size is reached); for large datasets use size to stream groups.

  • joinBy performs cartesian product for duplicate keys; 1:N joins are fine, M:N should be avoided in hot paths.

  • combineBy always produces a full cartesian product; guard with .filter() to reduce downstream load.

See Closure Operators Migration Guide for side-by-side comparisons with built-in operators.

Related pages