Class: KubeMQ::PubSubClient

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

Overview

Client for KubeMQ pub/sub messaging — events (fire-and-forget) and events store (durable with replay).

Inherits connection management and channel CRUD from BaseClient. Subscription methods run on background threads and accept a block for incoming messages.

Examples:

Send and subscribe to events

client = KubeMQ::PubSubClient.new(address: "localhost:50000")

token = KubeMQ::CancellationToken.new
sub = KubeMQ::PubSub::EventsSubscription.new(channel: "notifications")
client.subscribe_to_events(sub, cancellation_token: token) do |event|
  puts "Received: #{event.body}"
end

client.send_event(
  KubeMQ::PubSub::EventMessage.new(channel: "notifications", body: "hello")
)
sleep 1
token.cancel
client.close

See Also:

Instance Attribute Summary

Attributes inherited from BaseClient

#config, #transport

Instance Method Summary collapse

Methods inherited from BaseClient

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

Constructor Details

#initialize(**kwargs) ⇒ PubSubClient

Returns a new instance of PubSubClient.



33
34
35
36
37
 
# File 'lib/kubemq/pubsub/client.rb', line 33

def initialize(**kwargs)
  super
  @senders = []
  @senders_mutex = Mutex.new
end

Instance Method Details

#close ⇒ void

This method returns an undefined value.

Closes the client, all open senders, and releases resources.

Closes any KubeMQ::PubSub::EventSender or KubeMQ::PubSub::EventStoreSender instances created through this client before closing the transport. This method is idempotent.



389
390
391
392
393
394
395
396
397
398
399
 
# File 'lib/kubemq/pubsub/client.rb', line 389

def close
  senders_mutex.synchronize do
    senders_list.each do |s|
      s.close
    rescue StandardError
      nil
    end
    senders_list.clear
  end
  super
end

#create_events_channel(channel_name:) ⇒ Boolean

Creates an events channel on the broker.

Parameters:

  • channel_name (String) —

    name for the new channel

Returns:

  • (Boolean) —

    true on success

Raises:

See Also:



408
409
410
 
# File 'lib/kubemq/pubsub/client.rb', line 408

def create_events_channel(channel_name:)
  create_channel(channel_name: channel_name, channel_type: ChannelType::EVENTS)
end

#create_events_sender(on_error: nil) ⇒ PubSub::EventSender

Creates a streaming event sender for high-throughput publishing.

The sender holds a persistent gRPC stream open. Close it explicitly when finished, or it will be closed when the client is closed.

Parameters:

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

    callback receiving Error on stream failures

Returns:

Raises:

See Also:



86
87
88
89
90
91
 
# File 'lib/kubemq/pubsub/client.rb', line 86

def create_events_sender(on_error: nil)
  ensure_connected!
  sender = PubSub::EventSender.new(transport: @transport, client_id: @config.client_id, on_error: on_error)
  senders_mutex.synchronize { senders_list << sender }
  sender
end

#create_events_store_channel(channel_name:) ⇒ Boolean

Creates an events store channel on the broker.

Parameters:

  • channel_name (String) —

    name for the new channel

Returns:

  • (Boolean) —

    true on success

Raises:

See Also:



419
420
421
 
# File 'lib/kubemq/pubsub/client.rb', line 419

def create_events_store_channel(channel_name:)
  create_channel(channel_name: channel_name, channel_type: ChannelType::EVENTS_STORE)
end

#create_events_store_sender ⇒ PubSub::EventStoreSender

Creates a streaming events store sender for high-throughput publishing with synchronous per-message confirmation.

Close the sender explicitly when finished, or it will be closed when the client is closed.

Returns:

Raises:

See Also:



252
253
254
255
256
257
 
# File 'lib/kubemq/pubsub/client.rb', line 252

def create_events_store_sender
  ensure_connected!
  sender = PubSub::EventStoreSender.new(transport: @transport, client_id: @config.client_id)
  senders_mutex.synchronize { senders_list << sender }
  sender
end

#delete_events_channel(channel_name:) ⇒ Boolean

Deletes an events channel from the broker.

Parameters:

  • channel_name (String) —

    name of the channel to delete

Returns:

  • (Boolean) —

    true on success

Raises:

See Also:



430
431
432
 
# File 'lib/kubemq/pubsub/client.rb', line 430

def delete_events_channel(channel_name:)
  delete_channel(channel_name: channel_name, channel_type: ChannelType::EVENTS)
end

#delete_events_store_channel(channel_name:) ⇒ Boolean

Deletes an events store channel from the broker.

Parameters:

  • channel_name (String) —

    name of the channel to delete

Returns:

  • (Boolean) —

    true on success

Raises:

See Also:



441
442
443
 
# File 'lib/kubemq/pubsub/client.rb', line 441

def delete_events_store_channel(channel_name:)
  delete_channel(channel_name: channel_name, channel_type: ChannelType::EVENTS_STORE)
end

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

Lists events 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:



451
452
453
 
# File 'lib/kubemq/pubsub/client.rb', line 451

def list_events_channels(search: nil)
  list_channels(channel_type: ChannelType::EVENTS, search: search)
end

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

Lists events store 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:



461
462
463
 
# File 'lib/kubemq/pubsub/client.rb', line 461

def list_events_store_channels(search: nil)
  list_channels(channel_type: ChannelType::EVENTS_STORE, search: search)
end

#send_event(message) ⇒ PubSub::EventSendResult

