Class: RuoteAMQP::ParticipantProxy

Inherits:
Object
  • Object
show all
Includes:
Ruote::LocalParticipant
Defined in:
lib/ruote-amqp/participant.rb

Overview

AMQP Participants

The RuoteAMQP::ParticipantProxy allows you to send workitems (serialized as JSON) or messages to any AMQP queues right from the process definition. When combined with the RuoteAMQP::Receiver you can easily leverage an extremely powerful local/remote participant combinations.

For local/remote participants The local part of the RuoteAMQP::ParticipantProxy relies on the presence of a RuoteAMQP::Receiver. Workitems are sent to the remote participant and the local part does not normally reply to the engine. Instead the engine will continue when a reply is received on the ‘ruote_workitems’ queue (see RuoteAMQP::Receiver).

Of course, the standard :forget => true format can be used even with remote particpants and :forget can even be set as a default in the options.

A simple way to create a remote participant to act upon workitems is to use the daemon-kit ruote responder.

Simple AMQP messages are treated as ‘fire and forget’ and the flow will continue when the local participant has queued the message for sending. (As there is no meaningful way to receive a workitem in reply).

Configuration

AMQP configuration is handled by directly manipulating the values of the AMQP.settings hash, as provided by the AMQP gem. No AMQP defaults are set by the participant.

Usage

Define the queue used by an AMQP participant :

engine.register_participant(
  :delete_user, RuoteAMQP::ParticipantProxy, 'queue' => 'user_manager')

Sending a workitem to the remote participant defined above:

Ruote.process_definition do
  sequence do
    delete_user
  end
end

Let the local participant reply to the engine without involving the receiver

Ruote.process_definition do
  sequence do
    delete_user :forget => true
  end
end

Setting up the participant in a slightly more ‘raw’ way:

engine.register_participant(
  :amqp, RuoteAMQP::ParticipantProxy )

Sending a workitem to a specific queue:

Ruote.process_definition do
  sequence do
    amqp :queue => 'test', 'command' => '/run/regression_test'
  end
end

Setup a ‘fire and forget’ participant that always replies to the engine:

engine.register_participant(
  :jfdi, RuoteAMQP::ParticipantProxy, 'forget' => true )

Sending a message example to a specific queue (both steps are equivalent):

Ruote.process_definition do
  sequence do
    amqp :queue => 'test', :message => 'foo'
    amqp :queue => 'test', :message => 'foo', :forget => true
  end
end

AMQP notes

The participant currently only makes use of direct exchanges. Possible future improvements might see use for topic and fanout exchanges as well.

The direct exchanges are always marked as durable by the participant, and messages are marked as persistent by default (see #RuoteAMQP)

Direct Known Subclasses

Participant

Instance Method Summary collapse

Constructor Details

#initialize(options) ⇒ ParticipantProxy

The following parameters are used in the process definition.

An options hash with the same keys to provide defaults is accepted at registration time (see above).

  • :queue => (string) The AMQP queue used by the remote participant. nil by default.

  • :forget => (bool) Whether the flow should block until the remote participant replies. false by default



121
122
123
124
125
126
127
128
129
130
131
132
133
 
# File 'lib/ruote-amqp/participant.rb', line 121

def initialize(options)

  RuoteAMQP.start!

  @options = {
    'queue' => nil,
    'forget' => false,
  }.merge(options.inject({}) { |h, (k, v)|
    h[k.to_s] = v; h
  })
    #
    # the inject is here to make sure that all options have String keys
end

Instance Method Details

#cancel(fei, flavour) ⇒ Object 



174
175
176
177
178
179
 
# File 'lib/ruote-amqp/participant.rb', line 174

def cancel(fei, flavour)
  #
  # TODO : sending a cancel item is not a bad idea, especially if the
  #        job done over the amqp fence lasts...
  #
end

#consume(workitem) ⇒ Object

Process the workitem at hand. By default the workitem will be published to the direct exchange specified in the queue workitem parameter. You can specify a message workitem parameter to have that sent instead of the workitem.



140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/ruote-amqp/participant.rb', line 140

def consume(workitem)

  target_queue = determine_queue(workitem)

  raise 'no queue specified (outbound delivery)' unless target_queue

  q = MQ.queue(target_queue, :durable => true)
  forget = determine_forget(workitem)

  opts = {
    :persistent => RuoteAMQP.use_persistent_messages?,
    :content_type => 'application/json' }

  if message = workitem.fields['message'] || workitem.params['message']

    forget = true # sending a message implies 'forget' => true

    q.publish(message, opts)

  else

    q.publish(encode_workitem(workitem), opts)
  end

  reply_to_engine(workitem) if forget
end

#do_not_thread ⇒ Object

The current AMQP (0.6.7) has 1 queue per thread. If you let the default “one thread per participant consume call” kick in, you’ll end up with 1 queue per consume call (and…)

So, by returning true here, we force the queue to be always the same.

Many thanks to github.com/weifeng365 for reporting this issue and suggesting the fix.

TODO : should we have something to close queues when the engine / worker

shuts down ?
or is it already covered in the #stop ?



194
195
196
197
 
# File 'lib/ruote-amqp/participant.rb', line 194

def do_not_thread

  true
end

#stop ⇒ Object

(Stops the underlying queue subscription)



169
170
171
172
 
# File 'lib/ruote-amqp/participant.rb', line 169

def stop

  RuoteAMQP.stop!
end