Module: Ractor::Pipeline::Stages

Included in:
Branch, Source
Defined in:
lib/ractor/pipeline.rb

Overview

Common stage-building vocabulary for Source and Branch.

Instance Method Summary collapse

Instance Method Details

#filter_pipe(lanes: 1, &block) ⇒ Object

Send the element itself downstream iff the block returns truthy.



88
89
90
 
# File 'lib/ractor/pipeline.rb', line 88

def filter_pipe(lanes: 1, &block)
  add_stage(:filter_pipe, lanes, block)
end

#flat_pipe(lanes: 1, &block) ⇒ Object

The block returns an each-able object; each of its elements is sent downstream (1 input -> N outputs).



94
95
96
 
# File 'lib/ractor/pipeline.rb', line 94

def flat_pipe(lanes: 1, &block)
  add_stage(:flat_pipe, lanes, block)
end

#pipe(lanes: 1, &block) ⇒ Object

Apply the block to each element and send the return value downstream.



83
84
85
 
# File 'lib/ractor/pipeline.rb', line 83

def pipe(lanes: 1, &block)
  add_stage(:pipe, lanes, block)
end

#stages ⇒ Object 



80
 
# File 'lib/ractor/pipeline.rb', line 80

def stages = @stages ||= []

#tee(*branches) ⇒ Object

Broadcast each element to every branch. Branch outputs are merged (unordered) into one downstream stream.

stream(src).tee(
pipe{ A(it) },
pipe{ B(it) },
).to_a


105
106
107
108
109
110
111
112
 
# File 'lib/ractor/pipeline.rb', line 105

def tee(*branches)
  branches.each do |br|
    raise ArgumentError, "tee branch must be a stage list (use pipe{}/filter_pipe{})" unless Branch === br
    raise ArgumentError, "tee branch must have at least one stage" if br.stages.empty?
  end
  stages << Stage.new(:tee, nil, nil, branches.map(&:stages))
  self
end