Class: NNQ::Engine::SocketLifecycle

Inherits:
Object
  • Object
show all
Defined in:
lib/nnq/engine/socket_lifecycle.rb

Overview

Owns the socket-level state: ‘:new → :open → :closing → :closed` and the captured parent task for the socket’s task tree.

Engine delegates state queries here and uses it to coordinate the ordering of close-time side effects. Mirrors OMQ’s SocketLifecycle without the heartbeat/mechanism/monitor machinery nnq doesn’t need.

Defined Under Namespace

Classes: InvalidTransition

Constant Summary collapse

STATES = 
%i[new open closing closed].freeze
TRANSITIONS = 
{
  new:     %i[open closed].freeze,
  open:    %i[closing closed].freeze,
  closing: %i[closed].freeze,
  closed:  [].freeze,
}.freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ SocketLifecycle

Returns a new instance of SocketLifecycle.



57
58
59
60
61
62
63
64
65
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 57

def initialize
  @state             = :new
  @parent_task       = nil
  @on_io_thread      = false
  @peer_connected    = Async::Promise.new
  @all_peers_gone    = Async::Promise.new
  @reconnect_enabled = true
  @barrier           = nil
end

Instance Attribute Details

#all_peers_gone ⇒ Async::Promise (readonly)

Returns resolves with true the first time the connection set becomes empty after at least one peer connected. Edge-triggered: does not re-arm on reconnect.

Returns:

  • (Async::Promise) —

    resolves with true the first time the connection set becomes empty after at least one peer connected. Edge-triggered: does not re-arm on reconnect.



44
45
46
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 44

def all_peers_gone
  @all_peers_gone
end

#barrier ⇒ Async::Barrier? (readonly)

Returns holds every socket-scoped task (connection supervisors, reconnect loops, accept loops). NNQ::Engine#close calls barrier.stop to cascade teardown through every per-connection barrier in one shot.

Returns:

  • (Async::Barrier, nil) —

    holds every socket-scoped task (connection supervisors, reconnect loops, accept loops). NNQ::Engine#close calls barrier.stop to cascade teardown through every per-connection barrier in one shot.



50
51
52
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 50

def barrier
  @barrier
end

#on_io_thread ⇒ Boolean (readonly)

Returns true if parent_task is the shared Reactor thread.

Returns:

  • (Boolean) —

    true if parent_task is the shared Reactor thread



35
36
37
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 35

def on_io_thread
  @on_io_thread
end

#parent_task ⇒ Async::Task? (readonly)

Returns root of the socket’s task tree.

Returns:

  • (Async::Task, nil) —

    root of the socket’s task tree



32
33
34
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 32

def parent_task
  @parent_task
end

#peer_connected ⇒ Async::Promise (readonly)

Returns resolves with the first connected peer (or nil if the socket closes before anyone connects).

Returns:

  • (Async::Promise) —

    resolves with the first connected peer (or nil if the socket closes before anyone connects)



39
40
41
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 39

def peer_connected
  @peer_connected
end

#reconnect_enabled ⇒ Boolean

Returns when false, the engine must not schedule new reconnect attempts. Default true.

Returns:

  • (Boolean) —

    when false, the engine must not schedule new reconnect attempts. Default true.



54
55
56
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 54

def reconnect_enabled
  @reconnect_enabled
end

#state ⇒ Symbol (readonly)

Returns:

  • (Symbol) 


29
30
31
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 29

def state
  @state
end

Instance Method Details

#alive? ⇒ Boolean

Returns:

  • (Boolean) 


71
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 71

def alive?   = @state == :new || @state == :open

#capture_parent_task(task, on_io_thread:) ⇒ Boolean

Captures task as this socket’s task tree root. Transitions ‘:new → :open`. Idempotent: second call is a no-op.

Parameters:

  • task (Async::Task)
  • on_io_thread (Boolean) —

    true when task is the shared NNQ::Reactor root task (vs. the caller’s own Async task)

Returns:

  • (Boolean) —

    true on first-time capture, false if already captured



81
82
83
84
85
86
87
88
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 81

def capture_parent_task(task, on_io_thread:)
  return false if @parent_task
  @parent_task  = task
  @on_io_thread = on_io_thread
  @barrier      = Async::Barrier.new(parent: @parent_task)
  transition!(:open)
  true
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


70
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 70

def closed?  = @state == :closed

#closing? ⇒ Boolean

Returns:

  • (Boolean) 


69
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 69

def closing? = @state == :closing

#finish_closing! ⇒ Object

Transitions ‘:closing → :closed` (or `:new → :closed` for never-opened sockets).



99
100
101
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 99

def finish_closing!
  transition!(:closed)
end

#open? ⇒ Boolean

Returns:

  • (Boolean) 


68
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 68

def open?    = @state == :open

#resolve_all_peers_gone_if_empty(connections) ⇒ Object

Resolves ‘all_peers_gone` if we had peers and now have none. Idempotent.

Parameters:

  • connections (Hash) —

    current connection map



107
108
109
110
111
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 107

def resolve_all_peers_gone_if_empty(connections)
  return unless @peer_connected.resolved? && connections.empty?
  return if @all_peers_gone.resolved?
  @all_peers_gone.resolve(true)
end

#start_closing! ⇒ Object

Transitions ‘:open → :closing`.



92
93
94
 
# File 'lib/nnq/engine/socket_lifecycle.rb', line 92

def start_closing!
  transition!(:closing)
end