Class: Cordis::Events

Inherits:
Object
  • Object
show all
Defined in:
lib/cordis/events.rb

Overview

Event system (upstream 4.0's EventsService): registering a listener is itself a fiber effect, so listeners are removed automatically when the fiber unloads. emit/bail/waterfall are synchronous; parallel requires a reactor.

Defined Under Namespace

Classes: Hook

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(ctx) ⇒ Events

Returns a new instance of Events.



14
15
16
17
 
# File 'lib/cordis/events.rb', line 14

def initialize(ctx)
  @ctx = ctx
  @hooks = Hash.new { |hash, key| hash[key] = [] }
end

Instance Attribute Details

#hooks ⇒ Object (readonly)

Returns the value of attribute hooks.



12
13
14
 
# File 'lib/cordis/events.rb', line 12

def hooks
  @hooks
end

Instance Method Details

#bail(name) ⇒ Object Also known as: serial

Sequential calls; returns the first non-nil/false result. serial is the same semantics named for async listeners: Ruby async is blocking-style, a listener just blocks, so the two share one implementation.



44
45
46
47
48
49
50
 
# File 'lib/cordis/events.rb', line 44

def bail(name, *)
  @hooks[name].dup.each do |hook|
    result = hook.callback.call(*)
    return result if bailed?(result)
  end
  nil
end

#emit(name) ⇒ Object

Synchronous fan-out; return values ignored; listener exceptions propagate (matches upstream).



36
37
38
39
 
# File 'lib/cordis/events.rb', line 36

def emit(name, *)
  @hooks[name].dup.each { |hook| hook.callback.call(*) }
  nil
end

#once(ctx, name, prepend: false, &callback) ⇒ Object 



28
29
30
31
32
33
 
# File 'lib/cordis/events.rb', line 28

def once(ctx, name, prepend: false, &callback)
  dispose = register(ctx, name, prepend: prepend) do |*args|
    dispose.call
    callback.call(*args)
  end
end

#parallel(name) ⇒ Object

Call all listeners concurrently without short-circuiting: exceptions are collected into an AggregateError after everything ran (requires Sync/Async).

Raises:



72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/cordis/events.rb', line 72

def parallel(name, *)
  parent = Async::Task.current
  tasks = @hooks[name].dup.map do |hook|
    parent.async { hook.callback.call(*) }
  end
  errors = []
  tasks.each do |task|
    task.wait
  rescue StandardError => e
    errors << e
  end
  raise AggregateError, errors unless errors.empty?

  nil
end

#register(ctx, name, prepend: false, &callback) ⇒ Object 



19
20
21
22
23
24
25
26
 
# File 'lib/cordis/events.rb', line 19

def register(ctx, name, prepend: false, &callback)
  list = @hooks[name]
  hook = Hook.new(ctx, callback)
  ctx.fiber.effect("ctx.on(#{name.inspect})") do
    prepend ? list.unshift(hook) : list.push(hook)
    -> { list.reject! { |h| h.equal?(hook) } }
  end
end

#waterfall(name, &fallback) ⇒ Object

Middleware chain: each listener receives an extra next (callable); not calling next stops the chain; the fallback block runs only when every listener passed through.



55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/cordis/events.rb', line 55

def waterfall(name, *, &fallback)
  chain = @hooks[name].dup
  index = -1
  step = nil
  step = lambda do |*current|
    index += 1
    if index < chain.size
      chain[index].callback.call(*current, step)
    elsif fallback
      fallback.call(*current)
    end
  end
  step.call(*)
end