Class: AceMQ::AMQP::Patterns::Scheduler

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

Overview

Delivering a message later.

Patterns::Scheduler.on(mq) do |scheduler|
scheduler.in(4 * 3600, invoice, to: "invoice.due", exchange: "billing")
scheduler.at(renewal_date, policy, to: "policy.renew", exchange: "policies")
end

Why not a per-message time to live

The obvious implementation is to set expiration on the message, drop it in a queue nobody consumes, and let it dead-letter to its destination. It is what most articles suggest and it is wrong for anything but a single fixed delay, because a classic queue expires messages only at its head.

Put a four-hour message in, then a one-minute message behind it, and the one-minute message is delivered in four hours. Nothing reports this: the queue looks healthy, the message is not lost, it is simply late by a factor nobody predicted. It is the single most common way a home-made scheduler fails, and it fails in production under mixed load rather than in testing under uniform load.

What this does instead

A small ladder of queues, each with a uniform time to live, and a message hops through them until it is due:

acemq.schedule.1s  acemq.schedule.10s  acemq.schedule.1m
acemq.schedule.10m  acemq.schedule.1h

Every message in a given rung has the same delay, so head-of-line expiry is not a problem — the head is always the message due soonest. Each expiry returns the message to this scheduler, which either delivers it or puts it in the largest rung that does not overshoot. A four-hour delay is four one-hour hops; a ninety-second delay is one minute, then three tens.

The cost is honest and worth stating: a long delay is several broker round trips rather than one, and delivery is accurate to about the smallest rung rather than to the second. A scheduler that must fire at 09:00:00.000 exactly is a scheduler, not a message broker.

The alternative is RabbitMQ's delayed-message-exchange plugin, which does this properly and is a plugin — so it is not available everywhere, and a library that silently required it would be a library that works on your laptop.

The names are the contract

Every name, argument and header here is shared with the Java, Go, .NET and Python libraries. A Ruby service and a Java service scheduling on one broker declare the same five rungs and the same control queue, so a rung declared with a different time to live is a PRECONDITION_FAILED for whichever of them declares second, and a header spelled differently is a message that reaches the control queue and cannot be routed. Nothing in this class is a local decision.

Defined Under Namespace

Classes: Verbatim

Constant Summary collapse

EXCHANGE =

Where a message waits, and where it comes back to be re-examined.

"acemq.schedule"
CONTROL =

The queue every expired message returns to.

"acemq.schedule.due"
TARGET_EXCHANGE =

Deliberately not the x-acemq- prefix. That one is reserved: Headers.reserved? matches it, Envelope refuses it in an application's own header hash outright, and what does arrive under it is materialised as an envelope field rather than handed back as a header. A scheduler header using it would be refused on publish, and the version of this that got past that would be gone on consume.

"x-schedule-exchange"
TARGET_ROUTING_KEY =

The routing key the message should eventually carry.

"x-schedule-routing-key"
DUE_AT =

When it is due, as epoch milliseconds — the same integer x-acemq-first-seen is written as, and what Java's Instant#toEpochMilli produces. A formatted timestamp would be friendlier to read in the management console and would put a parser, a time zone and a resolution into the contract; an integer has none of those to disagree about.

"x-schedule-due-at"
CONTENT_TYPE =

What the payload was encoded as when it was scheduled.

Carried because the scheduler republishes bytes rather than objects, and a consumer picks its codec from the content type. Publishing pre-encoded bytes under application/octet-stream produces a message the intended consumer cannot decode — it arrives, it is the right bytes, and nothing can read it.

"x-schedule-content-type"
RUNGS =

The rungs, longest first, in seconds.

Five of them, spanning a second to an hour. More rungs mean finer accuracy and more queues; fewer mean more hops for a long delay. This spread delivers a one-day message in twenty-four hops and a one-minute message in one, which is the right way round — short delays are common and want to be cheap.

[3600, 600, 60, 10, 1].freeze
MESSAGE_TYPE =

What a scheduled message's type is, in the envelope.

"ScheduledMessage"
PREFETCH =

How many expiries the control consumer holds at once.

50

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection) ⇒ Scheduler

Returns a new instance of Scheduler.

Parameters:



240
241
242
243
244
245
246
247
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 240

def initialize(connection)
  @connection = connection
  @bytes = BytesCodec.new
  @lock = Mutex.new
  @counters = { scheduled: 0, delivered: 0, hops: 0, malformed: 0 }
  self.class.declare(connection)
  @subscription = subscribe(connection)
end

Class Method Details

.arguments_for(seconds) ⇒ Hash

The arguments a rung queue has to carry, and the whole of them.

These three, classic, durable, and no others. A fourth argument here is a queue the Java library cannot declare.

Parameters:

  • seconds (Numeric)

Returns:

  • (Hash) 


213
214
215
216
217
218
219
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 213

def self.arguments_for(seconds)
  {
    "x-message-ttl" => (seconds * 1000).round,
    "x-dead-letter-exchange" => EXCHANGE,
    "x-dead-letter-routing-key" => CONTROL
  }
end

.declare(connection) ⇒ Array<String>

Declares the exchange, the five rungs and the control queue, and binds them.

Called by on, and public because a deployment that applies its topology up front should be able to include the scheduler's without starting a consumer. Declaring is idempotent, so doing both is free.

