Class: KubeMQ::Transport::ConnectionStateMachine Private

Inherits:
Object
  • Object
show all
Defined in:
lib/kubemq/transport/state_machine.rb

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Note:

This class is thread-safe. All state reads and transitions are mutex-protected.

Finite state machine governing the gRPC connection lifecycle.

Tracks transitions between ConnectionState values and fires registered callbacks on state changes. Invalid transitions raise Error.

Valid transition graph:

IDLE -> CONNECTING, CLOSED
CONNECTING -> READY, CLOSED, RECONNECTING, IDLE
READY -> RECONNECTING, CLOSED
RECONNECTING -> CONNECTING, READY, CLOSED
CLOSED -> (terminal)

See Also:

Constant Summary collapse

VALID_TRANSITIONS =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future.

Map of each state to its permitted successor states.

{
  ConnectionState::IDLE => [ConnectionState::CONNECTING, ConnectionState::CLOSED],
  ConnectionState::CONNECTING => [ConnectionState::READY, ConnectionState::CLOSED,
                                  ConnectionState::RECONNECTING, ConnectionState::IDLE],
  ConnectionState::READY => [ConnectionState::RECONNECTING, ConnectionState::CLOSED],
  ConnectionState::RECONNECTING => [ConnectionState::CONNECTING, ConnectionState::READY,
                                    ConnectionState::CLOSED],
  ConnectionState::CLOSED => []
}.freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ ConnectionStateMachine

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Creates a new state machine starting in ConnectionState::IDLE.



40
41
42
43
44
45
46
47
48
49
 
# File 'lib/kubemq/transport/state_machine.rb', line 40

def initialize
  @state = ConnectionState::IDLE
  @mutex = Mutex.new
  @callbacks = {
    on_connected: nil,
    on_disconnected: nil,
    on_reconnecting: nil,
    on_error: nil
  }
end

Instance Attribute Details

#state ⇒ Integer (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns the current ConnectionState value.

Returns:



37
38
39
 
# File 'lib/kubemq/transport/state_machine.rb', line 37

def state
  @state
end

Instance Method Details

#closed? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns whether the connection has been permanently closed.

Returns:

  • (Boolean) —

    true if in the terminal CLOSED state



129
130
131
 
# File 'lib/kubemq/transport/state_machine.rb', line 129

def closed?
  current_state == ConnectionState::CLOSED
end

#current_state ⇒ Integer

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns the current state with mutex synchronization.

Returns:



115
116
117
 
# File 'lib/kubemq/transport/state_machine.rb', line 115

def current_state
  @mutex.synchronize { @state }
end

#on_connected {|server_info| ... } ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Registers a callback invoked when the connection reaches READY.

Yields:

  • (server_info) —

    called with the broker's ServerInfo

Yield Parameters:

  • server_info (ServerInfo, nil) —

    broker information



56
57
58
 
# File 'lib/kubemq/transport/state_machine.rb', line 56

def on_connected(&block)
  @mutex.synchronize { @callbacks[:on_connected] = block }
end

#on_disconnected {|reason| ... } ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Registers a callback invoked when the connection reaches CLOSED.

Yields:

  • (reason) —

    called with a human-readable close reason

Yield Parameters:

  • reason (String) —

    reason for disconnection



65
66
67
 
# File 'lib/kubemq/transport/state_machine.rb', line 65

def on_disconnected(&block)
  @mutex.synchronize { @callbacks[:on_disconnected] = block }
end

#on_error {|error| ... } ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Registers a callback invoked on invalid state transitions.

Yields:

  • (error) —

    called with the Error describing the invalid transition

Yield Parameters:

  • error (Error) —

    the transition error



83
84
85
 
# File 'lib/kubemq/transport/state_machine.rb', line 83

def on_error(&block)
  @mutex.synchronize { @callbacks[:on_error] = block }
end

#on_reconnecting {|attempt| ... } ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Registers a callback invoked when the connection enters RECONNECTING.

Yields:

  • (attempt) —

    called with the current reconnect attempt number

Yield Parameters:

  • attempt (Integer) —

    attempt counter



74
75
76
 
# File 'lib/kubemq/transport/state_machine.rb', line 74

def on_reconnecting(&block)
  @mutex.synchronize { @callbacks[:on_reconnecting] = block }
end

#ready? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns whether the connection is in the READY state.

Returns:

  • (Boolean) —

    true if connected and operational



122
123
124
 
# File 'lib/kubemq/transport/state_machine.rb', line 122

def ready?
  current_state == ConnectionState::READY
end

#transition!(new_state, **context) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Transitions to a new state, firing the appropriate callback.

Parameters:

  • new_state (Integer) —

    the target ConnectionState value

  • context (Hash) —

    optional context passed to the callback (+:server_info+, :attempt, or :reason)

Raises:



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/kubemq/transport/state_machine.rb', line 94

def transition!(new_state, **context)
  callback = nil
  @mutex.synchronize do
    valid = VALID_TRANSITIONS.fetch(@state, [])
    unless valid.include?(new_state)
      error = KubeMQ::Error.new("Invalid state transition from #{@state} to #{new_state}")
      err_cb = @callbacks[:on_error]
      err_cb&.call(error)
      raise error
    end

    old_state = @state
    @state = new_state
    callback = resolve_callback(new_state, old_state, context)
  end
  callback&.call
end