Class: Plumb::StreamClass

Inherits:
Object
  • Object
show all
Includes:
Steppable
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

Attributes included from Steppable

#name

Instance Method Summary collapse

Methods included from Steppable

#===, #>>, #as_node, #build, #check, #defer, #freeze, included, #inspect, #invalid, #invoke, #match, #meta, #node_name, #not, #pipeline, #policy, #to_s, #transform, #value, wrap, #|

Methods included from Callable

#metadata, #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 (Steppable) —

    type of the elements in the stream



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

def initialize(element_type: Types::Any)
  @element_type = Steppable.wrap(element_type)
  freeze
end

Instance Attribute Details

#element_type ⇒ Object (readonly)

Returns the value of attribute element_type.



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

def element_type
  @element_type
end

Instance Method Details

#[](element_type) ⇒ Object

return a new Stream definition.

Parameters:

  • element_type (Steppable) —

    the type of the elements in the stream



30
31
32
 
# File 'lib/plumb/stream_class.rb', line 30

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

#call(result) ⇒ Result::Valid, Result::Invalid

The [Step] interface

Parameters:

Returns:



37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/plumb/stream_class.rb', line 37

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

  enum = Enumerator.new do |y|
    result.value.each do |e|
      y << element_type.resolve(e)
    end
  end

  result.valid(enum)
end

#filtered ⇒ Step

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

Returns:

  • (Step) —

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



50
51
52
53
54
55
 
# File 'lib/plumb/stream_class.rb', line 50

def filtered
  self >> Step.new(nil, 'filtered') do |result|
    set = result.value.lazy.filter_map { |e| e.value if e.valid? }
    result.valid(set)
  end
end