README

This library provides communication between multiple services. Supports sync and async communications.

This library uses:

https://github.com/needle-project/laravel-rabbitmq - Library for supports AMPQ. Some class was overrides for support current library logic.
https://github.com/laravel/horizon - code base of supervisor part

Base logic

Base logic builds on package structure. This structure very similar to .Net MassTransit packages. Package contains:

package meta
message type
message

Algorithm:

Sender creates package with message
Sender submit package to consumer
Consumer defines requested package by message type and deserialize package
Consumer executes logic with received package

Synchronous communications

Sender creates package with message
Sender submit package to consumer with synchronous method
- Library serialized package to json
- Library submits http request to consumer host
- Consumer process some logic with received package
- Consumer returns new package with response
Sender receives response package from consumer

Asynchronous communications

Asynchronous communication provides by RabbitMQ

Sender creates package with message
Sender submit package to consumer with asynchronous method
- Library serialized package to json
- Library submits package to the RabbitMQ exchange
- RabbitMQ submits received package to queue
Consumer listens RabbitMQ queue
Consumer receives package from queue
Consumer process some logic with received package

Installation

composer require smskin/laravel-service-bus
php artisan vendor:publish --provider="SMSkin\ServiceBus\Providers\ServiceProvider"
Configure RabbitMQ connection in service-bus.php config file
Run php artisan service-bus:setup from initialize RabbitMQ exchanges and queues
Add service bus consumer command to the supervisor conf

Supervisor config to provides service bus consumer

[program:laravel-service-bus]
process_name=%(program_name)s
command=php /var/www/html/artisan service-bus:supervisor
autostart=true
autorestart=true
user=www-data
group=www-data
redirect_stderr=true
numprocs=1
stderr_logfile=/dev/stderr
stdout_logfile=/dev/stdout

Artisan commands

service-bus:setup - command for initialize RabbitMQ exchanges and queues
service-bus:supervisor - service bus supervisor. Start processes for consume packages from RabbitMQ queues
service-bus:delete-all - command deletes exchanges and queues from RabbitMQ
service-bus:list - command lists exchanges and queues
service-bus:publish - command for submit test message to RabbitMQ exchange (for tests)
service-bus:consume - consumer command for receive packages from RabbitMQ queue (uses by supervisor command)

Service bus supervisor

Supervisor command run consumer process for each registered consumer (Config file service-bus.php, section supervisor)

Overriding default enums

You can create new enum and override calling it in service-bus.php config file. It provides extend the default consumers and exchanges.

For example

Create Exchanges enum
Extends it from base SMSkin\ServiceBus\Enums\Exchanges
Change the items method for provide new exchanges
Replace default enum in service-bus.php config file (line 14)

Classes

Package - base exchange data package
Message - package payload (DTO)
Processor - class, that running on receive incoming package. Can return new Package in synchronous exchange mode.

Exchanges

id - internal id
connection - id of connection (SMSkin\ServiceBus\Enums\Connections)
rabbitMqName - RabbitMQ name of exchange
attributes - RabbitMQ exchange attributes
- exchangeType (exchange_type)
- passive
- durable
- autoDelete (auto_delete)
- internal
- nowait
- throwExceptionOnRedeclare (throw_exception_on_redeclare)
- throwExceptionOnBindFail (throw_exception_on_bind_fail)

Queues

id - internal id
connection - id of connection (SMSkin\ServiceBus\Enums\Connections)
rabbitMqName - RabbitMQ name of queue
attributes
- passive
- durable
- autoDelete (auto_delete)
- internal
- nowait
- exclusive
- bind - array of binds
  - exchange - id of exchange (SMSkin\ServiceBus\Enums\Exchanges)
  - routing_key
- arguments
  - x-max-priority - maximal message priority

Publishers

id - internal id
exchange - id of exchange (SMSkin\ServiceBus\Enums\Exchanges)

Consumers

id - internal id
queue - id of queue (SMSkin\ServiceBus\Enums\Queues)
prefetchCount

Hosts

id - internal id
host - url of consumer host (for provide synchronous service bus)

Exchange package

Package must provide links to processor and message classes. Package register in Packages enum ( SMSkin\ServiceBus\Enums\Packages - can be overrides). Exchange packages has very simple structure, that support reproduce this logic in any languages.

Example of serialized package:

