Class: Ractor::Pipeline::Source

Inherits:
Object
  • Object
show all
Includes:
Stages
Defined in:
lib/ractor/pipeline.rb

Overview

A pipeline with a source. Terminal operations are defined here.

Defined Under Namespace

Classes: Node

Instance Method Summary collapse

Methods included from Stages

#filter_pipe, #flat_pipe, #pipe, #stages, #tee

Constructor Details

#initialize(source, batch: 1) ⇒ Source

Returns a new instance of Source.

Raises:

  • (ArgumentError) 


135
136
137
138
139
140
 
# File 'lib/ractor/pipeline.rb', line 135

def initialize(source, batch: 1)
  raise ArgumentError, "source must respond to #each" unless source.respond_to?(:each)
  raise ArgumentError, "batch: must be an Integer >= 1" unless Integer === batch && batch >= 1
  @source = source
  @batch = batch
end

Instance Method Details

#count ⇒ Object 



163
164
165
166
167
 
# File 'lib/ractor/pipeline.rb', line 163

def count
  n = 0
  run { n += 1 }
  n
end

#each(&block) ⇒ Object

-- terminal operations (executed in the caller Ractor) -----------

Raises:

  • (ArgumentError) 


144
145
146
147
148
 
# File 'lib/ractor/pipeline.rb', line 144

def each(&block)
  raise ArgumentError, "no block given" unless block
  run { |msg| block.call(msg) }
  self
end

#first(n = nil) ⇒ Object

Terminates the pipeline as soon as enough elements are received.



170
171
172
173
174
175
176
177
178
179
180
 
# File 'lib/ractor/pipeline.rb', line 170

def first(n = nil)
  want = n || 1
  result = []
  if want > 0
    run do |msg|
      result << msg
      throw STOP, true if result.size >= want
    end
  end
  n ? result : result.first
end

#reduce(initial, &block) ⇒ Object

Raises:

  • (ArgumentError) 


150
151
152
153
154
155
 
# File 'lib/ractor/pipeline.rb', line 150

def reduce(initial, &block)
  raise ArgumentError, "no block given" unless block
  acc = initial
  run { |msg| acc = block.call(acc, msg) }
  acc
end

#to_a ⇒ Object 



157
158
159
160
161
 
# File 'lib/ractor/pipeline.rb', line 157

def to_a
  result = []
  run { |msg| result << msg }
  result
end