Class: Plumb::StreamClass

Inherits:
Object
  • Object
show all
Includes:
Composable, CovariantFusion
Defined in:
lib/plumb/stream_class.rb

Overview

A stream that validates each element. Example:

row = Types::Tuple[String, Types::Lax::Integer]
csv_stream = Types::Stream[row]

stream = csv_stream.parse(CSV.new(File.new('data.csv')).to_enum)
stream.each |result|
result.valid? # => true
result.value # => ['name', 10]
end

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from CovariantFusion

#fuse_with

Methods included from Composable

#&, #/, #>>, #absorb_input, #absorb_output, #accepted_type, #as_node, #build, #check, #defer, #fusable_step?, #fuse_with, #generate, #idempotent?, included, #invalid, #invoke, #match, #metadata, #not, #output_type, #pipeline, #policy, resolve_operand, #static, #subtype_identity, #to_json_schema, #to_mermaid, #to_plumb_type, #to_s, #transform, #value, #value_preserving?, #where, #with, wrap, #|

Methods included from Callable

#parse, #resolve

Constructor Details

#initialize(element_type: Types::Any) ⇒ StreamClass

Returns a new instance of StreamClass.

Parameters:

  • element_type (Hash) (defaults to: Types::Any) —

    a customizable set of options

Options Hash (element_type:):

  • the (Composable) —

    type of the elements in the stream



24
25
26
27
28
 
# File 'lib/plumb/stream_class.rb', line 24

def initialize(element_type: Types::Any)
  @element_type = Composable.wrap(element_type)
  @children = [@element_type].freeze
  freeze
end

Instance Attribute Details

#children ⇒ Object (readonly)

Returns the value of attribute children.



21
22
23
 
# File 'lib/plumb/stream_class.rb', line 21

def children
  @children
end

Class Method Details

.each_interface ⇒ Object

A Stream consumes any enumerable (it checks #each at runtime in #call), not another Stream — so it reports Any as its input and opts out of the #>> composition check. This lets a producer of a raw Enumerator/Array (eg. a CSV enumerator) feed a Stream. Covariant Stream <= Stream subtyping is unaffected (that goes through #subtype_of? / #children). Memoized at the class level (instances are frozen; Types isn't loaded yet when this file is) — #call hits it on every data invocation.



46
 
# File 'lib/plumb/stream_class.rb', line 46

def self.each_interface = @each_interface ||= Types::Interface[:each]

Instance Method Details

#[](element_type) ⇒ Object 



35
36
37
 
# File 'lib/plumb/stream_class.rb', line 35

def [](element_type)
  self.class.new(element_type:)
end

#call(result) ⇒ Result

The [Step] interface

Parameters:

Returns:



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/plumb/stream_class.rb', line 53

def call(result)
  result = input_type.call(result)
  return result unless result.valid?
  # return result.invalid(errors: 'is not an Enumerable') unless result.value.respond_to?(:each)

  # Snapshot the source enumerable into a local so the lazy Enumerator closes
  # over IT, not the result cursor. The cursor may be reused/mutated after we
  # return (eg. a Stream nested in an Array, whose cursor is reset per
  # element) — closing over the snapshot keeps each stream bound to its own
  # input regardless.
  source = result.value
  enum = Enumerator.new do |y|
    source.each do |e|
      y << @element_type.resolve(e)
    end
  end

  # Copying #valid (not #valid!): the returned result's value IS the lazy
  # enum, so the result must be a fresh object the caller can hold while its
  # own cursor moves on.
  result.valid(enum)
end

#filtered ⇒ Composable

Returns a step that resolves to an Enumerator that filters out invalid elements.

Returns:

  • (Composable) —

    a step that resolves to an Enumerator that filters out invalid elements



77
78
79
80
81
82
 
# File 'lib/plumb/stream_class.rb', line 77

def filtered
  self >> Function.opaque(inspect: 'filtered', identity: [:filtered_stream, self]) do |result|
    set = result.value.lazy.filter_map { |e| e.value if e.valid? }
    result.valid(set)
  end
end

#input_type ⇒ Object 



48
 
# File 'lib/plumb/stream_class.rb', line 48

def input_type = StreamClass.each_interface

#with_children(children) ⇒ Object

return a new Stream definition.

Parameters:

  • element_type (Composable) —

    the type of the elements in the stream

See Also:



33
 
# File 'lib/plumb/stream_class.rb', line 33

def with_children(children) = self[children.first]