Class: AceMQ::AMQP::Patterns::Requester

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/patterns/request_reply.rb

Overview

Sends a request and waits for the reply.

prices = Patterns::Requester.new(mq, to: "price.requests")
quote = prices.call({ "sku" => "X-1" })
prices.close

One requester is meant to be kept and reused. It holds a queue and a consumer, and building one per request means a queue per request.

Defined Under Namespace

Classes: Waiter

Constant Summary collapse

DEFAULT_TIMEOUT =

How long to wait for an answer when nothing else is said.

30.0

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection, to:, exchange: "", reply_to: nil, timeout: DEFAULT_TIMEOUT) ⇒ Requester

Returns a new instance of Requester.

Parameters:

  • connection (Connection)
  • to (String) —

    the routing key requests go out on

  • exchange (String) (defaults to: "") —

    empty for the default exchange

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

    the queue replies come back on. One is generated when this is not given: exclusive, transient and auto-deleting, belonging to this process and going away with it, because a reply queue that outlived its requester would collect answers nobody is waiting for. Name one only when replies have to survive a restart.

  • timeout (Numeric) (defaults to: DEFAULT_TIMEOUT) —

    seconds to wait for a reply



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/acemq/amqp/patterns/request_reply.rb', line 94

def initialize(connection, to:, exchange: "", reply_to: nil, timeout: DEFAULT_TIMEOUT)
  @connection = connection
  @to = to
  @exchange = exchange
  @timeout = timeout.to_f
  @generated = reply_to.nil? || reply_to.to_s.empty?
  @reply_queue = @generated ? "acemq-reply-#{SecureRandom.uuid}" : reply_to.to_s
  @waiting = {}
  @lock = Mutex.new
  declare_reply_queue
  # Replies are taken as raw bytes and decoded by the caller in {#call},
  # as Java, .NET and Python do. Decoded here, a reply this codec cannot
  # read was parked by the consumer and the caller waited out its whole
  # timeout to be told nothing came back; decoded in {#call}, the caller
  # gets the codec's error at once.
  #
  # No retry policy and no dead-letter or parking queue, whatever the
  # connection's policy is. This handler cannot fail — it hands bytes to
  # a waiter and accepts — so those could only add durable queues to an
  # ephemeral reply queue, and they would outlive it.
  @consumer = connection.consume(@reply_queue, codec: BytesCodec.new,
                                               retry_policy: RetryPolicy.none,
                                               dead_letters: false) do |m|
    deliver(m)
    Ack.accept
  end
end

Instance Attribute Details

#reply_queue ⇒ Object (readonly)

Returns the value of attribute reply_queue.



82
83
84
 
# File 'lib/acemq/amqp/patterns/request_reply.rb', line 82

def reply_queue
  @reply_queue
end

#timeout ⇒ Object (readonly)

Returns the value of attribute timeout.



82
83
84
 
# File 'lib/acemq/amqp/patterns/request_reply.rb', line 82

def timeout
  @timeout
end

Instance Method Details

#call(request, **fields) ⇒ Object

Sends a request and waits for its reply.

Parameters:

  • request (Object) —

    anything the codec will encode

  • fields (Hash) —

    envelope fields for the request

Returns:

  • (Object) —

    the reply's payload

Raises:



131
132
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
 
# File 'lib/acemq/amqp/patterns/request_reply.rb', line 131

def call(request, **fields)
  # The correlation identifier is what pairs a reply with its request,
  # so it is generated here rather than taken from the caller: a caller
  # reusing one across two requests would have them answer each other.
  correlation = SecureRandom.uuid
  waiter = Waiter.new
  @lock.synchronize { @waiting[correlation] = waiter }
  # A monotonic clock, because this is a duration: the wall clock can
  # step backwards over an NTP correction and hand a round trip a
  # negative one.
  started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  # The pessimistic word until something better is known. Every way out
  # of the block below sets it, including the raise, and a failure that
  # is none of the three named ones is still a round trip that did not
  # come back with an answer.
  outcome = Telemetry::Outcome::FAILED
  type = nil

  begin
    type = publish(request, correlation, fields).type
    reply = answer(waiter.await(@timeout), correlation)
    outcome = Telemetry::Outcome::ANSWERED
    reply
  rescue RequestTimedOut
    outcome = Telemetry::Outcome::TIMED_OUT
    raise
  ensure
    @lock.synchronize { @waiting.delete(correlation) }
    record(type, outcome, Process.clock_gettime(Process::CLOCK_MONOTONIC) - started)
  end
end

#close ⇒ Object

Stops consuming replies. The generated queue goes with it.



164
165
166
167
168
 
# File 'lib/acemq/amqp/patterns/request_reply.rb', line 164

def close
  @consumer&.cancel
  @consumer = nil
  nil
end