Class: KubeMQ::CQClient

Inherits:
BaseClient show all
Defined in:
lib/kubemq/cq/client.rb

Overview

Client for KubeMQ commands and queries — synchronous request/reply RPC.

Commands are fire-and-confirm (no response body); queries return data and support server-side caching. Inherits connection management and channel CRUD from BaseClient.

Examples:

Send a command and handle the response

client = KubeMQ::CQClient.new(address: "localhost:50000")
cmd = KubeMQ::CQ::CommandMessage.new(
  channel: "commands.users",
  body: '{"action": "create"}',
  timeout: 5000
)
response = client.send_command(cmd)
puts "Executed: #{response.executed}"
client.close

See Also:

Instance Attribute Summary

Attributes inherited from BaseClient

#config, #transport

Instance Method Summary collapse

Methods inherited from BaseClient

#close, #closed?, #create_channel, #delete_channel, #initialize, #list_channels, #ping, #purge_queue_channel

Constructor Details

This class inherits a constructor from KubeMQ::BaseClient

Instance Method Details

#create_commands_channel(channel_name:) ⇒ Boolean

Creates a commands channel on the broker.

Parameters:

  • channel_name (String) —

    name for the new channel

Returns:

  • (Boolean) —

    true on success

Raises:

See Also:



389
390
391
 
# File 'lib/kubemq/cq/client.rb', line 389

def create_commands_channel(channel_name:)
  create_channel(channel_name: channel_name, channel_type: ChannelType::COMMANDS)
end

#create_queries_channel(channel_name:) ⇒ Boolean

Creates a queries channel on the broker.

Parameters:

  • channel_name (String) —

    name for the new channel

Returns:

  • (Boolean) —

    true on success

Raises:

See Also:



400
401
402
 
# File 'lib/kubemq/cq/client.rb', line 400

def create_queries_channel(channel_name:)
  create_channel(channel_name: channel_name, channel_type: ChannelType::QUERIES)
end

#delete_commands_channel(channel_name:) ⇒ Boolean

Deletes a commands channel from the broker.

Parameters:

  • channel_name (String) —

    name of the channel to delete

Returns:

  • (Boolean) —

    true on success

Raises:

See Also:



411
412
413
 
# File 'lib/kubemq/cq/client.rb', line 411

def delete_commands_channel(channel_name:)
  delete_channel(channel_name: channel_name, channel_type: ChannelType::COMMANDS)
end

#delete_queries_channel(channel_name:) ⇒ Boolean

Deletes a queries channel from the broker.

Parameters:

  • channel_name (String) —

    name of the channel to delete

Returns:

  • (Boolean) —

    true on success

Raises:

See Also:



422
423
424
 
# File 'lib/kubemq/cq/client.rb', line 422

def delete_queries_channel(channel_name:)
  delete_channel(channel_name: channel_name, channel_type: ChannelType::QUERIES)
end

#list_commands_channels(search: nil) ⇒ Array<ChannelInfo>

Lists commands channels, with optional name filtering.

Parameters:

  • search (String, nil) (defaults to: nil) —

    substring filter for channel names

Returns:

  • (Array<ChannelInfo>) —

    matching channels with metadata

Raises:

See Also:



432
433
434
 
# File 'lib/kubemq/cq/client.rb', line 432

def list_commands_channels(search: nil)
  list_channels(channel_type: ChannelType::COMMANDS, search: search)
end

#list_queries_channels(search: nil) ⇒ Array<ChannelInfo>

Lists queries channels, with optional name filtering.

Parameters:

  • search (String, nil) (defaults to: nil) —

    substring filter for channel names

Returns:

  • (Array<ChannelInfo>) —

    matching channels with metadata

Raises:

See Also:



442
443
444
 
# File 'lib/kubemq/cq/client.rb', line 442

def list_queries_channels(search: nil)
  list_channels(channel_type: ChannelType::QUERIES, search: search)
end

#send_command(message) ⇒ CQ::CommandResponse

