Message Queue Module

Introduction

The MessageQueue module provides a simple, lightweight message queueing mechanism
for SilverStripe applications. It supports the following features:

• queueing actions for long running process execution
• message sending between SilverStripe installations
• bi-directional interaction with external messaging systems ssuch as ApacheMQ.

Requirements

• Stomp.php if Stomp is being used (Note: experimental only)
• processOnShutdown option requires *nix or OS X (won't work on Windows)

Installation

Extract messagequeue into the base folder of your SilverStripe application. Default
configuration applies, documented below. Ensure its called "messagequeue".
In mysite/_config.php, put any code for setting the interface configuration
of the module.

Key Characteristics

The behaviour of the module is mostly determined by it's configuration, set by
the application. The configuration is set via the methods
MessageQueue::add_interface() and MessageQueue::remove_interface(). These are
used to set one or more named interfaces.

Each interface provides one or more named queues. Each interface also
specifies an implementation, a class that implements a message queue or
interfaces to an external component that is the message queue. The module
includes three implementations:

• SimpleDBMQ - this implements message queueing within SilverStripe, in the
  database. This implementation has no external dependencies.
• SimpleInterSSMQ - this implements sending messages to another SilverStripe
  installation over HTTP.
• StompMQ - this implements an interface to external messaging queueing
  components (such as ApacheMQ) using the Stomp protocol.

There are two primary operations:

• send - the application can send a message to a queue.
• receive - messages can be received from a queue, and delivered to the
  application.

Message Format

A message is encapsulated in a MessageFrame object, which contains:

• headers - a map of name => value pairs.
• body - the message body.

Send Behaviour

Messages are sent using MessageQueue::send(). This nominates a queue and the
message body. First, the interface to send through is determined from the queue
name. The selected interface is the first whose queueName option matches the
queue name provided. The queueName can be a regular expression, an array of
queue names, a single queue name, or null. If null, it will match the queue
name, so acts as a catch-all.

The interface configuration also includes an encoding. send() will encode the
message using this encoding. The default is "php-serialize", which calls that
PHP function to encode the message. The effect of this is that any PHP object
can be passed and received as a message, useful if the application is sending
messages to itself.

Once the message is encoded, it is sent via the specified implementation class.

Receive Behaviour

There are two distinct phases to receiving messages:

• Getting the messages from the queue implementation. Retrieval is delegated to
  the implementation class.
• Delivering the messages to the application. This is done centrally by the
  MessageQueue class so the behaviour is consistent.

Message receiving can be triggered in different ways depending on the
circumstance:

• via a cron job or other external trigger.
• in a sub-process initiated after php shutdown (so that events queued in the
  application can be processed near-simultaneously without introducing delay in
  the user interaction, and for buffered queues, messages to remote systems can
  be delivered outside the page request process)
• programmatically within the application

Messages are generally received in bulk, with options to limit this. Once a set
of messages is received via the implementation class, each message is decoded
according to the encoding option on the interface, and delivery is attempted.

There are 3 message delivery options:

• If the message is a PHP object with an execute() method, the method is
  executed.
• If the interface provides a callback, that is called with the message as a
  parameter.
• If the interface specifies that the application fetches its own messages, the
  messages are only delivered in response to the application requesting them
  (this option is for where the application's normal page request execution is
  to process the messages synchronously with the user interaction.)

Notes:

• Queue implementation classes guarantee that any set of messages retrieved are
  done atomically, so that multiple processes can process a queue but
  guaranteeing that each message delivery attempt is done once. A message is
  only successfully delivered once.

Output Buffering

When sending messages to a remote system, it is often beneficial to buffer the
outgoing messages and send them outside of the PHP request that is sending
them. Output buffering is very easy to configure, and involves setting up
another queue to act as the buffer. For example:

MessageQueue::add_interface("default", array(
	"queues" => array("remote"),
	"implementation" => "SimpleInterSSMQ",
	"implementation_options" => array(
		"remoteServer" => "http://myothersite.com/SimpleInterSSMQ_Accept"
	),
	"encoding" => "php_serialize",
	"send" => array(
		"buffer" => "remote_buffer",
		"onShutdown" => "flush"
	),
	"delivery" => array(
		"onerror" => array("log")
	)
));

MessageQueue::add_interface("buffer", array(
	"queues" => array("remote_buffer"),
	"implementation" => "SimpleDBMQ",
	"send" => array(
		"onShutdown" => "none"
	),
	"delivery" => array(
			 "onerror" => array("log")
	)
));

The main points are:

• The queue remote specifies a buffer in the send options. This is the name
  of the buffer queue.
• onShutdown specifies that the queue should be flushed on shutdown, which
  is done in another process.
• Creating another queue, remote_buffer, using the SimpleDBMQ interface. This
  queue is configured not to process on shutdown (it shouldn't be explicitly
  consumed)

When a message is sent to remote, the message is actually sent to the buffer
queue. When the remote queue is flushed, it reads back the messages queued on
the buffer queue, and at that point sends them via real configured interface
(in this case, SimpleInterSSMQ).

Exception Handling

If an exception occurs during delivery (i.e. during the execution of an object's
execute() method or the callback), the 'onerror' section of the interfaces
config determines what gets done. In general, this is an array of commands
which can include:

• log the error via SS_Log::log()
• drop the message
• re-queue the message, on the same queue (for retry), or onto another queue
  (e.g. might have a queue for errors)

Auto-executing Messages

As a convenience, a class MethodInvocationMessage is provided which encapsulates
a method call in one of the following forms:

• A static method call with parameters
• An instance method call to a DataObject instance with parameters
• An instance method call to an arbitrary object with parameters

When a message is received that is an object of this class, it will
automatically be executed, rather than delivered to the application via
callback. A further feature of this class is that user_errors are trapped and
thrown as exceptions, so user_errors are subject to the same exception handling
as real exceptions in the message processing engine.

In general, any class that is serializable and implements MessageExecutable can
be called this way. This is particularly useful for easy creation of actions or
commands to execute in a long running process. These messages are considered
"self-delivering", although are subject to exception processing.

Configuration Options and Examples

Default Configuration

The default configuration is:

MessageQueue::add_interface("default", array(
	"queues" => "/.*/",
	"implementation" => "SimpleDBMQ",
	"encoding" => "php_serialize",
	"send" => array(
		"onShutdown" => "all"
	),
	"delivery" => array(
		"onerror" => array(
			"log"
		)
	)
));

The effective behaviour is that messages sent to any queue will be processed on
PHP shutdown. (Note: if this option is set, messages sent from another PHP
shutdown function will not be consumed).

Example 1:

MessageQueue::send(
	"myqueue",
	new MethodInvocationMessage("MyClass", "someStatic", "p1", 2)
);

This will cause the static method MyClass::someStatic("p1", 2) to be called in
a sub-process that is initiated from PHP shutdown. Errors will be logged.

NOTE: If you don't want the default behaviour in your site, you must call
MessageQueue::remove_interface("default") before adding the interfaces you want
to use.

Multiple Queues

MessageQueue::add_interface("myinterface", array(
	"queues" => array("queue1", "queue2"),
	"implementation" => "SimpleDBMQ",
	"encoding" => "php_serialize",
	"delivery" => array(
		"onerror" => array(
			"log",
			"requeue" => "queue2"
		)
	)
));

MessageQueue::add_interface("default", array(
	"queues" => "/.*/",
	"implementation" => "SimpleDBMQ",
	"encoding" => "php_serialize",
	"send" => array(
		"onShutdown" => "all"
	),
	"delivery" => array(
		"onerror" => array(
			"log"
		)
)));

This configuration has two explicitly named queues, queue1 and queue2. They will
not be processed on shutdown, so MessageQueue_Consume must be explicitly called
to process messages received on these queues. Errors on either of these queues
will be logged and re-queued onto queue2. Messages sent to any other queue will
be handled by the second interface, which is processed on PHP shutdown.

SimpleInterSSMQ

This interface provides a simple way to send messages from one SilverStripe installation to another without requiring
any additional installed software. It works by the sender initiating an HTTP request to a controller at the destination
whch accepts messages.

An example configuration on the send is:

MessageQueue::add_interface("default", array(
	"queues" => array("mydest"),
		"implementation" => "SimpleInterSSMQ",
		"implementation_options" => array(
			"remoteServer" => "http://mydestination.com/SimpleInterSSMQ_Accept"
		),
		"encoding" => "php_serialize",
		"send" => array(
			"buffer" => "mydest_buffer",
			"onShutdown" => "flush"
		),
	"delivery" => array(
		"onerror" => array(
			"log"
		)
	)
));

MessageQueue::add_interface("buffer", array(
        "queues" => array("mydest_buffer"),
        "implementation" => "SimpleDBMQ",
        "delivery" => array(
		"onerror" => array("log")
	)
));

It sets up a queue called mydest. The implementation_options specify the remote accepting controller. This example
also specifies a buffer queue called mydest_buffer. When messages are sent to mydest, they are buffered into
mydest_buffer, and actually in a process initiated by PHP shutdown for better user performance.

On the destination, the following should be present in mysite/_config.php:
SimpleInterSSMQ_Accept::setEnabled(true);

This is required because SimplerInterSSMQ_Accept controller is disabled by default for security purposes.

To send a message from the source, it is a simple message send:

	MessageQueue::send("mydest", $someObject);

or even a self-invoking message:

	MessageQueue::send("mydest", new MethodInvocationMessage("SomeClass", "someMethod", $parameter));

ApacheMQ

(This example is incomplete. We need to document how to pass authentication
details thru, and how to use the durable clients feature of Stomp.)

	MessageQueue::add_interface("myinterface", array(
		"queues" => array(
			"stompqueue1",
			"stompqueue2"
		),
		"implementation" => "StompMQ",
		"encoding" => "raw",
		"delivery" => array(
			"onerror" => array(
				"log",
				"requeue" => "queue2"
			)
		)
	));

	MessageQueue::add_interface("default", array(
		"queues" => "background",
		"implementation" => "SimpleDBMQ",
		"encoding" => "php_serialize",
		"send" => array(
			"onShutdown" => "all"
		),
		"delivery" => array(
			"onerror" => array(
					"log"
			)
		)
	));

In this example, two queues "stompqueue1" and "stompqueue2" are defined, and
message processing is handled through the StompMQ class. These queue names are
passed directly to Stomp, so they are the queue names identified externally, not
just within the SilverStripe application.

The second interface provides the "background" queue, with internal queuing and
processing on shutdown as before.

Error Handling Options

The following configuration snippet shows the currently available forms for
processing delivery exceptions.

	...
	"delivery" => array(
		"log",
		"requeue",
		"requeue" => "errorQueue",
		"callback" => array("MyClass", "method"),
		"drop"
	),
	...

It is a list of commands of these forms, so more than one action can be taken.

• "log" logs the message via SS_Log::log. Note that SS_Log has options for
  where errors are logged, including notification email. This needs to be
  configured separately in the application.
• "requeue" puts the message back in the same queue for later processing.
  (existing queue behaviour will exclude the message being executed again in
  the same queue consumption call)
• "requeue" => "queue" puts the message onto the named queue for later
  processing.
• "callback" => $method invokes the specified method. $method should be a
  valid callback definition. The callback function is passed two parameters,
  the exception object and the messageframe that failed to be delivered.
• "drop" does nothing. If used alone, the exceptioned message will be dropped.

Specifying a Callback for Delivery

	...
	"delivery" => array(
		"callback" => array("MyClass", "method")
	),
	...

With this option, any message received that does not implement
MethodExecutable will be passed to the specified callback function. The
value is a method specifier for call_user_func_array, so can identify a static
function by supplying the class name and a static method name. The signature
of the callback is function callback($msgFrame, $config)

It is passed the incoming message frame (decoded) and the configuration of
the interface from which it was received.

Custom Shutdown Handling

If you want to process messages on shutdown, but your application requires a shutdown
function which queues messages, there is in an issue because the shutdown functions
are executed by PHP in the order in which they are executed. A way to to handle this
is as follows. In the interface configuration, use the registerShutdown property.

	...
	"send" => array(
		"onShutdown" => "all",
		"registerShutdown" => false,
	)
	...

Then in your custom shutdown function do the following:

	
	MessageQueue::send("myqueue", new MethodInvocationMessage("MyClass", "my_method"));
	

	// force MessageQueue to spawn the process that handles the messages as it
	// normally would on shutdown.
	MessageQueue::consume_on_shutdown();

Retriggering Queue Processing

When messages are sent on shutdown, the default behaviour is to initiate a process
that sends all messages in the queue. Sometimes you might want to limit the number
of messages sent in a single process, but do want to send all the messages. For example,
if you have to perform a memory-intensive operation on a large number of objects, attempting
to send all the messages in a single process may cause PHP to run out of memory. Message queue provides
options retrigger and onShutdownMessageLimit that can be used to work around this.
onShutdownMessageLimit sets a limit on the number of items in the queue that are sent
by a single PHP process executed asynchronously. retrigger causes the asynchronous process
to initiate a further process if there are still unsent messages in the queue its processing.

To configure this behaviour, do the following:

	...
	"send" => array(
		"onShutdown" => "all",
		"retrigger" => "yes",                   // on consume, retrigger if there are more items
		"onShutdownMessageLimit" => "1"         // one message per async process
	)
	...

Notes:

• in the general case, the initial shutdown will result in a "chain" of sychronous
  PHP processes that will in time clear all messages from the queue.
• this does not guarantee that there is only one consumer of the queue. If two separate
  HTTP requests both send messages to the queue, it is possible that two processes are both
  sending the messages (however, a given message will only be executed by one of them.) Care
  must be taken if the processes can adversely interact.

Queue Syntaxes

The "queues" option in an interface configuration can be one of the following
forms:

• "queues" => "myqueue" specifies a single named queue.
• "queues" => array("queue1", "queue2") specifies a list of named queues.
• `"queues" => "/.*AppQ$/" specifies a regular expression to match against
  queue names. The regular expression must start and finish with
  forward slash. In this example, any queue name ending with
  AppQ, such as MyAppQ, will be matched against the interface.

Specifying Requeuing for Delivery

	...
	"delivery" => array(
		"requeue" => array(
			"queue" => "otherQueue",
			"immediate" => true
		)
	),
	...

With this option, you can specify that when delivery is attempted, it is put
into another queue. If the immediate option is set, the delivery of the message
on that queue is attempted in-process. If immediate is false, no further
immediate attempt is made to deliver the message from the other queue - it will
be delivered according to the deliver execution rules of that queue.

Initiating Message Queue Processing

There are two distinct processes that can be initiated on a queue, as follows:

• flush has effect only on a buffered queue, and will cause the buffered
  messages to be sent to the real destination.
• consume will cause messages on the queue to be retrieved and delivered.

Typically, one or both of these actions can be executed on a queue.

On PHP Shutdown

To initiate queue processing on the PHP shutdown of the process that initiated
the send, you need to set the onShutdown option on the interface
configuration. onShutdown can be a single option as a string, or an array of
option strings. The valid options are:

• flush - invokes flush of queue.
• consume - invokes consumption of the queue.
• all - flush and consume
• none - do neither - no process will be invoked.

Flush is invoked before consume.

By default, this calls the MessageQueue_Process controller in a sub-process,
using 'sake'. This process can continue to execute after the main request
process has finished.

In some environments (particularly where there are multiple PHP binaries
that are not compiled the same - MacOS X built-in vs MAMP is a classic
example), 'sake' make not work. If this is the case, you can call the
following in mysite/_config.php:

	MessageQueue::set_onshutdown_option("phppath", $pathToPhp);

If this is set, the provided php binary is used instead of sake in the
sub-process.

Note: this may vary between development, testing and production environments.

Externally Using Sake

Messages in a queue can be processed using the command:

	sake MessageQueue_Process "queue=myqueue&actions=all"

This will flush and then consume all messages on myqueue, and delivering them
to the application.

You can limit the number of entries processed:

	sake MessageQueue_Process "queue=myqueue&limit=10&actions=consume"

The actions query field can be a comma-separated list containing flush,
consume, all or none.

You can schedule queue consumption using cron.

Externally Using wget

In environments where there is no external php binary (e.g. only mod_php), you
may need to use wget to initiate the call to the MessageQueue_Consume
controller.

To Do

• Complete StompMQ and test it. Specific functions not supported at this stage
  includes authentication.
• More test cases
• Option on queue consumption to process messages atomically. That is, do one
  message at a time, so that if there is a failure, we haven't lost a whole
  bunch of messages. We can also use transactions in the implementor class so
  that a complete failure can still leave the message there. Upside: more
  robust. Downside: worse performance (quite a bit more overhead)
• Options for re-try messages with exceptions. May be frequency of re-tries,
  max number of re-tries. Need to either: use headers to hold info (but cannot
  be guaranteed across different implementers); or push it to the implementer
  layer to handle.

Known Issues

Message Consumption on PHP Shutdown Issues on MacOS X w/MAMP

If the message queue appears to clearing on shutdown, but the messages are
not being delivered (callback not being executed for example), enable
debugging. If you see this message:

Symbol not found: __cg_jpeg_resync_to_restart

You need to ensure that /Applications/MAMP/Library/bin/envvars contains:

	DYLD_LIBRARY_PATH="/Applications/MAMP/Library/lib:$DYLD_LIBRARY_PATH"
	export DYLD_FALLBACK_LIBRARY_PATH=/Applications/MAMP/Library/lib

Diagnosing Message Queue Processing on PHP Shutdown

By default when a process is initiated to clear a queue on PHP shutdown, the
process redirects output to /dev/null. To assist in debugging these processes,
call MessageQueue::set_debugging can be called to set a directory to write log
files to, and both stdout and stderr are redirected to real files in that
directory.

DataObject Sent Remotely

Currently when a DataObject is sent as message body, it is serialised as a
DataObject with a specified class and ID. When sent to a remote system there
cannot be a guarantee that the class exists, or there is an object of that
class and ID. Add an option to the configuration, possibly as a different
serialisation, so that messages are serialised by value, not reference, for
remote sends.

To Do

• Allow messages to be received from a buffered queue, particularly remotely.
• SimpleInterSSMQ to implement pull behaviour, not just push.
• Example in docs of using SimpleInterSSMQ to capture onPublish and
  have the changes re-published on a remote site.
• Provide for a single consumer of a queue, so that it can be guaranteed
  that no two messages are being sent simultaneously by different processes.