This library provides a flexible abstraction layer for working with queues.
It can be installed in whichever way you prefer, but we recommend Composer.
~$ composer require graze/queue
Queue operations center around lists of Message objects. Whether you're sending one or multiple Messages, it's always an array. Workers work only on one Message object at a time, whether a list of one or multiple is received from the queue.
useAws\Sqs\SqsClient;
useGraze\Queue\Adapter\SqsAdapter;
useGraze\Queue\Client;
useGraze\Queue\Message\MessageInterface;
$client = newClient(newSqsAdapter(newSqsClient([
'region' => 'us-east-1',
'version' => '2012-11-05',
'credentials' => [
'key' => 'ive_got_the_key',
'secret' => 'ive_got_the_secret'
],
]), 'queue_name'));
// Producer$client->send([
$client->create('foo'),
]);
// Consumer$client->receive(function (MessageInterface$msg) {
var_dump($msg->getBody());
var_dump($msg->getMetadata()->getAll());
});The Adapter object is used to fulfill the low level requests to the queue provider.
Currently supported queue providers are:
The Handler object is used to execute worker callables with a list of received messages and handle Acknowledgement.
The current handlers are:
- Batch Acknowledgement to acknowledge batches
- Eager Acknowledgement to acknowledge immediately after work
- Null Acknowledgement for development
useGraze\Queue\Client;
useGraze\Queue\Adapter\ArrayAdapter;
useGraze\Queue\Handler\BatchAcknowledgementHandler;
useGraze\Queue\Message\MessageInterface;
// Create client with the Batch Acknowledgement Handler.$client = newClient(newArrayAdapter(), [
'handler' => newBatchAcknowledgementHandler(),
]);
// Receive a maximum of 10 messages.$client->receive(function (MessageInterface$message) {
// Do some work.
}, 10);Polling a queue is supported by passing null as the limit argument to the
receive method. The second argument given to your worker is a Closure you
should use to stop polling when you're finished working. Make sure you use a
handler that acknowledges work effectively too!
Note that the individual Adapter objects may decide to stop polling at any time. A likely scenario where it may stop would be if the queue is of finite length and all possible messages have been received.
useGraze\Queue\Client;
useGraze\Queue\Adapter\ArrayAdapter;
useGraze\Queue\Handler\BatchAcknowledgementHandler;
useGraze\Queue\Message\MessageInterface;
// Create client with the Batch Acknowledgement Handler.$client = newClient(newArrayAdapter(), [
'handler' => newBatchAcknowledgeHandler(100), // Acknowledge after 100 messages.
]);
// Poll until `$done()` is called.$client->receive(function (MessageInterface$message, Closure$done) {
// Do some work.// You should always define a break condition (i.e. timeout, expired session, etc).if ($breakCondition) $done();
}, null);The content of this library is released under the MIT License by Nature Delivered Ltd.
You can find a copy of this license in LICENSE or at http://opensource.org/licenses/mit.