{
    "messageId": "e5e471ae-62e3-46c5-93fd-6dd589c32748",
    "correlationId": "2712f617-4d55-4851-ad2c-5c841224e251",
    "conversationId": null,
    "initiatorId": null,
    "sourceAddress": null,
    "destinationAddress": null,
    "messageType": [
        "urn:message:TEST_ASYNC_LOCAL"
    ],
    "message": {
        "name": "Sergey"
    },
    "sentTime": "2022-06-07T10:19:15.152620Z",
    "host": null
}

Example of package:

<?php

namespace App\Modules\ServiceBus\Packages;

use App\Modules\ServiceBus\Enums\Packages;
use App\Modules\ServiceBus\Packages\Messages\TestMessage;
use App\Modules\ServiceBus\Packages\Processors\TestMessageProcessor;
use SMSkin\ServiceBus\Packages\BasePackage;

class TestMessagePackage extends BasePackage
{
    public function package(): string
    {
        return Packages::TEST_SYNC;
    }

    protected function messageClass(): string
    {
        return TestMessage::class;
    }

    public function getProcessorClass(): string
    {
        return TestMessageProcessor::class;
    }
}

Example of processor class:

<?php

namespace App\Modules\ServiceBus\Packages\Processors;

use App\Modules\ServiceBus\Packages\TestMessagePackage;
use Illuminate\Support\Facades\Log;
use SMSkin\ServiceBus\Packages\BasePackage;
use SMSkin\ServiceBus\Packages\Processors\BaseProcessor;

class TestMessageProcessor extends BaseProcessor
{
    public function __construct(protected TestMessagePackage|BasePackage $package)
    {
        parent::__construct($package);
    }

    public function execute(): ?BasePackage
    {
        Log::debug('Received package', $this->package->toArray());
        return null;
    }
}

Example of message class:

<?php

namespace App\Modules\ServiceBus\Packages\Messages;

use SMSkin\ServiceBus\Packages\Messages\BaseMessage;

class TestMessage extends BaseMessage
{
    public string $name;

    public function fromArray(array $data): static
    {
        $this->name = $data['name'];
        return $this;
    }

    public function toArray(): array
    {
        return [
            'name' => $this->name
        ];
    }

    /**
     * @param string $name
     * @return TestMessage
     */
    public function setName(string $name): TestMessage
    {
        $this->name = $name;
        return $this;
    }
}

Example of registration package in Packages enum

<?php

namespace App\Modules\ServiceBus\Enums;

use App\Modules\ServiceBus\Packages\TestMessagePackage;
use Illuminate\Support\Collection;
use SMSkin\ServiceBus\Enums\Models\PackageItem;

class Packages extends \SMSkin\ServiceBus\Enums\Packages
{
    public const TEST_ASYNC_LOCAL = 'TEST_ASYNC_LOCAL';

    /**
     * @return Collection<PackageItem>
     */
    protected static function getItems(): Collection
    {
        return parent::getItems()->merge([
            (new PackageItem)
                ->setId(self::TEST_ASYNC_LOCAL)
                ->setClass(TestMessagePackage::class)
        ]);
    }
}

Submitting

Example of submitting synchronous command

$result = (new ServiceBus)->syncPublish(
            (new SyncPublishRequest)
                ->setHost(Hosts::LOCALHOST)
                ->setPackage((new TestSyncMessagePackage)
                    ->setMessageType(Packages::TEST_ASYNC_LOCAL)
                    ->setMessageId(Str::uuid()->toString())
                    ->setCorrelationId(Str::uuid()->toString())
                    ->setSentTime(now())
                    ->setMessage(
                        (new TestMessage)
                            ->setString1('a1')
                            ->setString2('b2')
                    ))
        );
dd($result);

Example of submitting asynchronous command

(new ServiceBus)->asyncPublish(
            (new AsyncPublishRequest)
                ->setPublisher(\App\Modules\ServiceBus\Enums\Publishers::TEST_LOCAL)
                ->setRoutingKey('*')
                ->setPackage(
                    (new TestMessagePackage())
                        ->setPackage(\App\Modules\ServiceBus\Enums\Packages::TEST_ASYNC_LOCAL)
                        ->setMessageId(Str::uuid()->toString())
                        ->setCorrelationId(Str::uuid()->toString())
                        ->setSentTime(now())
                        ->setMessage(
                            (new \App\Modules\ServiceBus\Packages\Messages\TestMessage())
                                ->setName('Sergey')
                        )
                )
                ->setProperties([
                    'priority' => 0
                ])
        );

priority - priority of the message

smskin / laravel-service-bus

Maintainers

Details