Note:

Timeout is in milliseconds.

Sends a command to a responder and waits for confirmation.

Examples:

cmd = KubeMQ::CQ::CommandMessage.new(
  channel: "commands.orders",
  body: '{"action": "cancel", "id": 42}',
  timeout: 10_000
)
response = client.send_command(cmd)
puts "Success" if response.executed

Parameters:

Returns:

Raises:

See Also:



54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/kubemq/cq/client.rb', line 54

def send_command(message)
  Validator.validate_channel!(message.channel, allow_wildcards: false)
  Validator.validate_content!(message.metadata, message.body)
  Validator.validate_timeout!(message.timeout)
  ensure_connected!

  proto = Transport::Converter.request_to_proto(message, @config.client_id, RequestType::COMMAND)
  response = @transport.kubemq_client.send_request(proto)
  hash = Transport::Converter.proto_to_command_response(response)
  CQ::CommandResponse.new(**hash)
end

#send_query(message) ⇒ CQ::QueryResponse

Note:

Timeout is in milliseconds.

Sends a query to a responder and waits for a data response.

Queries support server-side caching via cache_key and cache_ttl on the KubeMQ::CQ::QueryMessage.

Examples:

query = KubeMQ::CQ::QueryMessage.new(
  channel: "queries.users",
  body: '{"user_id": 42}',
  timeout: 10_000,
  cache_key: "user-42",
  cache_ttl: 60_000
)
response = client.send_query(query)
puts "Data: #{response.body} (cache_hit=#{response.cache_hit})"

Parameters:

Returns:

Raises:

See Also:



245
246
247
248
249
250
251
252
253
254
255
256
 
# File 'lib/kubemq/cq/client.rb', line 245

def send_query(message)
  Validator.validate_channel!(message.channel, allow_wildcards: false)
  Validator.validate_content!(message.metadata, message.body)
  Validator.validate_timeout!(message.timeout)
  Validator.validate_cache!(message.cache_key, message.cache_ttl) if message.cache_key
  ensure_connected!

  proto = Transport::Converter.request_to_proto(message, @config.client_id, RequestType::QUERY)
  response = @transport.kubemq_client.send_request(proto)
  hash = Transport::Converter.proto_to_query_response(response)
  CQ::QueryResponse.new(**hash)
end

#send_response(response) ⇒ void

This method returns an undefined value.

Sends a response to a received command or query.

Call this from within a #subscribe_to_commands or #subscribe_to_queries block to reply to the sender.

Parameters:

Raises:

See Also:



204
205
206
207
208
209
210
211
 
# File 'lib/kubemq/cq/client.rb', line 204

def send_response(response)
  Validator.validate_response!(response.request_id, response.reply_channel)
  ensure_connected!

  proto = Transport::Converter.response_message_to_proto(response, @config.client_id)
  @transport.kubemq_client.send_response(proto)
  nil
end

#subscribe_to_commands(subscription, cancellation_token: nil, on_error: nil) {|command| ... } ⇒ Subscription

Note:

Wildcard channels are NOT supported for commands.

Subscribes to incoming commands on a channel. Runs on a background thread; incoming commands are delivered to the provided block.

The block should process the command and send a response via #send_response. The subscription auto-reconnects with exponential backoff on transient failures.

rubocop:disable Metrics/AbcSize, Metrics/MethodLength -- subscription with auto-reconnect loop

Examples:

token = KubeMQ::CancellationToken.new
sub = KubeMQ::CQ::CommandsSubscription.new(channel: "commands.orders")
client.subscribe_to_commands(sub, cancellation_token: token) do |cmd|
  # process and respond
  client.send_response(
    KubeMQ::CQ::CommandResponseMessage.new(
      request_id: cmd.id,
      reply_channel: cmd.reply_channel,
      executed: true
    )
  )
end

