Inteve\MessageQueue

Simple message queue

Installation

Download a latest package or use Composer:

composer require inteve/message-queue

Inteve\MessageQueue requires PHP 8.0 or later.

Usage

$exceptionHandler = function (\Throwable $e) {
	\Tracy\Debugger::log($e, \Tracy\Debugger::EXCEPTION);
};
$dateTimeFactory = new MyDateTimeFactory; // implementation of Phig\DateTimeFactory
$adapter = new MemoryAdapter($exceptionHandler);
$manager = new Manager($adapter, $dateTimeFactory);

Insert message

$manager->create(
	queue: 'name-of-queue',
	data: [
		'field' => 'value',
		'field2' => 'value2',
	]
);

Consume message

Process message from given queue:

$manager->fetch(
	queue: 'name-of-queue',
	handler: function (array $data) {
		$data['field'];
		$data['field2'];
	}
);

Process messages from ALL queues (logs missing handlers via $exceptionHandler):

$message = $manager->multiFetch(
	handlers: [
		'name-of-queue' => function (array $data) {
			$data['field'];
			$data['field2'];
		}
		'name-of-queue-B' => function (array $data) {
			$data['field'];
			$data['field2'];
		}
	],
	limit: 10 // number of messages processed in multiFetch() call
);

Process messages from SPECIFIC queues:

$message = $manager->multiFetch(
	handlers: [
		'name-of-queue' => function (array $data) {
			$data['field'];
			$data['field2'];
		}
		'name-of-queue-B' => function (array $data) {
			$data['field'];
			$data['field2'];
		}
	],
	limit: 10, // number of messages processed in multiFetch() call
	fetchFrom: $mananger::FROM_LISTED_ONLY
);

---

License: New BSD License
Author: Jan Pecha, https://www.janpecha.cz/