Class: Tina4::QueueBackends::RabbitmqBackend

Inherits:
Object
  • Object
show all
Defined in:
lib/tina4/queue_backends/rabbitmq_backend.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ RabbitmqBackend

Returns a new instance of RabbitmqBackend.



6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 6

def initialize(options = {})
  require "bunny"
  @connection = Bunny.new(
    host: options[:host] || "localhost",
    port: options[:port] || 5672,
    username: options[:username] || "guest",
    password: options[:password] || "guest",
    vhost: options[:vhost] || "/"
  )
  @connection.start
  @channel = @connection.create_channel
  @queues = {}
  @exchanges = {}
  @max_retries = options[:max_retries] || 3
rescue LoadError
  raise "RabbitMQ backend requires the 'bunny' gem. Install with: gem install bunny"
end

Instance Attribute Details

#max_retries ⇒ Object

Queue propagates its own configuration onto the backend after construction.



25
26
27
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 25

def max_retries
  @max_retries
end

Instance Method Details

#acknowledge(message) ⇒ Object 



150
151
152
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 150

def acknowledge(message)
  complete(message)
end

#clear(_topic) ⇒ Object

Not performable on RabbitMQ - raises naming the backend and the operation.

clear() empties the queue. RabbitMQ cannot address messages by status, so the only thing it could do is queue.purge the WHOLE live queue. This used to do exactly that and return 0, silently destroying every pending job. Draining a live broker on a status-addressed clear is data loss.

Raises:

  • (NotImplementedError) 


236
237
238
239
240
241
242
243
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 236

def clear(_topic)
  raise NotImplementedError,
        "The rabbitmq queue backend cannot perform clear(): RabbitMQ " \
        "cannot address messages by status (basic.get pops the head of " \
        "the queue), so a status-addressed clear would have to drain the " \
        "entire live queue and destroy pending work. Use the file or " \
        "mongodb backend."
end

#close ⇒ Object

Close the AMQP channel and connection and release the socket.

IDEMPOTENT by construction: the handles are dropped in an ensure, so a second close finds nothing and returns. Before 3.13.95 they were left set, and Bunny raises on closing an already-closed channel - so a shutdown path that ran twice crashed on the second pass.



256
257
258
259
260
261
262
263
264
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 256

def close
  @channel&.close
  @connection&.close
ensure
  @channel = nil
  @connection = nil
  @queues = {}
  @exchanges = {}
end

#complete(_message) ⇒ Object

Acknowledge the in-flight message as done (terminal). Named complete() to match the lite/mongo backends AND the Job#complete lifecycle, which calls backend.complete (not acknowledge) — so job.complete now actually acks the broker message instead of being a silent no-op. multiple:false acks only this delivery. The stored tag is cleared so a double-complete is a safe no-op rather than a second ack on an unknown tag.



87
88
89
90
91
92
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 87

def complete(_message)
  return unless @last_delivery_tag

  @channel.acknowledge(@last_delivery_tag, false)
  @last_delivery_tag = nil
end

#dead_letter(message) ⇒ Object 



98
99
100
101
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 98

def dead_letter(message)
  dlq = get_queue("#{message.topic}.dead_letter")
  dlq.publish(message.to_json, persistent: true)
end

#dead_letters(topic, max_retries: 3) ⇒ Object

Dead-lettered jobs, read back from the .dead_letter queue this backend writes itself. RabbitMQ's own dead-letter EXCHANGE is not queryable, but the queue Tina4 maintains is -- so this ANSWERS rather than refusing, and a dead-letter handler written against the file backend finds the same jobs here (invariant 3).

Drain-and-republish: a read must not consume. Every message popped is published straight back, so the queue is unchanged by the read. Returns plain Hashes with string keys, matching the lite backend — the parsed body already carries id/topic/payload/attempts/error, so a caller reads it the same way on every backend.



165
166
167
168
169
170
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 165

def dead_letters(topic, max_retries: 3)
  drain_dead_letters(topic).map do |data|
    data["status"] = "dead"
    data
  end
end

#dequeue(topic) ⇒ Object 



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 53

def dequeue(topic)
  queue = get_queue(topic)
  # Manual ack: do NOT let bunny's default auto-ack remove the message on
  # pop. The message stays in-flight (unacked) until complete() acks it, so
  # a consumer crash before complete() makes the broker redeliver it
  # (at-least-once delivery) — parity with the Python/PHP masters, whose
  # basic_get uses auto_ack=false / no-ack=false. With the old auto-ack pop
  # the stored delivery_tag had already been acked, so a later
  # channel.acknowledge raised PRECONDITION_FAILED and closed the channel.
  delivery_info, _properties, payload = queue.pop(manual_ack: true)
  return nil unless payload

  data = JSON.parse(payload)
  # attempts and error MUST be carried back. Rebuilding the Job from
  # topic/payload/id alone reset attempts to 0 on every redelivery, so
  # fail()'s attempts >= max_retries check could never trip and a poison
  # job would be retried forever instead of dead-lettering.
  msg = Tina4::Job.new(
    topic: data["topic"],
    payload: data["payload"],
    id: data["id"],
    attempts: data["attempts"] || 0,
    error: data["error"]
  )
  @last_delivery_tag = delivery_info.delivery_tag
  msg
end

#enqueue(message) ⇒ Object 



27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 27

