Class: FiberStream::Sink

Inherits:
Object
  • Object
show all
Defined in:
lib/fiber_stream/sink.rb,
sig/fiber_stream.rbs

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(&run) ⇒ Sink

Returns a new instance of Sink.



165
166
167
 
# File 'lib/fiber_stream/sink.rb', line 165

def initialize(&run)
  @run = run
end

Class Method Details

.all?(&block) ⇒ Boolean

Creates a sink that returns whether all elements match a predicate.

The sink pulls upstream until the block returns false or nil, or upstream completes. It returns false after a falsey predicate result and true when every predicate result is truthy, including for empty upstream.

Returns:

  • (Boolean)

Raises:

  • (ArgumentError) 


72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/fiber_stream/sink.rb', line 72

def self.all?(&block)
  raise ArgumentError, "missing block" unless block

  new do |stream|
    loop do
      value = stream.next
      break true if Pull.done?(value)
      break false unless block.call(value)
    end
  end
end

.any?(&block) ⇒ Boolean

Creates a sink that returns whether any element matches a predicate.

The sink pulls upstream until the block returns a truthy value or upstream completes. It returns true after a truthy predicate result and false when no element matches.

Returns:

  • (Boolean)

Raises:

  • (ArgumentError) 


55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/fiber_stream/sink.rb', line 55

def self.any?(&block)
  raise ArgumentError, "missing block" unless block

  new do |stream|
    loop do
      value = stream.next
      break false if Pull.done?(value)
      break true if block.call(value)
    end
  end
end

.build(&run) ⇒ Object

:nodoc:



161
162
163
 
# File 'lib/fiber_stream/sink.rb', line 161

def self.build(&run) # :nodoc:
  new(&run)
end

.count ⇒ void

This method returns an undefined value.

Creates a sink that counts all stream elements.

The sink consumes upstream until normal completion and returns the number of elements observed. It does not store consumed elements.



88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/fiber_stream/sink.rb', line 88

def self.count
  new do |stream|
    count = 0

    Pull.each_value(stream) do
      count += 1
    end

    count
  end
end

.find(&block) ⇒ void

This method returns an undefined value.

Creates a sink that returns the first element matching a predicate.

The sink pulls upstream until the block returns a truthy value or upstream completes. It returns the original matching element, or nil when no element matches. Matching nil elements are returned as nil, following Ruby's Enumerable#find ambiguity.

Raises:

  • (ArgumentError) 


38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/fiber_stream/sink.rb', line 38

def self.find(&block)
  raise ArgumentError, "missing block" unless block

  new do |stream|
    loop do
      value = stream.next
      break nil if Pull.done?(value)
      break value if block.call(value)
    end
  end
end

.first ⇒ void

This method returns an undefined value.

Creates a sink that returns the first stream element.

The sink pulls at most one element. It returns nil when upstream completes before producing a value.



25
26
27
28
29
30
 
# File 'lib/fiber_stream/sink.rb', line 25

def self.first
  new do |stream|
    value = stream.next
    Pull.done?(value) ? nil : value
  end
end

.fold(initial, &block) ⇒ void

This method returns an undefined value.

Creates a sink that folds all stream elements into an accumulator.

The sink consumes upstream until normal completion. It returns the final accumulator, or the initial accumulator when upstream is empty. Exceptions raised by the block fail the stream and are re-raised from Source#run_with. FiberStream assigns the initial accumulator directly; it does not duplicate or freeze that object.

Raises:

  • (ArgumentError) 


107
108
109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/fiber_stream/sink.rb', line 107

def self.fold(initial, &block)
  raise ArgumentError, "missing block" unless block

  new do |stream|
    accumulator = initial

    Pull.each_value(stream) do |value|
      accumulator = block.call(accumulator, value)
    end

    accumulator
  end
end

.foreach(&block) ⇒ void

This method returns an undefined value.

Creates a sink that runs a block for each stream element.

The sink consumes upstream until normal completion, calls the block once per element in input order, and returns the number of elements whose block completed successfully. Exceptions raised by the block fail the stream and are re-raised from Source#run_with.

Raises:

  • (ArgumentError) 


127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/fiber_stream/sink.rb', line 127

def self.foreach(&block)
  raise ArgumentError, "missing block" unless block

  new do |stream|
    count = 0

    Pull.each_value(stream) do |value|
      block.call(value)
      count += 1
    end

    count
  end
end

.io(io, close: false, flush: false) ⇒ Sink[String, Integer]

Creates a sink that writes String chunks to an IO-like object.

The sink consumes upstream until normal completion and returns the number of chunks successfully written. It requires a scheduler-backed non-blocking fiber before write, flush, or normal close operations. The IO object is closed only when close: true is passed, and flushed on normal completion only when flush: true is passed.

Parameters:

  • io (Object)
  • close: (Boolean) (defaults to: false)
  • flush: (Boolean) (defaults to: false)

Returns:

  • (Sink[String, Integer])

Raises:

  • (TypeError) 


149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/fiber_stream/sink.rb', line 149

def self.io(io, close: false, flush: false)
  raise TypeError, "io must respond to write" unless io.respond_to?(:write)
  raise TypeError, "close must be true or false" unless [true, false].include?(close)
  raise TypeError, "flush must be true or false" unless [true, false].include?(flush)
  raise TypeError, "io must respond to close" if close && !io.respond_to?(:close)
  raise TypeError, "io must respond to flush" if flush && !io.respond_to?(:flush)

  new do |stream|
    IOSink.new(io, close, flush).run(stream)
  end
end

.to_a ⇒ void

This method returns an undefined value.

Creates a sink that collects all stream elements into an Array.

The sink consumes upstream until normal completion and returns the collected array as the stream materialized value.



9
10
11
12
13
14
15
16
17
18
19
 
# File 'lib/fiber_stream/sink.rb', line 9

def self.to_a
  new do |stream|
    values = []

    Pull.each_value(stream) do |value|
      values << value
    end

    values
  end
end

Instance Method Details

#run_stream(stream) ⇒ Object

:nodoc:



171
172
173
 
# File 'lib/fiber_stream/sink.rb', line 171

def run_stream(stream) # :nodoc:
  @run.call(stream)
end