THIS PACKAGE IS PRIMARILY INTENDED FOR INTERNAL/PRIVATE USE IN OWN PROJECTS. IF IT MEETS YOUR NEEDS, FEEL FREE TO USE IT, BUT IN CASE OF ANY MODIFICATION REQUESTS, I WILL CONSIDER MY OWN NEEDS FIRST.
It is still in a very early development phase, so I do not really recommend using it for now, because anything can change on it at any time and previous functions may break.
The package is part of the rabbitmq-multiverse.
This package is an intermediate layer between RabbitMQ and Laravel Queue.
The package is based on vladimir-yuldashev/laravel-queue-rabbitmq package, which adds RabbitMQ as a queue driver to Laravel.
This package extends the functionality of the original package by adding the ability to send and receive events and RPC calls through RabbitMQ messages.
Since the microservice architecture has become very popular, I needed a library that provides the possibility of communicating with services written in different programming languages or frameworks.
Laravel has a powerful queue system, but it is a closed Laravel-only system. This package allows you to communicate through messages between Laravel and/or other non-Laravel microservices.
On the top of simple JSON messages, utilizes the Laravel Queue and Event system, which perfectly does the rest of the job.
You can install this package via composer using this command:
composer require djereg/laravel-rabbitmqThe package will automatically register itself.
The configuration is done through environment variables.
# Set the queue connection to rabbitmq
QUEUE_CONNECTION=rabbitmq
RABBITMQ_HOST=rabbitmq
RABBITMQ_PORT=5672
RABBITMQ_USER=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_VHOST=/
RABBITMQ_QUEUE=queue-name
RABBITMQ_EXCHANGE_NAME=exchange-name
RABBITMQ_EXCHANGE_TYPE=directTo start the consumer, just run the following command:
php artisan rabbitmq:consumeProvides an event based asynchronous communication between services.
Create an event class that extends MessagePublishEvent provided by this package.
# app/Events/UserCreated.php
namespace App\Events;
use Djereg\Laravel\RabbitMQ\Events\MessagePublishEvent;
class UserCreated extends MessagePublishEvent
{
// Set the event name
protected string $event = 'user.created';
public function __construct(private User $user)
{
$this->user = $user;
}
// Create a payload method that returns the data to be sent
public function payload(): array
{
return [
'user_id' => $this->user->id,
];
}
}And after just dispatch the event like any other Laravel event.
event(new UserCreated($user));Create an event listener that extends MessageEventListener provided by this package.
The working mechanism is a little bit different from the Laravel event listeners.
First, you have to specify the events you want to listen to in the $listen property.
Next, instead of public handle() method, you have to define the onEvent() method.
This is because the handle() method is already used under the hood by the base MessageEventListener class.
# app/Listeners/NotifyUser.php
namespace App\Listeners;
use Djereg\Laravel\RabbitMQ\Listeners\MessageEventListener;
class NotifyUser extends MessageEventListener {
// Specify the events you want to listen to.
// You can listen to multiple events by adding them to the array.
public static array $listen = [
'user.created',
// 'user.updated',
// 'user.deleted',
// etc
];
// The method that will be called when the event is received.
// The event object is passed as an argument containing the event name and payload.
protected function onEvent(MessageEvent $event): void {
}
// You can also define separate methods for each event.
// The method name must be in the format on{EventName} where {EventName}
// is the StudlyCase format of the event defined in the $listen property.
// When both methods are defined, the on{EventName} method will be called.
protected function onUserCreated(MessageEvent $event): void {
//
}
// If no on{EventName} or onEvent method is defined, an exception will be thrown.
}Since the event listeners are processed synchronously by default, if an error occurs, the job will fail and will be retried by the retry mechanism, if it is enabled and configured.
If multiple listeners listening to the same event, the processing will stop at the first listener that throws an error and the rest of the listeners will not be processed.
You have multiple options to prevent this behavior:
- Run the listener in the try-catch block and handle the error in the listener.
- Put the listeners to the queue and process the events asynchronously. This way the failed listener (job) will not block the processing of the other listeners.
The events are processed synchronously by default, but you can process them asynchronously by implementing the
ShouldQueue interface.
# app/Listeners/NotifyUser.php
namespace App\Listeners;
use Djereg\Laravel\RabbitMQ\Listeners\MessageEventListener;
class NotifyUser extends MessageEventListener implements ShouldQueue
{
// Listener content
}When starting the consumer, it automatically creates the exchange and the queue if they do not exist,
but to register the events listening to, you have to modify the EventServiceProvider to extend the
EventServiceProvider provided by this package.
In Laravel 11 the EventServiceProvider does not exist by default, so you have to create and register it manually.
See the example below.
# app/Providers/EventServiceProvider.php
namespace App\Providers;
use Djereg\Laravel\RabbitMQ\Providers\EventServiceProvider as ServiceProvider;
class EventServiceProvider extends ServiceProvider
{
// Provider content
}The service provider discovers all event listeners from the app/Listeners directory and the routing keys will be bound
automatically when the consumer started.
A synchronous-like communication between services.
Uses the JSON-RPC 2.0 protocol for communication.
To call remote procedures, you have to create an instance of the Client class and inject it into the service where you
want to use it.
The best way is to do it in a service provider.
# app/Providers/AppServiceProvider.php
namespace App\Providers;
use Djereg\Laravel\RabbitMQ\Services\Client;
class AppServiceProvider extends ServiceProvider {
public function register(): void
{
$this->app->singleton(UserService::class, function($app) {
// Instantiate the client with the remote service name and the queue connection
$client = new Client('users', $app['queue.connection'])
// Inject the client into the service
return new UserService($client);
});
}
}Anyway, you can create the client instance wherever you want, but remember to pass the queue connection as the second argument.
$client = new Client('users', app('queue.connection'));In the service where you injected the client, you can call the remote procedures.
# app/Services/UserService.php
namespace App\Services;
use Djereg\Laravel\RabbitMQ\Services\Client;
class UserService {
public function __construct(private Client $users) {}
public function getUser(int $id): mixed
{
// Call the remote procedure
$user = $this->users->call('get', ['id' => $id]);
// Process the response and return it
}
}Register the ProcedureServiceProvider provided by this package.
The service provider will automatically discover all procedures in the app/Procedures directory by default.
The automatic discovery runs only when the application is started in console mode.
If you want to customize the service provider, you can create your own that extends the ProcedureServiceProvider class
provided by this package and register it.
# app/Providers/ProcedureServiceProvider.php
namespace App\Providers;
use Djereg\Laravel\RabbitMQ\Providers\ProcedureServiceProvider as ServiceProvider;
class ProcedureServiceProvider extends ServiceProvider
{
//
}When the service provider is registered, you can create the procedures in the app/Procedures directory.
Create a class that extends the Procedure class and define the method property with the name of the procedure.
# app/Procedures/GetUser.php
namespace App\Procedures;
use Djereg\Laravel\RabbitMQ\Procedures\Procedure;
class GetUser extends Procedure {
// Set the procedure name that will be called by the client
public static string $name = 'get';
public function __invoke(int $id): mixed
{
// Get the user from the database and return it
}
// OR
public function handle(int $id): mixed
{
// Get the user from the database and return it
}
}You can define the __invoke() or handle() method to process the procedure call.
If both methods are defined, an exception will be thrown as multiple handlers are not allowed for one procedure.
The same applies to multiple procedure classes with the same name.
When a procedure call message is received, the request body is passed to datto/php-json-rpc server component, which processes the request and calls the matching procedure, and finally returns the response object, which is sent back to the requester.
The Laravel Queue is also supported by this package. You can send jobs to the queue and the consumer will process them as the original Laravel queue worker.
The package emits events during the message processing.
Dispatched before the message is being published.
use Djereg\Laravel\RabbitMQ\Events\MessagePublishing;Dispatched after the message is published.
use Djereg\Laravel\RabbitMQ\Events\MessagePublished;Dispatched before the message is being processed.
use Djereg\Laravel\RabbitMQ\Events\MessageReceived;Dispatched after the message is processed.
use Djereg\Laravel\RabbitMQ\Events\MessageProcessed;- NO TESTS! I know, I know. I will write them soon.