Class: KubeMQ::Queues::DownstreamReceiver

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

Overview

Note:

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

Streaming receiver for queue messages over a persistent gRPC downstream.

Created via KubeMQ::QueuesClient#create_downstream_receiver. Keeps a bidirectional stream open for polling with transactional acknowledgement. Call #poll to receive messages and use the returned QueuePollResponse for batch or per-message actions.

See Also:

Constant Summary collapse

RESPONSE_TIMEOUT =

Maximum seconds to wait for a downstream response.

60

Instance Method Summary collapse

Constructor Details

#initialize(transport:, client_id:) ⇒ DownstreamReceiver

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

Parameters:



27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/kubemq/queues/downstream_receiver.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 receiver and its underlying gRPC stream.

Sends a close-by-client request to the broker and shuts down the stream. This method is idempotent — calling it multiple times is safe.



118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/kubemq/queues/downstream_receiver.rb', line 118

def close
  @mutex.synchronize do
    return if @closed

    @closed = true
  end
  wake_all_pending!

  begin
    close_request = ::Kubemq::QueuesDownstreamRequest.new(
      RequestID: SecureRandom.uuid,
      ClientID: @client_id,
      RequestTypeData: :CloseByClient
    )
    @request_queue.push(close_request)
    sleep(0.5)
  rescue StandardError
    # best-effort
  end

  @request_queue.push(:close)
  @stream_thread&.join(5)
end

#poll(request) ⇒ QueuePollResponse

Polls the queue channel for messages.

Blocks until messages arrive or the request's wait_timeout elapses. Returns a QueuePollResponse with received messages and transaction action methods.

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
 
# File 'lib/kubemq/queues/downstream_receiver.rb', line 56

def poll(request)
  raise ClientClosedError if @mutex.synchronize { @closed }
  unless @mutex.synchronize { @stream_alive }
    raise StreamBrokenError.new(
      'Queue downstream stream is broken',
      cause: @mutex.synchronize { @stream_error }
    )
  end

  Validator.validate_channel!(request.channel, allow_wildcards: false)
  Validator.validate_queue_poll!(request.max_items, request.wait_timeout)

  request_id = SecureRandom.uuid
  proto_request = ::Kubemq::QueuesDownstreamRequest.new(
    RequestID: request_id,
    ClientID: @client_id,
    RequestTypeData: :Get,
    Channel: request.channel,
    MaxItems: request.max_items,
    WaitTimeout: request.wait_timeout * 1000,
    AutoAck: request.auto_ack
  )

  timeout = [request.wait_timeout + 5, RESPONSE_TIMEOUT].min
  response = send_and_wait(request_id, proto_request, timeout)
  raise TimeoutError, 'Queue poll timed out' unless response

  build_poll_response(response)
end

#send_action(transaction_id:, type:, channel: nil, sequence_range: []) ⇒ Object?

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.

Sends a transaction action over the downstream stream.

Parameters:

  • transaction_id (String) —

    the transaction to act upon

  • type (Symbol) —

    action type (e.g., :AckAll, :NAckAll, :ReQueueAll)

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

    destination channel for requeue actions

  • sequence_range (Array<Integer>) (defaults to: []) —

    message sequences for range actions

Returns:

  • (Object, nil) —

    the broker response, if any

Raises:



96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/kubemq/queues/downstream_receiver.rb', line 96

def send_action(transaction_id:, type:, channel: nil, sequence_range: [])
  raise ClientClosedError, 'Downstream receiver is closed' if @mutex.synchronize { @closed }

  request_id = SecureRandom.uuid
  proto_request = ::Kubemq::QueuesDownstreamRequest.new(
    RequestID: request_id,
    ClientID: @client_id,
    RequestTypeData: type,
    RefTransactionId: transaction_id,
    ReQueueChannel: channel || '',
    SequenceRange: sequence_range
  )

  send_and_wait(request_id, proto_request, 10)
end