Sends a single event to a channel. Fire-and-forget semantics — the server does not guarantee persistence.

Examples:

msg = KubeMQ::PubSub::EventMessage.new(
  channel: "events.orders",
  metadata: "order-created",
  body: '{"id": 42}',
  tags: { "env" => "production" }
)
result = client.send_event(msg)
puts "Sent: #{result.id}" if result.sent

Parameters:

Returns:

Raises:

See Also:



62
63
64
65
66
67
68
69
70
71
 
# File 'lib/kubemq/pubsub/client.rb', line 62

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

  proto = Transport::Converter.event_to_proto(message, @config.client_id, store: false)
  result = @transport.kubemq_client.send_event(proto)
  hash = Transport::Converter.proto_to_event_result(result)
  PubSub::EventSendResult.new(**hash)
end

#send_event_store(message) ⇒ PubSub::EventStoreResult

Sends a single event to a durable events store channel.

Unlike #send_event, the broker persists events store messages and supports replay via #subscribe_to_events_store.

Examples:

msg = KubeMQ::PubSub::EventStoreMessage.new(
  channel: "orders.created",
  body: '{"id": 42}'
)
result = client.send_event_store(msg)
puts "Stored: #{result.id}" if result.sent

Parameters:

Returns:

Raises:

See Also:



229
230
231
232
233
234
235
236
237
238
 
# File 'lib/kubemq/pubsub/client.rb', line 229

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

  proto = Transport::Converter.event_to_proto(message, @config.client_id, store: true)
  result = @transport.kubemq_client.send_event(proto)
  hash = Transport::Converter.proto_to_event_result(result)
  PubSub::EventStoreResult.new(**hash)
end

#subscribe_to_events(subscription, cancellation_token: nil, on_error: nil) {|event| ... } ⇒ Subscription

Note:

Wildcard channels (containing * or >) are supported for events.

Subscribes to real-time events on a channel. Runs on a background thread; incoming events are delivered to the provided block.

The subscription automatically reconnects with exponential backoff on transient failures. Use the returned Subscription or the cancellation_token to stop.

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

Examples:

token = KubeMQ::CancellationToken.new
sub_config = KubeMQ::PubSub::EventsSubscription.new(
  channel: "events.>", group: "workers"
)
subscription = client.subscribe_to_events(sub_config, cancellation_token: token) do |event|
  puts "#{event.channel}: #{event.body}"
end
# later...
token.cancel
subscription.wait(5)

Parameters:

  • subscription (PubSub::EventsSubscription) —

    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:

  • (event) —

    called for each received event on a background thread

Yield Parameters:

Returns:

  • (Subscription) —

    handle to check status, cancel, or join

Raises:

See Also:



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
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
 
# File 'lib/kubemq/pubsub/client.rb', line 133

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

  Validator.validate_channel!(subscription.channel, allow_wildcards: true)
  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_events(proto_sub)
          Thread.current[:grpc_call] = stream
          reconnect_attempts = 0
          stream.each do |event_receive|
            break if cancellation_token.cancelled?

            hash = Transport::Converter.proto_to_event_received(event_receive)
            received = PubSub::EventReceived.new(**hash)
            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_events'))
            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_events_store(subscription, cancellation_token: nil, on_error: nil) {|event| ... } ⇒ Subscription

Note:

On reconnect the subscription automatically resumes from the last received sequence number, overriding the original start position.

Subscribes to durable events store messages on a channel. Runs on a background thread; incoming events are delivered to the provided block.

The subscription automatically reconnects with exponential backoff on transient failures. On reconnect, playback resumes from the last received sequence number to avoid duplicates.

rubocop:disable Metrics/AbcSize, Metrics/MethodLength -- subscription with auto-reconnect + sequence tracking

Examples:

Subscribe from the beginning

sub = KubeMQ::PubSub::EventsStoreSubscription.new(
  channel: "orders",
  start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_FIRST
)
token = KubeMQ::CancellationToken.new
client.subscribe_to_events_store(sub, cancellation_token: token) do |event|
  puts "seq=#{event.sequence}: #{event.body}"
end

Parameters:

  • subscription (PubSub::EventsStoreSubscription) —

    channel, group, and start position 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:

  • (event) —

    called for each received event on a background thread

Yield Parameters:

Returns:

  • (Subscription) —

    handle to check status, cancel, or join

Raises:

  • (ArgumentError) —

    if no block is given

  • (ValidationError) —

    if the subscription channel or start position is invalid

  • (ClientClosedError) —

    if the client has been closed

See Also:



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
378
379
 
# File 'lib/kubemq/pubsub/client.rb', line 300

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

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

  cancellation_token ||= CancellationToken.new
  nil

  thread = Thread.new do
    @transport.register_subscription(Thread.current)
    Thread.current[:cancellation_token] = cancellation_token
    last_sequence = 0
    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)
          if last_sequence.positive?
            proto_sub.EventsStoreTypeData = PubSub::EventStoreStartPosition::START_AT_SEQUENCE
            proto_sub.EventsStoreTypeValue = last_sequence + 1
          end
          stream = @transport.kubemq_client.subscribe_to_events(proto_sub)
          Thread.current[:grpc_call] = stream
          reconnect_attempts = 0
          stream.each do |event_receive|
            break if cancellation_token.cancelled?

            hash = Transport::Converter.proto_to_event_received(event_receive)
            received = PubSub::EventStoreReceived.new(**hash)
            last_sequence = received.sequence if received.sequence.positive?
            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_events_store'))
            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