Class: RuoteAMQP::Receiver

Inherits:
Ruote::Receiver
  • Object
show all
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

LaunchitemListener, WorkitemListener

Instance Attribute Summary collapse

Instance Method Summary collapse

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 |message|
    if AMQP.closing?
      # do nothing, we're going down
    else
      handle(message)
    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

#stop ⇒ Object 



94
95
96
97
 
# File 'lib/ruote-amqp/receiver.rb', line 94

def stop

  RuoteAMQP.stop!
end