Class: KubeMQ::PubSub::EventSender

Inherits:
Object
  • Object
show all
Defined in:
lib/kubemq/pubsub/event_sender.rb

Overview

Note:

The stream is fire-and-forget — #publish returns immediately without waiting for broker confirmation.

Streaming sender for fire-and-forget events over a persistent gRPC stream.

Created via KubeMQ::PubSubClient#create_events_sender. Keeps a bidirectional stream open for high-throughput publishing. Call #publish to enqueue events and #close when finished.

See Also:

Instance Method Summary collapse

Constructor Details

#initialize(transport:, client_id:, on_error: nil) ⇒ EventSender

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 a new instance of EventSender.

Parameters:

  • transport (Transport::GrpcTransport) —

    the gRPC transport

  • client_id (String) —

    client identifier for the stream

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

    callback invoked with error results from the stream



22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/kubemq/pubsub/event_sender.rb', line 22

def initialize(transport:, client_id:, on_error: nil)
  @transport = transport
  @client_id = client_id
  @on_error = on_error
  @request_queue = Queue.new
  @mutex = Mutex.new
  @closed = false
  @stream_alive = true
  @stream_error = nil
  start_stream!
end

Instance Method Details

#close ⇒ void

This method returns an undefined value.

Closes the sender and its underlying gRPC stream.

This method is idempotent — calling it multiple times is safe.



68
69
70
71
72
73
74
75
76
77
 
# File 'lib/kubemq/pubsub/event_sender.rb', line 68

def close
  @mutex.synchronize do
    return if @closed

    @closed = true
  end
  wake_all_pending!
  @request_queue.push(:close)
  @stream_thread&.join(5)
end

#publish(message) ⇒ nil

Publishes an event over the persistent stream.

The event is enqueued and sent asynchronously. This method returns immediately without waiting for broker acknowledgement.

Parameters:

Returns:

  • (nil)

Raises:

See Also:



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/kubemq/pubsub/event_sender.rb', line 47

def publish(message)
  raise ClientClosedError if @mutex.synchronize { @closed }
  unless @mutex.synchronize { @stream_alive }
    raise StreamBrokenError.new(
      'Event sender stream is broken; create a new sender',
      cause: @mutex.synchronize { @stream_error }
    )
  end

  Validator.validate_channel!(message.channel, allow_wildcards: false)
  Validator.validate_content!(message.metadata, message.body)
  proto = Transport::Converter.event_to_proto(message, @client_id, store: false)
  @request_queue.push(proto)
  nil
end