Module: Shifty::DSL

Defined in:
lib/shifty/dsl.rb

Defined Under Namespace

Classes: BatchContext

Instance Method Summary collapse

Instance Method Details

#batch_worker(options = {}, &block) ⇒ Object 



73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/shifty/dsl.rb', line 73

def batch_worker(options = {}, &block)
  options[:tags] ||= []
  options[:tags] << :batch
  options[:gathering] ||= 1

  ensure_regular_arity!(block) if block
  batch_full = block ||
    proc { |_, batch| batch.size >= options[:gathering] }

  options[:context] = BatchContext.new({batch_full: batch_full})

  Worker.new(options) do |value, supply, context|
    if value
      context.collection = [value]
      until context.batch_complete?(
        context.collection.last,
        context.collection
      )
        context.collection << supply.shift
      end
      context.collection.compact
    end
  end
end

#filter_worker(options = {}, &block) ⇒ Object 



53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/shifty/dsl.rb', line 53

def filter_worker(options = {}, &block)
  options[:tags] ||= []
  options[:tags] << :filter
  ensure_callable!(block)

  Worker.new(options) do |value, supply|
    while value && !block.call(value)
      value = supply.shift
    end
    value
  end
end

#handoff(something) ⇒ Object 



140
141
142
 
# File 'lib/shifty/dsl.rb', line 140

def handoff(something)
  Fiber.yield something
end

#relay_worker(options = {}, &block) ⇒ Object 



20
21
22
23
24
25
26
27
28
 
# File 'lib/shifty/dsl.rb', line 20

def relay_worker(options = {}, &block)
  options[:tags] ||= []
  options[:tags] << :relay
  ensure_regular_arity!(block)

  Worker.new(options) do |value|
    value && block.call(value)
  end
end

#side_worker(options = {}, &block) ⇒ Object 



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/shifty/dsl.rb', line 30

def side_worker(options = {}, &block)
  options[:tags] ||= []
  options[:tags] << :side_effect
  deprecate_mode_option!(options)
  ensure_regular_arity!(block)

  # The block must ask the worker for its policy at shift time, so the
  # local is captured by the closure before it is assigned. Not redundant.
  worker = nil
  worker = Worker.new(options) do |value| # standard:disable Style/RedundantAssignment
    value.tap do |v|
      next unless v
      # Under :isolated the block observes a private scratch copy, so
      # its mutations evaporate and the untouched value flows on.
      used_value = (worker.effective_policy == :isolated) ?
        Policy::Isolated.call(v, worker: worker) : v

      block.call(used_value)
    end
  end
  worker
end

#source_worker(argument = nil, &block) ⇒ Object 



3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
 
# File 'lib/shifty/dsl.rb', line 3

def source_worker(argument = nil, &block)
  ensure_correct_arity_for!(argument, block)

  series = series_from(argument)
  callable = setup_callable_for(block, series)

  return Worker.new(&callable) if series.nil?

  Worker.new(tags: [:source]) do
    series.each(&callable)

    loop do
      handoff nil
    end
  end
end

#splitter_worker(options = {}, &block) ⇒ Object 



98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/shifty/dsl.rb', line 98

def splitter_worker(options = {}, &block)
  options[:tags] ||= []
  options[:tags] << :splitter
  ensure_regular_arity!(block)

  Worker.new(options) do |value|
    if value.nil?
      value
    else
      parts = [block.call(value)].flatten
      while parts.size > 1
        handoff parts.shift
      end
      parts.shift
    end
  end
end

#trailing_worker(trail_length = 2) ⇒ Object

don't like that this is a second exception to accepting options..



117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
 
# File 'lib/shifty/dsl.rb', line 117

def trailing_worker(trail_length = 2)
  options = {tags: [:trailing]}
  trail = []
  Worker.new(options) do |value, supply|
    if value
      trail.unshift value
      if trail.size >= trail_length
        trail.pop
      end
      while trail.size < trail_length
        trail.unshift supply.shift
      end

      # Hand off a snapshot: the builder keeps mutating `trail` across
      # calls, and a downstream :frozen intake would freeze the live
      # closure array in place.
      trail.dup
    else
      value # hint: it's nil!
    end
  end
end