Parameters:

  • subscription (CQ::CommandsSubscription) —

    channel and group config

  • cancellation_token (CancellationToken, nil) (defaults to: nil) —

    token for cooperative cancellation (auto-created if nil)

  • on_error (Proc, nil) (defaults to: nil) —

    callback receiving Error on stream or callback failures

Yields:

  • (command) —

    called for each received command on a background thread

Yield Parameters:

Returns:

  • (Subscription) —

    handle to check status, cancel, or join

Raises:

See Also:



107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
 
# File 'lib/kubemq/cq/client.rb', line 107

def subscribe_to_commands(subscription, cancellation_token: nil, on_error: nil, &block)
  raise ArgumentError, 'Block required for subscribe_to_commands' unless block

  Validator.validate_channel!(subscription.channel, allow_wildcards: false)
  ensure_connected!

  cancellation_token ||= CancellationToken.new
  nil

  thread = Thread.new do
    @transport.register_subscription(Thread.current)
    Thread.current[:cancellation_token] = cancellation_token
    reconnect_attempts = 0
    begin
      loop do
        break if cancellation_token.cancelled?

        begin
          @transport.ensure_connected!
          proto_sub = Transport::Converter.subscribe_to_proto(subscription, @config.client_id)
          stream = @transport.kubemq_client.subscribe_to_requests(proto_sub)
          Thread.current[:grpc_call] = stream
          reconnect_attempts = 0
          stream.each do |request|
            break if cancellation_token.cancelled?

            received = CQ::CommandReceived.new(
              id: request.RequestID,
              channel: request.Channel,
              metadata: request.Metadata,
              body: request.Body,
              reply_channel: request.ReplyChannel,
              tags: request.Tags.to_h,
              timeout: request.Timeout,
              client_id: request.ClientID
            )
            begin
              block.call(received)
            rescue StandardError => e
              begin
                on_error&.call(Error.new("Callback error: #{e.message}", code: ErrorCode::CALLBACK_ERROR))
              rescue StandardError => nested
                Kernel.warn("[kubemq] on_error callback raised: #{nested.message}")
              end
            end
          end
        rescue CancellationError
          break
        rescue GRPC::BadStatus, StandardError => e
          @transport.on_disconnect! if e.is_a?(GRPC::Unavailable) || e.is_a?(GRPC::DeadlineExceeded)
          reconnect_attempts += 1
          begin
            if e.is_a?(GRPC::BadStatus)
              on_error&.call(ErrorMapper.map_grpc_error(e, operation: 'subscribe_commands'))
            else
              on_error&.call(Error.new(e.message, code: ErrorCode::STREAM_BROKEN))
            end
          rescue StandardError => cb_err
            Kernel.warn("[kubemq] on_error callback raised: #{cb_err.message}")
          end
          delay = [@config.reconnect_policy.base_interval *
            (@config.reconnect_policy.multiplier**(reconnect_attempts - 1)),
                   @config.reconnect_policy.max_delay].min
          sleep(delay) unless cancellation_token.cancelled?
        end
      end
    rescue StandardError => e
      Thread.current[:kubemq_subscription]&.mark_error(e)
    ensure
      begin; Thread.current[:grpc_call]&.cancel; rescue StandardError; end
      Thread.current[:kubemq_subscription]&.mark_closed
      @transport.unregister_subscription(Thread.current)
    end
  end

  sub_wrapper = KubeMQ::Subscription.new(thread: thread, cancellation_token: cancellation_token)
  thread[:kubemq_subscription] = sub_wrapper
  sub_wrapper
end

#subscribe_to_queries(subscription, cancellation_token: nil, on_error: nil) {|query| ... } ⇒ Subscription

Note:

Wildcard channels are NOT supported for queries.

Subscribes to incoming queries on a channel. Runs on a background thread; incoming queries are delivered to the provided block.

The block should process the query and send a response via #send_response. The subscription auto-reconnects with exponential backoff on transient failures.

rubocop:disable Metrics/AbcSize, Metrics/MethodLength -- subscription with auto-reconnect loop

Examples:

