Class: KubeMQ::PubSub::EventStoreSender

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

Overview

Note:

This class is thread-safe. Multiple threads may call #publish concurrently; each waits independently for its confirmation.

Streaming sender for durable events store over a persistent gRPC stream.

Created via KubeMQ::PubSubClient#create_events_store_sender. Unlike EventSender, each #publish call blocks until the broker confirms persistence, returning an EventStoreResult.

See Also:

Constant Summary collapse

SEND_TIMEOUT =

Maximum seconds to wait for broker confirmation per publish.

10

Instance Method Summary collapse

Constructor Details

#initialize(transport:, client_id:) ⇒ EventStoreSender

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 EventStoreSender.

Parameters:



27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/kubemq/pubsub/event_store_sender.rb', line 27

def initialize(transport:, client_id:)
  @transport = transport
  @client_id = client_id
  @request_queue = Queue.new
  @pending = {}
  @pending_mutex = Mutex.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.

Wakes any threads blocked in #publish and shuts down the stream. This method is idempotent — calling it multiple times is safe.



107
108
109
110
111
112
113
114
115
116
 
# File 'lib/kubemq/pubsub/event_store_sender.rb', line 107

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) ⇒ EventStoreResult

Publishes a durable event and waits for broker confirmation.

Blocks until the broker confirms persistence or the SEND_TIMEOUT elapses. Returns an KubeMQ::PubSub::EventStoreResult on success.

Parameters:

Returns:

Raises:

See Also:



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/kubemq/pubsub/event_store_sender.rb', line 55

def publish(message)
  raise ClientClosedError if @mutex.synchronize { @closed }
  unless @mutex.synchronize { @stream_alive }
    raise StreamBrokenError.new(
      'Event store 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: true)
  event_id = proto.EventID

  waiter = { mutex: Mutex.new, cv: ConditionVariable.new, result: nil }
  @pending_mutex.synchronize { @pending[event_id] = waiter }
  @request_queue.push(proto)

  waiter[:mutex].synchronize do
    deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + SEND_TIMEOUT
    while waiter[:result].nil?
      remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
      break if remaining <= 0

      waiter[:cv].wait(waiter[:mutex], remaining)
    end
  end
  @pending_mutex.synchronize { @pending.delete(event_id) }

  unless @mutex.synchronize { @stream_alive }
    raise StreamBrokenError.new(
      'Event store sender stream is broken; create a new sender',
      cause: @mutex.synchronize { @stream_error }
    )
  end

  result = waiter[:result]
  raise TimeoutError, "EventStore send timed out for event #{event_id}" unless result

  EventStoreResult.new(
    id: result.EventID,
    sent: result.Sent,
    error: result.Error.empty? ? nil : result.Error
  )
end