Class: Plumbing::Actor::Async

Inherits:
Worker show all
Defined in:
lib/plumbing/actor/async.rb

Overview

Processes an actor's messages on the Async reactor, ONE AT A TIME, IN ARRIVAL ORDER — a single sequential consumer loop preserves the actor guarantee.

NOTE: we deliberately do NOT use Async::Queue#async. That spawns a new task per item (bounded by a semaphore), so a single actor would deliver several messages concurrently and complete them out of order — which defeats the entire point of an actor. Concurrency belongs BETWEEN actors (each has its own worker/queue/consumer), never within one.

Defined Under Namespace

Classes: Message

Instance Method Summary collapse

Methods inherited from Worker

#build_message, #cancel_deferred, #post

Instance Method Details

#active? ⇒ Boolean

Returns:

  • (Boolean) 


32
 
# File 'lib/plumbing/actor/async.rb', line 32

def active? = !@queue.closed?

#after(delay, method:, sender: nil, params: {}, block: nil) ⇒ Object 



36
37
38
39
40
41
42
43
44
 
# File 'lib/plumbing/actor/async.rb', line 36

def after(delay, method:, sender: nil, params: {}, block: nil)
  message = build_message(method: method, sender: sender, params: params, block: block)
  deferral = Plumbing::Actor::Deferral.new
  Kernel.Async(transient: true) do |task|
    task.sleep delay
    dispatch(message) unless deferral.cancelled?
  end
  deferral
end

#call ⇒ Object Also known as: start 



21
22
23
24
25
26
27
 
# File 'lib/plumbing/actor/async.rb', line 21

def call
  Kernel.Async(transient: true) do
    while (message = @queue.dequeue)
      message.deliver
    end
  end
end

#dispatch(message) ⇒ Object 



34
 
# File 'lib/plumbing/actor/async.rb', line 34

def dispatch(message) = @queue.enqueue(message)

#message_class ⇒ Object 



46
 
# File 'lib/plumbing/actor/async.rb', line 46

def message_class = Plumbing::Actor::Async::Message

#stop ⇒ Object 



30
 
# File 'lib/plumbing/actor/async.rb', line 30

def stop = @queue.close