Module: AceMQ::AMQP::Patterns

Defined in:
lib/acemq/amqp/patterns.rb,
lib/acemq/amqp/patterns/sql.rb,
lib/acemq/amqp/patterns/saga.rb,
lib/acemq/amqp/patterns/outbox.rb,
lib/acemq/amqp/patterns/replay.rb,
lib/acemq/amqp/patterns/schema.rb,
lib/acemq/amqp/patterns/ordered.rb,
lib/acemq/amqp/patterns/streams.rb,
lib/acemq/amqp/patterns/pipeline.rb,
lib/acemq/amqp/patterns/scheduler.rb,
lib/acemq/amqp/patterns/sql_outbox.rb,
lib/acemq/amqp/patterns/sql_schema.rb,
lib/acemq/amqp/patterns/claim_check.rb,
lib/acemq/amqp/patterns/idempotency.rb,
lib/acemq/amqp/patterns/routing_slip.rb,
lib/acemq/amqp/patterns/request_reply.rb,
lib/acemq/amqp/patterns/consumer_group.rb,
lib/acemq/amqp/patterns/sql_idempotency.rb

Overview

Several consumers over one queue, started and stopped as one thing.

Defined Under Namespace

Modules: Claim, ClaimCheckStore, IdempotencyStore, OutboxStore, SQL Classes: ClaimCheckCodec, ConsumerGroup, FilesystemClaimCheckStore, InMemoryClaimCheckStore, InMemoryIdempotencyStore, InMemoryOutboxStore, InMemorySchemaRegistry, KeyedLocks, OutboxRecord, OutboxRelay, Pipeline, Replay, ReplayFailed, ReplayResult, RequestTimedOut, Requester, Responder, ResponderFailed, RoutingSlip, SQLIdempotencyStore, SQLOutboxStore, SQLSchemaRegistry, Saga, SagaResult, Scheduler, SchemaDefinition, SchemaNotFound, Step, StreamOffset, StreamRetention

Constant Summary collapse

REPLAYED_FROM_HEADER =

The queue a message was replayed out of.

"acemq-replayed-from"
REPLAYED_AT_HEADER =

When it was replayed, as RFC 3339.

"acemq-replayed-at"
REPLAY_COUNT_HEADER =

How many times it has been replayed.

"acemq-replay-count"
DEFAULT_STREAM_PREFETCH =

How many messages a stream consumer holds when nothing else is said.

RabbitMQ refuses a stream consumer with no prefetch at all, and the error it gives does not explain why, so there is always one.

The number is this library's choice and not part of the cross-language contract. Java and .NET default to 100 here, Go, Python and Ruby to 10, and the disagreement is deliberate: prefetch trades memory against throughput, the right answer depends on payload size and handler speed, and both of those belong to the application rather than to the protocol. A stream one library wrote is read by any of them whatever this says — the offset, the retention arguments and the message on the wire are the contract, and prefetch is a consumer-side setting that never leaves the channel. Do not harmonise this with Java's; state the number at the call site if it matters. See docs/streams.md.

10
SLIP_HEADER =

The itinerary, as JSON, on the message.

An application header, so it survives every hop and is readable by anything that can read JSON. The shape is shared with the other AceMQ libraries, which is why the keys inside it are routingKey and completedAt rather than anything more Rubyish: a slip written by a Go service is read by a Ruby one.

"acemq-routing-slip"
REPLY_TO_HEADER =

Where a responder should send its answer.

An application header as well as AMQP's own reply-to property, and the two always say the same thing. The header travels through the same envelope machinery as everything else and survives a hop through a service that rebuilds the message; the property is what a broker, a management console and the Java and .NET libraries understand. Writing only one of them is what kept a Java requester and a Ruby responder from talking to each other.

The rule is the same in all five libraries: a requester writes both, and a responder reads the header first and falls back to the property. Header first because it is the one that survives a rebuild — a service that reads a message and publishes a new one keeps the headers and usually drops the properties.

Deliberately without the x-acemq- prefix: that namespace belongs to the engine and is kept away from application headers, so a responder could never read this one if it were in there.

