Class: KubeMQ::Transport::GrpcTransport Private

Inherits:
Object
  • Object
show all
Defined in:
lib/kubemq/transport/grpc_transport.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.

Low-level gRPC transport managing a single broker connection.

Handles channel creation, TLS/mTLS credential setup, keepalive options, interceptor wiring, lazy connection, and coordinated shutdown. Reconnection is delegated to ReconnectManager; connection lifecycle is tracked by ConnectionStateMachine.

See Also:

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config) ⇒ GrpcTransport

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 transport bound to the given configuration.

Connection is established lazily on the first call to #ensure_connected!.

Parameters:



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 34

def initialize(config)
  @config = config
  @state_machine = ConnectionStateMachine.new
  @buffer = MessageBuffer.new
  @stub = nil
  @channel = nil
  @mutex = Mutex.new
  @connect_cv = ConditionVariable.new
  @subscriptions = []
  @last_connect_attempt = nil
  @state_machine.on_disconnected { @connect_cv.broadcast }
  @reconnect_manager = ReconnectManager.new(
    policy: @config.reconnect_policy,
    state_machine: @state_machine,
    reconnect_proc: method(:reconnect!),
    on_reconnected: method(:on_reconnected!)
  )
end

Instance Attribute Details

#buffer ⇒ Object (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. 



27
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 27

attr_reader :state_machine, :buffer

#state_machine ⇒ ConnectionStateMachine (readonly)

Returns tracks connection lifecycle states.

Returns:



27
28
29
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 27

def state_machine
  @state_machine
end

Instance Method Details

#ack_all_queue_messages_rpc(proto) ⇒ Kubemq::AckAllQueueMessagesResponse

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.

Acknowledges all messages in a queue channel via unary RPC.

Parameters:

Returns:



149
150
151
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 149

def ack_all_queue_messages_rpc(proto)
  kubemq_client.ack_all_queue_messages(proto, deadline: default_deadline)
end

#close ⇒ 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.

Closes the transport and releases all resources.

Transitions the state machine to CLOSED, stops the reconnect manager, cancels active subscriptions, flushes the message buffer, and closes the gRPC channel. This method is idempotent.



160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 160

def close
  @mutex.synchronize do
    return if @state_machine.closed?

    begin
      @state_machine.transition!(ConnectionState::CLOSED, reason: 'client closed')
    rescue KubeMQ::Error
      return
    end
  end

  @reconnect_manager.stop
  cancel_subscriptions
  flush_buffer(timeout: 5)

  begin
    @channel&.close
  rescue StandardError
    # best-effort
  end

  @connect_cv.broadcast
end

#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 transport has been closed.

Returns:

  • (Boolean) —

    true if #close has been called



187
188
189
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 187

def closed?
  @state_machine.closed?
end

#ensure_connected! ⇒ 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.

Blocks until the transport reaches the READY state, connecting lazily if needed. Waits on in-progress connection attempts.

Raises:



59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 59

def ensure_connected!
  return if @state_machine.ready?

  @mutex.synchronize do
    loop do
      return if @state_machine.ready?
      raise ClientClosedError if @state_machine.closed?

      if [ConnectionState::CONNECTING, ConnectionState::RECONNECTING].include?(@state_machine.state)
        @connect_cv.wait(@mutex, 15)
        next
      end
      break
    end

    connect_unlocked!
  end
end

#kubemq_client ⇒ Kubemq::Kubemq::Stub

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 gRPC stub, connecting lazily if needed.

Returns:

Raises:



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

def kubemq_client
  ensure_connected!
  @stub
end

#on_disconnect! ⇒ 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.

Signals a connection loss and starts the reconnect manager.

Transitions to RECONNECTING and spawns the background reconnect loop. No-op if the transport is already closed.



239
240
241
242
243
244
245
246
247
248
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 239

def on_disconnect!
  return if @state_machine.closed?

  begin
    @state_machine.transition!(ConnectionState::RECONNECTING, reason: 'connection lost')
  rescue KubeMQ::Error
    return
  end
  @reconnect_manager.start
end

#ping ⇒ ServerInfo

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.

Sends a ping to the broker and returns server information.

Returns:

  • (ServerInfo) —

    broker host, version, and uptime

Raises:



92
93
94
95
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 92

def ping
  result = @stub.ping(::Kubemq::Empty.new, deadline: default_deadline)
  ServerInfo.from_proto(result)
end

#receive_queue_messages_rpc(proto) ⇒ Kubemq::ReceiveQueueMessagesResponse

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.

Receives queue messages via unary RPC (simple API).

Parameters:

Returns:



141
142
143
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 141

def receive_queue_messages_rpc(proto)
  kubemq_client.receive_queue_messages(proto, deadline: default_deadline)
end

#recreate_channel! ⇒ 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.

Tears down and rebuilds the gRPC channel and stub.

Used by the reconnect manager after a connection loss.

Raises:



213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 213

def recreate_channel!
  @mutex.synchronize do
    unless @state_machine.state == ConnectionState::RECONNECTING
      begin
        @state_machine.transition!(ConnectionState::RECONNECTING, reason: 'channel recreation')
      rescue KubeMQ::Error
        # allow from any state for recreation
      end
    end
    begin
      @channel&.close
    rescue StandardError
      # best-effort
    end
    @channel = nil
    @stub = nil
    connect_unlocked!
  end
end

#register_subscription(thread) ⇒ 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 subscription thread for lifecycle tracking.

Parameters:

  • thread (Thread) —

    the subscription background thread



195
196
197
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 195

def register_subscription(thread)
  @mutex.synchronize { @subscriptions << thread }
end

#send_event(proto) ⇒ Kubemq::Result

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.

Sends a pub/sub event via unary RPC.

Parameters:

Returns:



101
102
103
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 101

def send_event(proto)
  kubemq_client.send_event(proto, deadline: default_deadline)
end

#send_queue_message_rpc(proto) ⇒ Kubemq::SendQueueMessageResult

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.

Sends a single queue message via unary RPC (simple API).

Parameters:

Returns:



125
126
127
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 125

def send_queue_message_rpc(proto)
  kubemq_client.send_queue_message(proto, deadline: default_deadline)
end

#send_queue_messages_batch_rpc(proto) ⇒ Kubemq::QueueMessagesBatchResponse

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.

Sends a batch of queue messages via unary RPC (simple API).

Parameters:

Returns:



133
134
135
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 133

def send_queue_messages_batch_rpc(proto)
  kubemq_client.send_queue_messages_batch(proto, deadline: default_deadline)
end

#send_request(proto) ⇒ Kubemq::Response

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.

Sends a command or query request via unary RPC.

Parameters:

Returns:



109
110
111
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 109

def send_request(proto)
  kubemq_client.send_request(proto, deadline: default_deadline)
end

#send_response_rpc(proto) ⇒ 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.

Sends a command/query response back to the broker via unary RPC.

Parameters:



117
118
119
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 117

def send_response_rpc(proto)
  kubemq_client.send_response(proto, deadline: default_deadline)
end

#unregister_subscription(thread) ⇒ 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.

Removes a subscription thread from lifecycle tracking.

Parameters:

  • thread (Thread) —

    the subscription background thread to remove



203
204
205
 
# File 'lib/kubemq/transport/grpc_transport.rb', line 203

def unregister_subscription(thread)
  @mutex.synchronize { @subscriptions.delete(thread) }
end