Class: OMQ::CLI::TransientMonitor

Inherits:
Object
  • Object
show all
Defined in:
lib/omq/cli/transient_monitor.rb

Overview

Monitors peer-disconnect events for –transient mode.

Starts an async task that waits until #ready! is called (signalling that at least one message has been exchanged), then waits for all peers to disconnect, disables reconnection, and either stops the task (send-only) or closes the read side of the socket (recv side).

Instance Method Summary collapse

Constructor Details

#initialize(sock, config, task, log_fn) ⇒ TransientMonitor

Returns a new instance of TransientMonitor.

Parameters:

  • sock (Object) —

    the OMQ socket to monitor

  • config (Config) —

    frozen CLI configuration

  • task (Async::Task) —

    parent async task

  • log_fn (Method) —

    callable for verbose logging



17
18
19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/omq/cli/transient_monitor.rb', line 17

def initialize(sock, config, task, log_fn)
  @barrier = Async::Promise.new
  task.async do
    @barrier.wait
    sock.all_peers_gone.wait unless sock.connection_count == 0
    log_fn.call("all peers disconnected, exiting")
    sock.reconnect_enabled = false
    if config.send_only?
      task.stop
    else
      sock.close_read
    end
  end
end

Instance Method Details

#ready! ⇒ Object

Signal that the first message has been sent or received. Idempotent – safe to call multiple times.



36
37
38
 
# File 'lib/omq/cli/transient_monitor.rb', line 36

def ready!
  @barrier.resolve(true) unless @barrier.resolved?
end