Class: MixinBot::Blaze::Reactor

Inherits:
Object
  • Object
show all
Defined in:
lib/mixin_bot/blaze/reactor.rb

Overview

Fiber-based Blaze receive loop — the hosted sibling of examples/blaze_async.rb.

Drives API#blaze_async inside its own Async reactor (created by #run, so it can be called from any plain thread — e.g. a Puma plugin's background thread or a forked child). The loop: connect, request pending messages, read frames serially, dispatch each decoded envelope to the configured handler, and acknowledge per the ack policy. Receipt confirmations (ACKNOWLEDGE_MESSAGE_RECEIPT frames) are consumed internally — never dispatched to the handler nor re-acknowledged. Dropped connections are re-established with bounded exponential backoff.

Handlers run serially on the reactor thread — never in concurrent fibers — so handler code may use per-thread resources like ActiveRecord connections safely. Handler exceptions are logged and do not end message delivery.

Reactor.new(handler: ->(envelope) { ... }).run   # blocks forever

Constant Summary collapse

CONNECT_TIMEOUT = 
10
KEEPALIVE_INTERVAL = 
30
INITIAL_BACKOFF = 
1
MAX_BACKOFF = 
30
STABLE_PERIOD =

a connection must live at least this long for the backoff to reset

30

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(handler:, api: MixinBot.api, ack_policy: MixinBot.config.blaze_ack_policy, connection_factory: nil, keepalive_interval: KEEPALIVE_INTERVAL, connect_timeout: CONNECT_TIMEOUT, sleeper: nil, logger: nil) ⇒ Reactor

Returns a new instance of Reactor.

Parameters:

  • api (MixinBot::API) (defaults to: MixinBot.api) —

    API used for frame codecs and connecting

  • handler (#call) —

    invoked with each decoded message envelope Hash (with action and data keys)

  • ack_policy (Symbol) (defaults to: MixinBot.config.blaze_ack_policy) —

    :on_receipt (ack before the handler runs) or :after_handler (ack only after the handler succeeds; failed messages are redelivered on reconnect)

  • connection_factory (#call, nil) (defaults to: nil) —

    returns a connected Async::WebSocket connection; defaults to api.blaze_async with a connect_timeout connect phase

  • keepalive_interval (Numeric) (defaults to: KEEPALIVE_INTERVAL) —

    seconds between client pings

  • connect_timeout (Numeric) (defaults to: CONNECT_TIMEOUT) —

    seconds allowed for the connect phase (DNS + TCP + TLS + WebSocket upgrade); established connections are never timed out on read

  • sleeper (#call, nil) (defaults to: nil) —

    backoff hook, called with seconds; defaults to Kernel#sleep (inject for tests)

  • logger (#call, nil) (defaults to: nil) —

    called with (level, exception_or_message); defaults to $stderr

Raises:



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/mixin_bot/blaze/reactor.rb', line 56

def initialize(handler:, api: MixinBot.api, ack_policy: MixinBot.config.blaze_ack_policy,
               connection_factory: nil, keepalive_interval: KEEPALIVE_INTERVAL,
               connect_timeout: CONNECT_TIMEOUT, sleeper: nil, logger: nil)
  raise MixinBot::ArgumentError, 'handler must respond to #call' unless handler.respond_to?(:call)
  unless %i[on_receipt after_handler].include?(ack_policy)
    raise MixinBot::ArgumentError, "ack_policy must be :on_receipt or :after_handler, got #{ack_policy.inspect}"
  end

  @api = api
  @handler = handler
  @ack_policy = ack_policy
  # No socket-level timeout: endpoint timeout: would apply to every read
  # (a quiet-but-healthy connection would be killed), not just the
  # connect phase. Liveness is the keepalive ping + the connect-phase
  # timeout applied in #connect!.
  @connection_factory = connection_factory || -> { api.blaze_async }
  @keepalive_interval = keepalive_interval
  @connect_timeout = connect_timeout
  @sleeper = sleeper || ->(seconds) { sleep seconds }
  @logger = logger || ->(level, detail) { warn "[mixin_blaze] #{level}: #{detail}" }
  @guard = Mutex.new
  @stopping = false
  @connection = nil
end

Instance Attribute Details

#ack_policy ⇒ Object (readonly)

Returns the value of attribute ack_policy.



36
37
38
 
# File 'lib/mixin_bot/blaze/reactor.rb', line 36

def ack_policy
  @ack_policy
end

#handler ⇒ Object (readonly)

Returns the value of attribute handler.



36
37
38
 
# File 'lib/mixin_bot/blaze/reactor.rb', line 36

def handler
  @handler
end

Instance Method Details

#run ⇒ void

This method returns an undefined value.

Runs the connect/read/reconnect loop, blocking the calling thread until #stop. Yields control to the fiber scheduler while waiting.



86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/mixin_bot/blaze/reactor.rb', line 86

def run
  Async do |task|
    @task = task
    backoff = INITIAL_BACKOFF

    until stopped?
      started_at = monotonic_time
      begin
        run_connection
        backoff = INITIAL_BACKOFF if stable?(started_at)
      rescue StandardError => e
        log.call :error, e
        backoff = INITIAL_BACKOFF if stable?(started_at)
      end

      break if stopped?

      sleep_backoff backoff
      backoff = [(backoff * 2), MAX_BACKOFF].min
    end
  end
end

#stop ⇒ void

This method returns an undefined value.

Stops the loop and closes the current connection. Safe to call from any thread (including while #run blocks in another one) and safe to call more than once.



115
116
117
118
119
120
121
122
123
124
 
# File 'lib/mixin_bot/blaze/reactor.rb', line 115

def stop
  @guard.synchronize do
    @stopping = true
    begin
      @connection&.close
    rescue StandardError
      # already dead; the read loop will notice on its own
    end
  end
end