def enqueue(message)
  if message.priority.to_i > 0
    raise NotImplementedError,
          "The rabbitmq queue backend cannot honour push(priority): RabbitMQ " \
          "orders a queue FIFO. Native priority needs the queue DECLARED " \
          "with an x-max-priority argument, and an existing queue cannot " \
          "be redeclared with one (the broker answers PRECONDITION_FAILED), " \
          "so enabling it would break every queue already in service. Use " \
          "the file or mongodb backend for prioritised jobs."
  end

  if message.available_at
    raise NotImplementedError,
          "The rabbitmq queue backend cannot honour push(delay_seconds): " \
          "RabbitMQ has no per-message delay in core. The " \
          "rabbitmq_delayed_message_exchange plugin is not part of a standard " \
          "broker, and the TTL + dead-letter workaround head-of-line blocks (a " \
          "long-delayed job holds up every shorter one behind it in the same " \
          "queue). Use the file or mongodb backend for delayed jobs, or " \
          "schedule the push itself."
  end

  queue = get_queue(message.topic)
  queue.publish(message.to_json, persistent: true)
end

#fail(job, error = "") ⇒ Object

Record a failed attempt, then retry it or dead-letter it.

This did not exist. Job#fail guarded on respond_to?(:fail) and silently degraded to in-memory bookkeeping, so job.fail() NEVER reached the broker: the delivery stayed unacked, no dead letter was written, and both failed() and dead_letters() reported nothing. The job was lost as far as the application could see.

AMQP basic.nack(requeue=true) returns the ORIGINAL body unmodified -- the protocol carries no delivery counter -- so a retry RE-PUBLISHES a body carrying the new count instead. That is what every AMQP client that counts attempts does (Celery, Spring AMQP's RepublishMessageRecoverer, laravel-queue-rabbitmq).



116
117
118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 116

def fail(job, error = "")
  job.attempts += 1
  job.error = error
  if job.attempts >= @max_retries
    dead_letter(job)
  else
    enqueue(job)
  end
  # Ack LAST: the re-publish (or dead-letter) is durable before the
  # original leaves the queue, so a crash in between redelivers rather
  # than loses. That is at-least-once, which is the contract.
  complete(job)
end

#failed(_topic, max_retries: 3) ⇒ Object

Raises:

  • (NotImplementedError) 


172
173
174
175
176
177
178
179
180
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 172

def failed(_topic, max_retries: 3)
  raise NotImplementedError,
        "The rabbitmq queue backend cannot answer failed(): a job that " \
        "failed but is still retryable is re-published to the main topic, " \
        "so it cannot be told apart from a normal pending message without " \
        "draining the live queue. Returning an empty list would claim " \
        "nothing has failed. Use dead_letters() for exhausted jobs, or the " \
        "file or mongodb backend to enumerate retryable failures."
end

#purge(_topic, _status) ⇒ Object

Not performable on RabbitMQ - raises naming the backend and the operation.

purge(status) removes jobs SELECTED BY STATUS. RabbitMQ cannot address messages by status: basic.get pops the head of the queue and the only bulk operation is queue.purge, which empties the WHOLE live queue regardless of status. This used to drain that queue on any non-dead status, destroying every pending job - the destructive no-op ADR-0022 invariant 6 forbids. Refusing by name is the honest answer.

Raises:

  • (NotImplementedError) 


221
222
223
224
225
226
227
228
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 221

def purge(_topic, _status)
  raise NotImplementedError,
        "The rabbitmq queue backend cannot perform purge(): RabbitMQ " \
        "cannot address messages by status (basic.get pops the head of " \
        "the queue), so a status-addressed purge would have to drain the " \
        "entire live queue and destroy pending work. Use the file or " \
        "mongodb backend."
end

#requeue(message) ⇒ Object 



94
95
96
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 94

def requeue(message)
  enqueue(message)
end

#retry(job, delay_seconds: 0) ⇒ Object

Explicit re-queue requested by the caller (job.retry). Always re-enqueues regardless of the retry limit -- a manual override, distinct from the automatic fail() path.



133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 133

def retry(job, delay_seconds: 0)
  if delay_seconds.to_f > 0
    raise NotImplementedError,
          "The rabbitmq queue backend cannot honour retry(delay_seconds): " \
          "RabbitMQ has no per-message delay in core, for the same reason " \
          "push(delay_seconds) is refused. Re-queueing immediately while " \
          "silently dropping the delay would run the job far sooner than " \
          "asked. Use the file or mongodb backend for delayed retries."
  end

  job.attempts += 1
  job.error = nil
  enqueue(job)
  complete(job)
  true
end

#retry_failed(_topic, max_retries: 3) ⇒ Object

Raises:

  • (NotImplementedError) 


182
183
184
185
186
187
188
189
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 182

def retry_failed(_topic, max_retries: 3)
  raise NotImplementedError,
        "The rabbitmq queue backend cannot perform retry_failed(): it must " \
        "first enumerate the failed-but-retryable jobs, which are back on " \
        "the main topic and indistinguishable from pending work. Returning " \
        "0 would claim nothing needed retrying. Use retry(job_id) with an " \
        "id you already hold, or the file or mongodb backend."
end

#retry_job(topic, job_id: nil, delay_seconds: 0) ⇒ Object

Move ONE dead-lettered job back to its main topic.



192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 192

def retry_job(topic, job_id: nil, delay_seconds: 0)
  found = nil
  keep = []
  drain_dead_letters(topic).each do |data|
    if found.nil? && (job_id.nil? || data["id"].to_s == job_id.to_s)
      found = data
    else
      keep << data
    end
  end
  keep.each { |data| publish_to("#{topic}.dead_letter", data) }
  return false unless found

  found["attempts"] = (found["attempts"] || 0) + 1
  found["status"] = "pending"
  found["error"] = nil
  found["topic"] = topic
  publish_to(topic, found)
  true
end

#size(topic) ⇒ Object 



245
246
247
248
 
# File 'lib/tina4/queue_backends/rabbitmq_backend.rb', line 245

def size(topic)
  queue = get_queue(topic)
  queue.message_count
end