token = KubeMQ::CancellationToken.new
sub = KubeMQ::CQ::QueriesSubscription.new(channel: "queries.users")
client.subscribe_to_queries(sub, cancellation_token: token) do |query|
  client.send_response(
    KubeMQ::CQ::QueryResponseMessage.new(
      request_id: query.id,
      reply_channel: query.reply_channel,
      body: '{"name": "Alice"}',
      executed: true
    )
  )
end

Parameters:

  • subscription (CQ::QueriesSubscription) —

    channel and group config

  • cancellation_token (CancellationToken, nil) (defaults to: nil) —

    token for cooperative cancellation (auto-created if nil)

  • on_error (Proc, nil) (defaults to: nil) —

    callback receiving Error on stream or callback failures

Yields:

  • (query) —

    called for each received query on a background thread

Yield Parameters:

Returns:

  • (Subscription) —

    handle to check status, cancel, or join

Raises:

See Also:



299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
 
# File 'lib/kubemq/cq/client.rb', line 299

def subscribe_to_queries(subscription, cancellation_token: nil, on_error: nil, &block)
  raise ArgumentError, 'Block required for subscribe_to_queries' unless block

  Validator.validate_channel!(subscription.channel, allow_wildcards: false)
  ensure_connected!

  cancellation_token ||= CancellationToken.new
  nil

  thread = Thread.new do
    @transport.register_subscription(Thread.current)
    Thread.current[:cancellation_token] = cancellation_token
    reconnect_attempts = 0
    begin
      loop do
        break if cancellation_token.cancelled?

        begin
          @transport.ensure_connected!
          proto_sub = Transport::Converter.subscribe_to_proto(subscription, @config.client_id)
          stream = @transport.kubemq_client.subscribe_to_requests(proto_sub)
          Thread.current[:grpc_call] = stream
          reconnect_attempts = 0
          stream.each do |request|
            break if cancellation_token.cancelled?

            received = CQ::QueryReceived.new(
              id: request.RequestID,
              channel: request.Channel,
              metadata: request.Metadata,
              body: request.Body,
              reply_channel: request.ReplyChannel,
              tags: request.Tags.to_h,
              timeout: request.Timeout,
              client_id: request.ClientID
            )
            begin
              block.call(received)
            rescue StandardError => e
              begin
                on_error&.call(Error.new("Callback error: #{e.message}", code: ErrorCode::CALLBACK_ERROR))
              rescue StandardError => nested
                Kernel.warn("[kubemq] on_error callback raised: #{nested.message}")
              end
            end
          end
        rescue CancellationError
          break
        rescue GRPC::BadStatus, StandardError => e
          @transport.on_disconnect! if e.is_a?(GRPC::Unavailable) || e.is_a?(GRPC::DeadlineExceeded)
          reconnect_attempts += 1
          begin
            if e.is_a?(GRPC::BadStatus)
              on_error&.call(ErrorMapper.map_grpc_error(e, operation: 'subscribe_queries'))
            else
              on_error&.call(Error.new(e.message, code: ErrorCode::STREAM_BROKEN))
            end
          rescue StandardError => cb_err
            Kernel.warn("[kubemq] on_error callback raised: #{cb_err.message}")
          end
          delay = [@config.reconnect_policy.base_interval *
            (@config.reconnect_policy.multiplier**(reconnect_attempts - 1)),
                   @config.reconnect_policy.max_delay].min
          sleep(delay) unless cancellation_token.cancelled?
        end
      end
    rescue StandardError => e
      Thread.current[:kubemq_subscription]&.mark_error(e)
    ensure
      begin; Thread.current[:grpc_call]&.cancel; rescue StandardError; end
      Thread.current[:kubemq_subscription]&.mark_closed
      @transport.unregister_subscription(Thread.current)
    end
  end

  sub_wrapper = KubeMQ::Subscription.new(thread: thread, cancellation_token: cancellation_token)
  thread[:kubemq_subscription] = sub_wrapper
  sub_wrapper
end