Class: RuoteAMQP::Receiver
- Inherits:
-
Ruote::Receiver
- Object
- Ruote::Receiver
- RuoteAMQP::Receiver
- Defined in:
- lib/ruote-amqp/receiver.rb
Overview
AMQP Receiver
Used in conjunction with the RuoteAMQP::Participant, the WorkitemListener subscribes to a specific direct exchange and monitors for incoming workitems. It expects workitems to arrive serialized as JSON.
Configuration
AMQP configuration is handled by directly manipulating the values of the AMQP.settings
hash, as provided by the AMQP gem. No defaults are set by the listener. The only option
parsed by the initializer of the workitem listener is the queue
key (Hash expected). If no queue
key is set, the listener will subscribe to the ruote_workitems
direct exchange for workitems, otherwise it will subscribe to the direct exchange provided.
Usage
Register the engine or storage with the listener:
RuoteAMQP::Receiver.new(engine_or_storage)
The workitem listener leverages the asynchronous nature of the amqp gem, so no timers are setup when initialized.
Options
:queue and :launchitems
See the RuoteAMQP::Participant docs for information on sending workitems out to remote participants, and have them send replies to the correct direct exchange specified in the workitem attributes.
Direct Known Subclasses
Instance Attribute Summary collapse
-
#queue ⇒ Object
readonly
Returns the value of attribute queue.
Instance Method Summary collapse
-
#decode_workitem(msg) ⇒ Object
(feel free to overwrite me).
-
#initialize(engine_or_storage, opts = {}) ⇒ Receiver
constructor
Starts a new Receiver.
- #stop ⇒ Object
Constructor Details
#initialize(engine_or_storage, opts = {}) ⇒ Receiver
Starts a new Receiver
Two arguments for this method.
The first oone should be a Ruote::Engine, a Ruote::Storage or a Ruote::Worker instance.
The second one is a hash for options. There are two known options :
:queue for setting the queue on which to listen (defaults to ‘ruote_workitems’).
The :launchitems option :
:launchitems => true
# the receiver accepts workitems and launchitems
:launchitems => false
# the receiver only accepts workitems
:launchitems => :only
# the receiver only accepts launchitems
68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
# File 'lib/ruote-amqp/receiver.rb', line 68 def initialize(engine_or_storage, opts={}) super(engine_or_storage) @launchitems = opts[:launchitems] @queue = opts[:queue] || (@launchitems == :only ? 'ruote_launchitems' : 'ruote_workitems') RuoteAMQP.start! if opts[:unsubscribe] MQ.queue(@queue, :durable => true).unsubscribe sleep 0.300 end MQ.queue(@queue, :durable => true).subscribe do || if AMQP.closing? # do nothing, we're going down else handle() end end end |
Instance Attribute Details
#queue ⇒ Object (readonly)
Returns the value of attribute queue.
45 46 47 |
# File 'lib/ruote-amqp/receiver.rb', line 45 def queue @queue end |
Instance Method Details
#decode_workitem(msg) ⇒ Object
(feel free to overwrite me)
101 102 103 104 |
# File 'lib/ruote-amqp/receiver.rb', line 101 def decode_workitem(msg) (Rufus::Json.decode(msg) rescue nil) end |