"acemq-reply-to"
ERROR_HEADER =

A responder's failure, carried back to whoever is waiting.

"acemq-error"

Class Method Summary collapse

Class Method Details

.by_correlation ⇒ Object

Orders by correlation identifier, which keeps one business action's messages in sequence.



73
74
75
 
# File 'lib/acemq/amqp/patterns/ordered.rb', line 73

def self.by_correlation
  ->(message) { message.envelope.correlation_id }
end

.by_header(name) ⇒ Object

Orders by an application header — a tenant, a customer, an aggregate identifier.



66
67
68
69
 
# File 'lib/acemq/amqp/patterns/ordered.rb', line 66

def self.by_header(name)
  name = name.to_s
  ->(message) { message.envelope.headers[name] }
end

.chain(handler, *middleware) ⇒ Proc

Wraps a handler in middleware.

mq.consume("orders.new", &Patterns.chain(
->(message) { place(message.payload) },
Patterns.with_logging { |line| logger.info(line) },
Patterns.with_timeout(10),
Patterns.with_idempotency(store)
))

The order reads outside-in: the first one named is the outermost, so logging above records what the timeout and the idempotency guard decided. They are applied in reverse to make that so, which is the order somebody reading the list expects rather than the order the code would fall into.

Parameters:

  • handler (#call) —

    the innermost handler

  • middleware (Array<#call>) —

    each given a handler, each returning one

Returns:



45
46
47
 
# File 'lib/acemq/amqp/patterns/pipeline.rb', line 45

def self.chain(handler, *middleware)
  middleware.reverse.reduce(handler) { |inner, wrap| wrap.call(inner) }
end

.declare_stream(connection, name, max_age: nil, max_bytes: nil, segment_bytes: nil) ⇒ Object

Declares a queue that keeps its messages.

Patterns.declare_stream(mq, "events", max_age: 7 * 24 * 3600,
                      max_bytes: 10 * 1024**3)

A stream is durable and can be neither exclusive nor auto-deleting. Those are set here rather than left to fail at the broker, whose refusal does not mention streams and reads like a bug in the caller's own code.

Parameters:

  • connection (Connection, Transport)
  • name (String)
  • max_age (Numeric, nil) (defaults to: nil) —

    seconds; discard messages older than this

  • max_bytes (Integer, nil) (defaults to: nil) —

    discard the oldest once this is exceeded

  • segment_bytes (Integer, nil) (defaults to: nil) —

    how large each file on disk gets



108
109
110
111
112
113
114
 
# File 'lib/acemq/amqp/patterns/streams.rb', line 108

def self.declare_stream(connection, name, max_age: nil, max_bytes: nil,
                        segment_bytes: nil)
  retention = StreamRetention.new(max_age: max_age, max_bytes: max_bytes,
                                  segment_bytes: segment_bytes)
  connection.declare_queue(name, durable: true, auto_delete: false, exclusive: false,
                                 arguments: retention.to_arguments)
end

.duration_argument(seconds) ⇒ 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.

A duration as RabbitMQ wants it: a number with a unit suffix rather than a bare number of anything.



202
203
204
205
206
207
208
209
 
# File 'lib/acemq/amqp/patterns/streams.rb', line 202

def self.duration_argument(seconds)
  seconds = seconds.to_i
  return "#{seconds / 86_400}D" if seconds.positive? && (seconds % 86_400).zero?
  return "#{seconds / 3600}h" if seconds.positive? && (seconds % 3600).zero?
  return "#{seconds / 60}m" if seconds.positive? && (seconds % 60).zero?

  "#{seconds}s"
end

.fingerprint(definition) ⇒ Object

Hashes a schema definition.

SHA-256 of the exact bytes, so two definitions differing only in whitespace hash differently and count as different schemas. Normalising would need a parser per format, and a registry that quietly treated two definitions as one because it mis-parsed them would be worse than one that is strict.



138
139
140
 
# File 'lib/acemq/amqp/patterns/schema.rb', line 138

def self.fingerprint(definition)
  Digest::SHA256.hexdigest(definition.to_s)
end

.follow_slip(connection, pipeline: nil, write: nil, &step) ⇒ Proc

Wraps a handler so the message carries on to its next stop.

mq.consume("charge-queue", &Patterns.follow_slip(mq) do |message|
charge(message.payload)      # the payload to send onwards
end)

The block returns the payload for the next stop, which may be the one it received or a changed copy. When the slip has no steps left the work is finished and nothing more is published.

The message is accepted only once the next one is out, so a failure to publish retries this step — which is why a step that changes anything should be idempotent.

Returning nil from the block ends the run there: nothing is published and the message is accepted. A step that decides a message goes no further is making a decision, not failing, and it is counted apart from both — the same rule then_publish follows.

Parameters:

Returns:

Raises:

  • (ArgumentError) 


439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 439

def self.follow_slip(connection, pipeline: nil, write: nil, &step)
  raise ArgumentError, "Patterns.follow_slip needs a block to do the work" unless step

  lambda do |message|
    slip = RoutingSlip.from(message.envelope, pipeline: pipeline)
    if slip.nil?
      next Ack.reject(FatalError.new(
                        "message #{message.id} has no routing slip, so there is " \
                        "nowhere to send it next"
                      ))
    end

    onwards = write.nil? ? slip : slip.as(write, pipeline: slip.pipeline || pipeline)
    carry_on(connection, onwards, message, step)
  rescue FatalError => e
    Ack.reject(e)
  end
end

.idempotent(store, key: nil, in_progress_delay: Ack::DEFAULT_IN_PROGRESS_DELAY, &handler) ⇒ Proc

Wraps a handler so a message that has already been handled is accepted without running it again.

store = Patterns::InMemoryIdempotencyStore.new(window: 6 * 3600)

mq.consume("orders.new", &Patterns.idempotent(store) do |message|
warehouse.reserve(message.payload)
Ack.accept
end)

A duplicate is accepted, not rejected. The work was done, so the message has been handled; dead-lettering it would raise an alarm about something that went right.

A redelivery that finds the message in progress -- claimed by a handler that is still running, or by one that failed and could not release its claim -- is neither run nor accepted: it is answered with Ack.in_progress, which puts it back after in_progress_delay without spending a retry attempt. Accepting it would lose the message if that claim never became a fact. A store with a claim method (both shipped stores) is asked for this three-way answer; one with only first_time? cannot tell in progress from done and is treated as before.

When the handler does not accept, the key is forgotten so that the retry can actually run. That ordering is the honest one — remembering a message that then failed would mean its retry silently does nothing — and it is also why this is a guard against duplicates rather than exactly-once. Between the handler finishing and the acknowledgement reaching the broker, a crash still leaves a message that will be delivered again. Only a store written in the same transaction as the work closes that gap, and no library can do that on the application's behalf.

Parameters:

  • store (IdempotencyStore) —

    where keys are remembered

  • key (#call, nil) (defaults to: nil) —

    a message's key, when it is not the message id. Use it when the natural key is in the payload: an order identifier that two different messages both carry, where handling either one twice is the thing to prevent.

  • in_progress_delay (Numeric) (defaults to: Ack::DEFAULT_IN_PROGRESS_DELAY) —

    seconds a message found in progress waits before it is looked at again. Keep it well under the store's claim timeout

Returns:

Raises:

  • (ArgumentError) 


196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
 
# File 'lib/acemq/amqp/patterns/idempotency.rb', line 196

def self.idempotent(store, key: nil, in_progress_delay: Ack::DEFAULT_IN_PROGRESS_DELAY,
                    &handler)
  raise ArgumentError, "Patterns.idempotent needs a block to wrap" unless handler

  lambda do |message|
    seen_as = (key ? key.call(message) : message.envelope.id).to_s
    if seen_as.empty?
      # Fatal rather than a retry: the key function will produce the same
      # nothing next time, and a guard that cannot key a message is not
      # guarding it.
      next Ack.reject(FatalError.new(
                        "message #{message.id} produced an empty idempotency key"
                      ))
    end

    run_once(store, seen_as, message, in_progress_delay, handler)
  end
end

.ordered(key, &handler) ⇒ Proc

Wraps a handler so messages sharing a key are never handled at once.

mq.consume("orders.new", concurrency: 16,
         &Patterns.ordered("x-order-id") { |message| apply(message.payload) })

Ordering per key, concurrency across keys.

What this does not do

It orders the handling of messages that have already been delivered. It cannot reorder ones the broker delivered out of order, and with several consumers on one queue it orders only within each process. Ordering across processes needs the messages to reach the same one to begin with, which is a routing decision rather than a handler one — a consistent hash exchange, or a queue per partition, which is what partition is for.

A message whose key comes out empty is handled with no ordering at all, because there is nothing to order it against.

Parameters:

  • key (#call, String, Symbol) —

    how to find a message's ordering key. Anything callable is given the message; a name is read as an application header, which is the common case and not worth a lambda.

Returns:

Raises:

  • (ArgumentError) 


50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/acemq/amqp/patterns/ordered.rb', line 50

def self.ordered(key, &handler)
  raise ArgumentError, "Patterns.ordered needs a block to wrap" unless handler

  pick = key.respond_to?(:call) ? key : by_header(key)
  locks = KeyedLocks.new

  lambda do |message|
    ordering = pick.call(message).to_s
    next handler.call(message) if ordering.empty?

    locks.holding(ordering) { handler.call(message) }
  end
end

.partition(key, count) ⇒ Integer

Maps a key onto one of count slots.

For deciding which queue or which worker a message belongs to, when ordering has to hold across processes rather than within one. FNV-1a rather than Ruby's own hash, and that is the whole point: Ruby randomises string hashes per process, so two workers would disagree about where a key belongs, and so would a Go publisher and a Ruby consumer. This gives the same answer everywhere, for ever.

Parameters:

  • key (String)
  • count (Integer) —

    how many partitions

Returns:

  • (Integer) —

    0 to count - 1



89
90
91
92
93
94
95
96
97
 
# File 'lib/acemq/amqp/patterns/ordered.rb', line 89

def self.partition(key, count)
  return 0 if count.to_i <= 1

  hash = 2_166_136_261
  key.to_s.each_byte do |byte|
    hash = ((hash ^ byte) * 16_777_619) & 0xFFFF_FFFF
  end
  hash % count
end

.partitioned_routing_key(base, key, partitions) ⇒ Object

A routing key with its partition on the end, for publishing into a queue-per-partition arrangement.

Patterns.partitioned_routing_key("orders", order_id, 8)   # => "orders.3"


103
104
105
 
# File 'lib/acemq/amqp/patterns/ordered.rb', line 103

def self.partitioned_routing_key(base, key, partitions)
  "#{base}.#{partition(key, partitions)}"
end

.read_stream(connection, stream, offset: StreamOffset.next, prefetch: DEFAULT_STREAM_PREFETCH, name: nil, **options, &handler) ⇒ Consumer

Reads a stream from a chosen position.

Patterns.read_stream(mq, "events", offset: Patterns::StreamOffset.first,
                   prefetch: 100) do |message|
project(message.payload)
Ack.accept
end

How this differs from consuming a queue

Acknowledging does not remove the message: a stream keeps everything until its retention policy discards it. What an acknowledgement does is advance this consumer's position, so restarting from next carries on rather than re-reading.

Rejecting does not dead-letter it either, because there is nothing to remove it from. A message that cannot be handled has to be dealt with by the handler — logged, copied to another queue, counted — and the stream moves on regardless. That is the trade a stream makes: nothing is lost, and nothing is retried for you.

Which is why the retry policy here defaults to RetryPolicy.none whatever the connection carries. A policy that republished a failed message onto a stream would append a second copy to it rather than redelivering the first, and a projection reading that stream would then see the message twice.

Parameters:

  • connection (Connection)
  • stream (String)
  • offset (StreamOffset) (defaults to: StreamOffset.next) —

    where to start

  • prefetch (Integer) (defaults to: DEFAULT_STREAM_PREFETCH) —

    required by RabbitMQ for a stream

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

    identifies this consumer to the broker, which is what makes server-side offset tracking possible

Returns:

Raises:

  • (ArgumentError) 


167
168
169
170
171
172
173
174
175
176
177
 
# File 'lib/acemq/amqp/patterns/streams.rb', line 167

def self.read_stream(connection, stream, offset: StreamOffset.next,
                     prefetch: DEFAULT_STREAM_PREFETCH, name: nil, **options, &handler)
  raise ArgumentError, "read_stream needs a block to handle messages" unless handler

  connection.consume(
    stream, prefetch: prefetch, tag: name,
            retry_policy: options.delete(:retry_policy) || RetryPolicy.none,
            arguments: { "x-stream-offset" => offset.to_argument },
            **options, &refusing_put_back(stream, handler)
  )
end

.record(connection, payload, to:, exchange: "", envelope: nil, codec: nil, **fields) ⇒ OutboxRecord

Encodes a payload into an outbox record, ready for store.add.

Call it inside the transaction that does the work:

db.transaction do
orders.insert(order)
store.add(Patterns.record(mq, event, to: "order.placed",
                          exchange: "orders-events",
                          type: "order.placed.v2"))
end

The envelope is built here, by the same rules Connection#publish uses and with this connection's origin, so a message that went through the outbox is indistinguishable on the wire from one that did not. That is the point: the outbox is a delivery mechanism, not a different kind of message.

Parameters:

  • connection (Connection) —

    whose codec and origin to use

  • payload (Object) —

    anything the codec will encode

  • to (String) —

    the routing key

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

    empty for the default exchange

  • envelope (Envelope, nil) (defaults to: nil) —

    one built elsewhere

  • fields (Hash) —

    envelope fields, when no envelope is given

Returns:



131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/acemq/amqp/patterns/outbox.rb', line 131

def self.record(connection, payload, to:, exchange: "", envelope: nil, codec: nil,
                **fields)
  if envelope && !fields.empty?
    raise ArgumentError,
          "record was given both an envelope and the fields to build one " \
          "(#{fields.keys.join(", ")}); pass one or the other"
  end

  codec = codec.nil? ? connection.codec : Codec.check!(codec)
  envelope ||= Envelope.new(origin: connection.origin, **fields)
  OutboxRecord.new(
    id: envelope.id, exchange: exchange, routing_key: to,
    body: codec.encode(payload), content_type: codec.content_type,
    headers: envelope.to_headers(to), created_at: Time.now
  )
end

.replay(connection, from:, exchange: "", routing_key: nil, limit: 0, deadline: 0, restart: true, &filter) ⇒ ReplayResult

Moves messages from a queue back onto an exchange.

result = Patterns.replay(mq, from: "orders.new.dlq",
                       exchange: "orders-events", limit: 500) do |envelope, _body|
envelope.error.include?("timeout")
end

result.to_s   # => "moved 37, skipped 463 (drained)"

The block decides which messages go. Returning false leaves one where it is, which is what makes a replay something that can be done in stages rather than all at once. Without a block every message goes.

Each message is stamped so a replayed one can be told from an original: REPLAYED_FROM_HEADER, REPLAYED_AT_HEADER and REPLAY_COUNT_HEADER. A consumer that needs to treat them differently can, and one that does not is unaffected.

A replayed message goes back on attempt one, with the reason that dead-lettered it cleared. Anything else does not work: a message dead-lettered on the last attempt of a five-attempt policy arrives back on attempt five, and the consumer gives up on it before the handler ever sees it — so an operator who has just fixed the bug moves two thousand messages from the dead-letter queue to the dead-letter queue. Pass restart: false to put back exactly what was there, which is what an audit wants, or a queue read by something that counts attempts itself.

Parameters:

  • connection (Connection)
  • from (String) —

    the queue the messages are on now

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

    where they go back to; empty publishes to a queue by name

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

    overrides the message's own. Nil keeps it, so a message goes back where it came from rather than everywhere.

  • limit (Integer) (defaults to: 0) —

    stop after this many; zero for no limit, which against a queue somebody is still writing to may mean never stopping

  • deadline (Numeric) (defaults to: 0) —

    stop after this many seconds; zero for none

  • restart (Boolean) (defaults to: true) —

    give each message a fresh set of attempts, clearing x-acemq-attempt back to 1 and x-acemq-error with it

Returns:

Raises:



96
97
98
99
100
101
102
103
104
 
# File 'lib/acemq/amqp/patterns/replay.rb', line 96

def self.replay(connection, from:, exchange: "", routing_key: nil, limit: 0,
                deadline: 0, restart: true, &filter)
  raise ArgumentError, "a replay needs a queue to read from" if from.to_s.empty?

  refuse_a_loop(from, exchange, routing_key)
  Replay.new(connection, from: from, exchange: exchange, routing_key: routing_key,
                         limit: limit, deadline: deadline, restart: restart,
                         filter: filter).run
end

.reply_address(message) ⇒ String

Where to send the answer: the header first, the AMQP property second.

Both, because the five libraries do not all write both yet and a responder that read only one of them could not answer half the fleet. A requester here writes both and they agree, so which one is read makes no difference; the order matters only for a request from somewhere else.

Parameters:

Returns:

  • (String) —

    empty when the request asked for no answer



455
456
457
458
 
# File 'lib/acemq/amqp/patterns/request_reply.rb', line 455

def self.reply_address(message)
  header = message.envelope.headers[REPLY_TO_HEADER].to_s
  header.empty? ? message.reply_to.to_s : header
end

.send_to(connection, step, slip, payload, fields = {}) ⇒ 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. 



513
514
515
516
517
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 513

def self.send_to(connection, step, slip, payload, fields = {})
  headers = fields.fetch(:headers, {}).merge(slip.to_headers)
  connection.publish(payload, to: step.routing_key, exchange: step.exchange,
                              **fields.merge(headers: headers, route: slip.to_route))
end

.serve(connection, queue, **options, &handler) ⇒ Responder

Answers requests on a queue.

responder = Patterns.serve(mq, "price.requests") do |message|
{ "price" => catalogue.price(message.payload["sku"]) }
end

The block returns the response rather than an Ack: a responder's job is to answer, and what to do with the request afterwards follows from whether it could. Raising sends the failure back to the caller, because a caller blocked on a reply should learn that it failed rather than wait out its timeout.

Parameters:

Returns:

Raises:

  • (ArgumentError) 


319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
 
# File 'lib/acemq/amqp/patterns/request_reply.rb', line 319

def self.serve(connection, queue, **options, &handler)
  raise ArgumentError, "Patterns.serve needs a block to answer with" unless handler

  # Made before the subscribe rather than after it, and closed over by the
  # handler rather than reached through the responder this returns. A
  # broker may hand the first request over from inside the subscribe —
  # which is what a queue with a backlog looks like from in here — and a
  # handler reaching for a responder the subscribe has not returned yet
  # would answer that request and count nothing. Silent, and only ever at
  # start-up, which is the worst place to lose the first number of the
  # day. .NET had to lift its counters out of the responder for exactly
  # this, and Java gets it from field initialisers running before the
  # constructor body.
  counters = Responder::Counters.new
  consumer = connection.consume(queue, **options) do |message|
    reply_to = reply_address(message)
    if reply_to.empty?
      # Counted before the delivery is settled, and settled rather than
      # retried: retrying cannot make a reply queue appear, so this is
      # dead-lettered. Anything above zero here means a caller is
      # publishing where it meant to request.
      counters.unanswerable!
      next Ack.reject(FatalError.new(
                        "request #{message.id} carries neither the #{REPLY_TO_HEADER} " \
                        "header nor a reply-to property, so there is nowhere to reply"
                      ))
    end

    answer(connection, reply_to, message, handler, counters)
  end
  Responder.new(consumer, counters)
end

.then_publish(connection, to:, exchange: "", &step) ⇒ Proc

Publishes the result of handling a message onwards.

mq.consume("orders.new", &Patterns.then_publish(mq, to: "shipment.requested",
                                              exchange: "shipping-events") do |m|
m.payload["digital"] ? nil : { "order_id" => m.payload["order_id"] }
end)

The step that makes a pipeline out of a chain of services: this one consumes, does its work, and publishes what comes out, carrying the correlation forward and recording what caused what.

Returning nil publishes nothing and accepts the message, which is how a step says "this one does not continue" without inventing an empty message for the next service to work out how to ignore.

The message is accepted only once the next one is out. A publish that fails retries the input, so the work runs again — which is why a step that changes anything should be idempotent.

Parameters:

  • connection (Connection)
  • to (String) —

    the routing key for what comes out

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

    empty for the default exchange

Returns:

Raises:

  • (ArgumentError) 


141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
 
# File 'lib/acemq/amqp/patterns/pipeline.rb', line 141

def self.then_publish(connection, to:, exchange: "", &step)
  raise ArgumentError, "Patterns.then_publish needs a block to do the work" unless step

  lambda do |message|
    outgoing = step.call(message)
    next Ack.accept if outgoing.nil?

    begin
      # Mandatory, because accepting is what makes this hop final: a next
      # message nothing is bound to is confirmed and dropped by the broker.
      connection.publish(outgoing, to: to, exchange: exchange, mandatory: true,
                                   correlation_id: message.envelope.correlation_id,
                                   causation_id: message.envelope.id)
    rescue StandardError => e
      next Ack.retry("the work for message #{message.id} is done but the next " \
                     "message did not go out: #{e.message}")
    end
    Ack.accept
  end
end

.with_idempotency(store, key: nil) ⇒ Object

idempotent as middleware, so it can sit in a chain.



109
110
111
 
# File 'lib/acemq/amqp/patterns/pipeline.rb', line 109

def self.with_idempotency(store, key: nil)
  ->(inner) { idempotent(store, key: key, &inner) }
end

.with_logging(&write) ⇒ Object

Records what happened to each message.

Patterns.with_logging { |line| logger.info(line) }

A block that is given a line, so it fits a Logger, a structured logger, puts or a test — without this library choosing a logging gem on anybody's behalf.

An exception is logged and then re-raised. Swallowing it would turn a failure into an acknowledgement, and not logging it would leave the middleware blind to the case somebody added it for.

Raises:

  • (ArgumentError) 


90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/acemq/amqp/patterns/pipeline.rb', line 90

def self.with_logging(&write)
  raise ArgumentError, "Patterns.with_logging needs a block to write with" unless write

  lambda do |inner|
    lambda do |message|
      started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
      begin
        ack = inner.call(message)
      rescue StandardError => e
        write.call(log_line(message, started, "raised #{e.class}: #{e.message}"))
        raise
      end
      write.call(log_line(message, started, describe_ack(ack)))
      ack
    end
  end
end

.with_ordering(key) ⇒ Object

ordered as middleware.



114
115
116
 
# File 'lib/acemq/amqp/patterns/pipeline.rb', line 114

def self.with_ordering(key)
  ->(inner) { ordered(key, &inner) }
end

.with_timeout(seconds) ⇒ Object

Reports a handler that ran longer than it was allowed.

An overrun is reported as a retry, whatever the handler said about itself. That is a choice between two imperfect answers: retrying work that may have succeeded risks doing it twice, and accepting work that may have failed loses it. Duplicates are a problem somebody can solve — see idempotent — and a lost message is not.

It reports, and does not interrupt. Ruby's Timeout.timeout would interrupt, by raising inside whatever line the handler happened to be on, which can leave a transaction half-written or a lock held by nobody; and it would not help anyway, because the message is held until the handler returns either way. What this buys is that a handler which quietly takes three minutes stops being invisible.

Parameters:

  • seconds (Numeric) 


65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/acemq/amqp/patterns/pipeline.rb', line 65

def self.with_timeout(seconds)
  lambda do |inner|
    lambda do |message|
      started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
      ack = inner.call(message)
      took = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started
      next ack if took <= seconds

      Ack.retry("handling message #{message.id} took #{took.round(3)} seconds, " \
                "longer than the #{seconds} it is allowed")
    end
  end
end