Class: AceMQ::Rails::Consumer

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/rails/consumer.rb

Overview

A handler for one queue, written as a class.

class OrdersConsumer < AceMQ::Rails::Consumer
queue "orders.new"
concurrency 4
retries max_attempts: 5, initial_delay: 1, max_delay: 60

def call(message)
  Order.place!(message.payload)
  accept
end
end

A class rather than a block in an initializer, for three reasons that come from Rails rather than from taste. It can be unit-tested by calling new.call(message) with no broker anywhere. It lives in app/consumers, where the autoloader finds it and a reload replaces it. And it has a name, which is what a metric label, a log line and a stack trace want.

Subclassing registers it. Nothing runs until the consumer process starts it: the registry holds classes, and the code that turns a class into a subscription lives in Runner, which the web server never loads.

Constant Summary collapse

Ack =

What #call must return. Re-exported so a consumer never has to reach into the library's namespace for the one constant it uses every time.

AceMQ::AMQP::Ack

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.arguments(table = nil) ⇒ Object

Arguments passed to basic.consume, for a stream's x-stream-offset and the like.



110
111
112
113
114
 
# File 'lib/acemq/rails/consumer.rb', line 110

def arguments(table = nil)
  return setting(:arguments) || {} if table.nil?

  @arguments = table
end

.broker_wait_threshold(seconds = nil) ⇒ Object

Where a retry of this consumer's messages waits. See retries.



102
103
104
105
106
 
# File 'lib/acemq/rails/consumer.rb', line 102

def broker_wait_threshold(seconds = nil)
  return setting(:broker_wait_threshold) if seconds.nil?

  @broker_wait_threshold = seconds
end

.codec(instance = nil) ⇒ Object

The codec for this queue, when it is not the application's.



76
77
78
79
80
 
# File 'lib/acemq/rails/consumer.rb', line 76

def codec(instance = nil)
  return setting(:codec) if instance.nil?

  @codec = AceMQ::AMQP::Codec.check!(instance)
end

.concurrency(count = nil) ⇒ Object

Messages worked on at once by one instance of this consumer.

One, unless the application's consumer.concurrency says otherwise. Raising it gives up the order the broker offers messages in — which is the point of raising it, and worth saying out loud because concurrency 8 reads like free throughput and is not.



62
63
64
65
66
 
# File 'lib/acemq/rails/consumer.rb', line 62

def concurrency(count = nil)
  return setting(:concurrency) if count.nil?

  @concurrency = Integer(count)
end

.inherited(subclass) ⇒ 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. 



133
134
135
136
 
# File 'lib/acemq/rails/consumer.rb', line 133

def inherited(subclass)
  super
  Registry.register(subclass)
end

.prefetch(count = nil) ⇒ Object

Unacknowledged messages this consumer holds at once.



69
70
71
72
73
 
# File 'lib/acemq/rails/consumer.rb', line 69

def prefetch(count = nil)
  return setting(:prefetch) if count.nil?

  @prefetch = Integer(count)
end

.queue(name = nil) ⇒ Object

The queue this consumer reads. Required, and not inherited.



50
51
52
53
54
 
# File 'lib/acemq/rails/consumer.rb', line 50

def queue(name = nil)
  return @queue if name.nil?

  @queue = name.to_s
end

.retries(policy = nil, **settings) ⇒ Object

The retry ladder for this queue.

retries max_attempts: 5, initial_delay: 1, max_delay: 60
retries AceMQ::AMQP::RetryPolicy.exponential(5, 1, 60)

broker_wait_threshold: decides where a wait is spent: shorter than it, in this process holding a prefetch slot and holding up a drain; longer, on a rung queue in the broker, where a restart does not have to survive it. It is a shutdown setting as much as a reliability one — see docs/lifecycle.md.



92
93
94
95
96
97
98
99
 
# File 'lib/acemq/rails/consumer.rb', line 92

def retries(policy = nil, **settings)
  return @retry_policy = policy if policy
  return setting(:retry_policy) if settings.empty?

  @broker_wait_threshold = settings.delete(:broker_wait_threshold) ||
                           @broker_wait_threshold
  @retry_policy = Configuration::ConsumerDefaults.new.assign(settings).retry_policy
end

.runnable? ⇒ Boolean

Whether this class is one the runner should subscribe.

A class with no queue is an abstract intermediate — the ApplicationConsumer holding shared behaviour is the normal Rails shape, and it must not produce a subscription to a queue called nil.

Returns:

  • (Boolean) 


130
 
# File 'lib/acemq/rails/consumer.rb', line 130

def runnable? = !queue.nil? && !queue.empty?

.tag(value = nil) ⇒ Object

The consumer tag, which is what the broker's management interface shows beside the subscription. The class name by default, because OrdersConsumer in a list of subscriptions is worth more than a UUID.



119
120
121
122
123
 
# File 'lib/acemq/rails/consumer.rb', line 119

def tag(value = nil)
  return @tag || name if value.nil?

  @tag = value.to_s
end

Instance Method Details

#call(message) ⇒ AceMQ::AMQP::Ack

Handles one message.

Must return an AMQP::Ack: #accept, #retry_later, #reject or #park. Raising is the other way to reject — the retry ladder decides what happens next, and an exception that reaches here is reported to Rails.error before it is settled, so an error tracker sees it.

Parameters:

  • message (AceMQ::AMQP::Message)

Returns:

  • (AceMQ::AMQP::Ack)

Raises:

  • (NotImplementedError) 


162
163
164
165
 
# File 'lib/acemq/rails/consumer.rb', line 162

def call(message)
  raise NotImplementedError,
        "#{self.class.name} must define #call(message) and return an Ack"
end