Class: RubyAsterisk::AMI::Reactor

Inherits:
Object
  • Object
show all
Includes:
EventListAggregation
Defined in:
lib/ruby-asterisk/ami/reactor.rb

Overview

Hosts the AMI connection in two plain OS threads:

writer_thread — blocks on Thread::Queue#pop and writes commands to the
              AMI socket.  On :stop it closes the socket, which unblocks
              the reader thread's readpartial call.

reader_thread — loops on socket.readpartial, parses AMI frames with Parser,
              resolves pending Promises or fires the on_event callback.

Using plain Threads (no Fiber scheduler) ensures deterministic shutdown on all Ruby versions: closing an IO from one Thread immediately raises IOError in any other Thread blocked on that IO.

Thread-safety contract:

- External callers use #send_command, #register_promise,
#reject_all_promises, #stop — all thread-safe.
- Promise resolution via Mutex+CV (Promise class unchanged).
- No cross-thread IO inside the reactor; one thread reads, one writes.

Instance Method Summary collapse

Constructor Details

#initialize(host, port, on_event: nil, on_disconnect: nil) ⇒ Reactor

Returns a new instance of Reactor.



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/ruby-asterisk/ami/reactor.rb', line 30

def initialize(host, port, on_event: nil, on_disconnect: nil)
  @host          = host
  @port          = port
  @on_event      = on_event
  @on_disconnect = on_disconnect

  @command_queue  = Thread::Queue.new
  @promises_mutex = Mutex.new
  @promises       = {}
  @buffers        = {} # action_id => Array<raw frame> for in-flight EventList replies
  @socket         = nil
  @writer_thread  = nil
  @reader_thread  = nil
  @stopping       = false
end

Instance Method Details

#register_promise(action_id, promise) ⇒ Object

Register a Promise keyed by ActionID.



67
68
69
 
# File 'lib/ruby-asterisk/ami/reactor.rb', line 67

def register_promise(action_id, promise)
  @promises_mutex.synchronize { @promises[action_id] = promise }
end

#reject_all_promises(error) ⇒ Object

Reject all pending promises with the given error (thread-safe).



81
82
83
84
85
86
87
 
# File 'lib/ruby-asterisk/ami/reactor.rb', line 81

def reject_all_promises(error)
  promises = @promises_mutex.synchronize do
    @buffers.clear
    @promises.values.tap { @promises.clear }
  end
  promises.each { |p| p.reject(error) }
end

#send_command(cmd) ⇒ Object

Send a raw AMI command string from any external thread.



62
63
64
 
# File 'lib/ruby-asterisk/ami/reactor.rb', line 62

def send_command(cmd)
  @command_queue.push(cmd.freeze)
end

#start ⇒ Object

Open the socket, consume the AMI banner, and start both threads. Raises if the connection fails.



48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/ruby-asterisk/ami/reactor.rb', line 48

def start
  @socket = TCPSocket.new(@host, @port)
  @socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_KEEPALIVE, true)
  @socket.gets # consume AMI banner ("Asterisk Call Manager/x.y\n")

  @writer_thread = Thread.new { writer_loop }
  @reader_thread = Thread.new { reader_loop }
  self
rescue StandardError
  close_socket
  raise
end

#stop ⇒ Object

Stop both threads gracefully. Blocks until they exit.



90
91
92
93
94
95
96
97
98
99
 
# File 'lib/ruby-asterisk/ami/reactor.rb', line 90

def stop
  return unless @writer_thread&.alive? || @reader_thread&.alive?

  @stopping = true
  @command_queue.push(:stop) # wakes writer_thread immediately
  @writer_thread&.join(2)
  @reader_thread&.join(2)
  @writer_thread = nil
  @reader_thread = nil
end

#unregister_promise(action_id) ⇒ Object

Remove a pending Promise without resolving it (e.g. after a caller timeout) so the pending map does not grow unbounded.



73
74
75
76
77
78
 
# File 'lib/ruby-asterisk/ami/reactor.rb', line 73

def unregister_promise(action_id)
  @promises_mutex.synchronize do
    @promises.delete(action_id)
    @buffers.delete(action_id)
  end
end