Class: Plumbing::Actor::Threaded

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

Overview

Processes an actor's messages on its own dedicated thread, ONE AT A TIME, IN ARRIVAL ORDER. Concurrency is between actors (each has its own thread and queue), never within one. Built on core Ruby — no concurrent-ruby. Arguments are passed by reference — there is no marshalling / Ractor-safe copying variant.

The Worker is a frozen Literal::Data, so mutable state lives in container objects (a Queue, a Mutex, a one-element Array holding the thread) rather than reassignable ivars. Closing the queue makes pop return nil, which cleanly stops the consumer.

Direct Known Subclasses

Rails

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) 


31
 
# File 'lib/plumbing/actor/threaded.rb', line 31

def active? = !@queue.closed?

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



38
39
40
41
42
43
44
45
46
47
 
# File 'lib/plumbing/actor/threaded.rb', line 38

def after(delay, method:, sender: nil, params: {}, block: nil)
  call
  message = build_message(method: method, sender: sender, params: params, block: block)
  deferral = Plumbing::Actor::Deferral.new
  Thread.new do
    sleep delay
    dispatch(message) unless deferral.cancelled?
  end
  deferral
end

#call ⇒ Object Also known as: start 



23
24
25
26
 
# File 'lib/plumbing/actor/threaded.rb', line 23

def call
  @lock.synchronize { @runner << Thread.new { run_loop } if @runner.empty? }
  self
end

#dispatch(message) ⇒ Object 



33
34
35
36
 
# File 'lib/plumbing/actor/threaded.rb', line 33

def dispatch(message)
  call
  @queue.push(message)
end

#message_class ⇒ Object 



49
 
# File 'lib/plumbing/actor/threaded.rb', line 49

def message_class = Plumbing::Actor::Threaded::Message

#stop ⇒ Object 



29
 
# File 'lib/plumbing/actor/threaded.rb', line 29

def stop = @queue.close