Class: KubeMQ::Queues::UpstreamSender

Inherits:
Object
  • Object
show all
Defined in:
lib/kubemq/queues/upstream_sender.rb

Overview

Note:

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

Streaming sender for queue messages over a persistent gRPC upstream.

Created via KubeMQ::QueuesClient#create_upstream_sender. Keeps a bidirectional stream open for high-throughput queue publishing. Each #publish call blocks until the broker confirms receipt.

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:) ⇒ UpstreamSender

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

Parameters:



26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/kubemq/queues/upstream_sender.rb', line 26

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.



123
124
125
126
127
128
129
130
131
132
 
# File 'lib/kubemq/queues/upstream_sender.rb', line 123

def close
  @mutex.synchronize do
    return if @closed

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

#publish(messages) ⇒ Array<QueueSendResult>

Publishes one or more queue messages and waits for broker confirmation.

Blocks until the broker confirms receipt or SEND_TIMEOUT elapses. Accepts a single QueueMessage or an array for batch sending.

rubocop:disable Metrics/AbcSize, Metrics/MethodLength -- single enqueue/wait path for stream send

Parameters:

Returns:

Raises:

See Also:



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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/kubemq/queues/upstream_sender.rb', line 56

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

  messages = [messages] unless messages.is_a?(Array)

  request_id = SecureRandom.uuid
  proto_messages = messages.map do |msg|
    Validator.validate_channel!(msg.channel, allow_wildcards: false)
    Transport::Converter.queue_message_to_proto(msg, @client_id)
  end

  proto_request = ::Kubemq::QueuesUpstreamRequest.new(
    RequestID: request_id,
    Messages: proto_messages
  )

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

  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(request_id) }

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

  response = waiter[:result]
  raise TimeoutError, 'Queue upstream send timed out' unless response

  raise MessageError.new(response.Error, operation: 'queue_send') if response.IsError && !response.Error.empty?

  response.Results.map do |r|
    QueueSendResult.new(
      id: r.MessageID,
      sent_at: r.SentAt,
      expiration_at: r.ExpirationAt,
      delayed_to: r.DelayedTo,
      error: r.IsError ? r.Error : nil
    )
  end
end