Module: Takagi::Base::ReactorManagement

Included in:
Takagi::Base, Controller
Defined in:
lib/takagi/base/reactor_management.rb

Overview

Manages reactor registration and lifecycle for observable/observer patterns.

Supports both inline reactor definitions and file-based reactor classes. Provides DSL for defining observables and observers.

Instance Method Summary collapse

Instance Method Details

#reactor(threads: 2, name: nil, interval: 1.0) { ... } ⇒ Object

Defines a reactor using a DSL block (inline definition)

The block is evaluated in the context of a new Reactor instance, allowing you to define observables and observers inline.

Examples:

Event-driven observable

reactor threads: 4 do
  observable '/temperature' do |emitter|
    emitter.on_event('sensor.temp.changed')
  end
end

Polling observable

reactor do
  observable_polling '/status', interval: 5.0 do
    { status: check_status, uptime: uptime }
  end
end

Remote observation

reactor do
  observe 'coap://sensor:5683/temp' do |data|
    process_temperature(data)
  end
end

Parameters:

  • threads (Integer) (defaults to: 2) —

    Number of threads for the reactor (default: 2)

  • name (String) (defaults to: nil) —

    Optional name for the reactor

  • interval (Float) (defaults to: 1.0) —

    Default polling interval in seconds (default: 1.0)

Yields:

  • Block defining reactor configuration



40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/takagi/base/reactor_management.rb', line 40

def reactor(threads: 2, name: nil, interval: 1.0, &block)
  reactor_name = name || "#{self.name.split('::').last.downcase}-reactor"
  reactor_instance = Takagi::Observable::Reactor.new(
    threads: threads,
    name: reactor_name,
    interval: interval
  )
  reactor_instance.instance_eval(&block)

  # Register with Observable::Registry
  Takagi::Observable::Registry.register(reactor_name.to_sym, reactor_instance)

  reactor_instance
end

#start_reactors ⇒ void

This method returns an undefined value.

Starts all registered reactors

Called automatically by server lifecycle management during boot. Can be called manually if needed.



92
93
94
 
# File 'lib/takagi/base/reactor_management.rb', line 92

def start_reactors
  Takagi::Observable::Registry.start_all
end

#stop_reactors ⇒ void

This method returns an undefined value.

Stops all registered reactors



99
100
101
 
# File 'lib/takagi/base/reactor_management.rb', line 99

def stop_reactors
  Takagi::Observable::Registry.stop_all
end

#use_reactor(name, klass_or_instance) ⇒ Object

Registers a file-based reactor class instance

Use this to register reactors defined in separate files (e.g., app/reactors/). The reactor class should inherit from Takagi::Observable::Reactor.

Examples:

Register reactor class

class IotReactor < Takagi::Observable::Reactor
  def initialize
    super(threads: 8, name: 'iot')
    setup_device_observables
  end
end

use_reactor :iot, IotReactor

Register reactor instance

use_reactor :sensors, SensorReactor.new

Parameters:

  • name (Symbol) —

    Unique identifier for the reactor

  • klass_or_instance (Class, Reactor) —

    Reactor class or instance



75
76
77
78
79
80
81
82
83
84
 
# File 'lib/takagi/base/reactor_management.rb', line 75

def use_reactor(name, klass_or_instance)
  reactor_instance = if klass_or_instance.is_a?(Class)
                       klass_or_instance.new
                     else
                       klass_or_instance
                     end

  Takagi::Observable::Registry.register(name.to_sym, reactor_instance)
  reactor_instance
end