Parameters:

  • connection (Connection, Transport) —

    anything answering declare_exchange, declare_queue and bind

Returns:

  • (Array<String>) —

    the queues, longest rung first, control last



186
187
188
189
190
191
192
193
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 186

def self.declare(connection)
  connection.declare_exchange(EXCHANGE, kind: :direct, durable: true)
  RUNGS.each do |rung|
    declare_queue(connection, rung_name(rung), arguments_for(rung))
  end
  declare_queue(connection, CONTROL, {})
  RUNGS.map { |rung| rung_name(rung) } + [CONTROL]
end

.describe(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. 



222
223
224
225
226
227
228
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 222

def self.describe(seconds)
  millis = (seconds * 1000).round
  return "#{millis / 3_600_000}h" if (millis % 3_600_000).zero?
  return "#{millis / 60_000}m" if (millis % 60_000).zero?

  "#{millis / 1000}s"
end

.epoch(value) ⇒ 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 header read back as epoch milliseconds, or nil when it is not a number.

Lenient, like Envelope.from_headers: a producer that wrote the integer as a string is still a producer whose message is due, and only a header that means nothing at all makes this nil.



338
339
340
341
342
343
344
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 338

def self.epoch(value)
  return nil if value.nil? || value == true || value == false

  Integer(value.to_s, 10)
rescue ArgumentError, TypeError
  nil
end

.millis(moment) ⇒ 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 moment as epoch milliseconds, which is what DUE_AT carries and what Java's Instant#toEpochMilli writes.



328
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 328

def self.millis(moment) = (moment.to_f * 1000).to_i

.on(connection) ⇒ Scheduler

Opens a scheduler on a connection, with its queues declared.

With a block the scheduler is yielded and closed afterwards, which is what Java's try-with-resources says; without one it runs until #close.

Parameters:

Returns:



165
166
167
168
169
170
171
172
173
174
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 165

def self.on(connection)
  scheduler = new(connection)
  return scheduler unless block_given?

  begin
    yield scheduler
  ensure
    scheduler.close
  end
end

.rung_name(seconds) ⇒ String

The queue a delay of this many seconds waits in.

A duration divisible by an hour renders as {n}h, one divisible by a minute as {n}m, and anything else as {n}s — Java's describe, kept to the letter, because the names it produces are the queues on the broker.

Parameters:

  • seconds (Numeric)

Returns:

  • (String) 


204
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 204

def self.rung_name(seconds) = "#{EXCHANGE}.#{describe(seconds)}"

Instance Method Details

#at(moment, payload, to:, exchange: "") ⇒ void

This method returns an undefined value.

Delivers a message at a moment.

The payload is encoded once, here, and carried as bytes from then on. Its content type goes with it, because that is how the eventual consumer chooses a codec.

Parameters:

  • moment (Time) —

    anything in the past delivers immediately

  • payload (Object) —

    anything the connection's codec will encode

  • to (String) —

    the routing key it should eventually carry

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

    where it should eventually go



274
275
276
277
278
279
280
281
282
283
284
285
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 274

def at(moment, payload, to:, exchange: "")
  codec = @connection.codec
  headers = {
    TARGET_EXCHANGE => exchange.to_s,
    TARGET_ROUTING_KEY => to.to_s,
    DUE_AT => self.class.millis(moment),
    CONTENT_TYPE => codec.content_type
  }
  count(:scheduled)
  route(codec.encode(payload), headers)
  nil
end

#close ⇒ Object

Stops the control consumer. The queues stay, and so do the messages waiting in them.



313
314
315
316
317
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 313

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

#delivered ⇒ Object

Messages that reached their destination.



291
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 291

def delivered = read(:delivered)

#hops ⇒ Object

How many times a message moved between rungs.

Divided by #delivered this is the average number of hops, which is the number to look at if the scheduler is busier than expected: long delays cost hops.



298
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 298

def hops = read(:hops)

#in(delay, payload, to:, exchange: "") ⇒ void

This method returns an undefined value.

Delivers a message after a delay.

in is a Ruby keyword, so this method is only ever reachable through its receiver — scheduler.in(...) — which is how it reads anyway.

Parameters:

  • delay (Numeric) —

    seconds; zero or negative delivers immediately

  • payload (Object) —

    anything the connection's codec will encode

  • to (String) —

    the routing key it should eventually carry

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

    where it should eventually go



259
260
261
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 259

def in(delay, payload, to:, exchange: "")
  at(Time.now + delay, payload, to: to, exchange: exchange)
end

#malformed ⇒ Object

Messages that reached the control queue without the headers a scheduled message carries, and were therefore dropped.

Not zero means something other than a scheduler is publishing into acemq.schedule, which it must not: those queues are an implementation detail of this class.



306
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 306

def malformed = read(:malformed)

#running? ⇒ Boolean

Whether the control consumer is still being delivered to.

Returns:

  • (Boolean) 


309
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 309

def running? = !!@subscription&.open?

#scheduled ⇒ Object

Messages handed to this scheduler.



288
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 288

def scheduled = read(:scheduled)

#to_s ⇒ Object 



319
320
321
322
 
# File 'lib/acemq/amqp/patterns/scheduler.rb', line 319

def to_s
  "Scheduler{rungs=#{RUNGS.map { |r| self.class.describe(r) }.join(", ")}, " \
    "scheduled=#{scheduled}, delivered=#{delivered}}"
end