Class: Rivulet::Stream

Inherits:
Object
  • Object
show all
Defined in:
lib/rivulet/stream.rb

Direct Known Subclasses

CountStream, MinMaxStream, StatsStream, SumStream

Instance Method Summary collapse

Constructor Details

#initialize(source) ⇒ Stream

Returns a new instance of Stream.



5
6
7
 
# File 'lib/rivulet/stream.rb', line 5

def initialize(source)
  @source = source
end

Instance Method Details

#iterate_windows(mode:, size: nil, rule: nil, step: 1, &block) ⇒ Object 



30
31
32
33
34
35
36
37
38
39
 
# File 'lib/rivulet/stream.rb', line 30

def iterate_windows(mode:, size: nil, rule: nil, step: 1, &block)
  case mode
  when :fixed     then iterate_fixed(size: size, step: step, &block)
  when :variable  then iterate_variable(rule: rule, &block)
  when :satisfied then iterate_satisfied(rule: rule, &block)
  # :nocov:
  else raise ArgumentError, "unknown mode: #{mode}"
  # :nocov:
  end
end

#max_window(&rule) ⇒ Object 



21
22
23
 
# File 'lib/rivulet/stream.rb', line 21

def max_window(&rule)
  WindowBuilder.new(stream: self, mode: :variable, rule: rule)
end

#min_window(&goal) ⇒ Object 



25
26
27
 
# File 'lib/rivulet/stream.rb', line 25

def min_window(&goal)
  WindowBuilder.new(stream: self, mode: :satisfied, rule: goal)
end

#tumbling(size, &block) ⇒ Object 



17
18
19
 
# File 'lib/rivulet/stream.rb', line 17

def tumbling(size, &block)
  windows(size, step: size, &block)
end

#windows(size, step: 1, &block) ⇒ Object

Raises:

  • (ArgumentError) 


9
10
11
12
13
14
15
 
# File 'lib/rivulet/stream.rb', line 9

def windows(size, step: 1, &block)
  raise ArgumentError, "size must be positive" unless size.positive?
  raise ArgumentError, "step must be positive" unless step.positive?

  builder = WindowBuilder.new(stream: self, mode: :fixed, size: size, step: step)
  block ? builder.each_window(&block) : builder
end