PHP/LaravelRabbitMQ
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added PublishEvent listener connection and queue configuration feature

Browse Source
This commit is contained in:
Viktor Smagin
2025-12-31 16:39:48 +04:00
parent 7ae585ae29
commit 181d3de9f7
11 changed files with 347 additions and 319 deletions
Download Patch File Download Diff File Expand all files Collapse all files

442
README.md
View File

@@ -1,14 +1,9 @@
# Laravel RabbitMQ Events
A package for working with RabbitMQ messages as Laravel events. Automatically serialize and publish events to RabbitMQ, and consume messages from queues by converting them back into events.
A package for working with RabbitMQ messages as Laravel events.
## Features
**Automatic event serialization** — Events implementing the `Broadcast` interface are automatically serialized and sent to RabbitMQ
**Message consumption** — Command to consume messages from RabbitMQ with automatic deserialization and event dispatching
**Default bindings** — The `BroadcastEvent` trait sets standard connection parameters for an event
**Flexible configuration** — Support for multiple connections and parameterization for each event
**Microservices architecture** — Ideal for data exchange between services
Automatically serialize and publish events to RabbitMQ, and consume
messages from queues by converting them back into events.
## Requirements
@@ -18,97 +13,28 @@ A package for working with RabbitMQ messages as Laravel events. Automatically se
## Installation
Install via Composer:
1. Install via Composer:
```bash
composer require diffhead/laravel-rabbitmq
```
The package will be automatically registered thanks to Laravel Service Provider Discovery.
## Configuration
Publish the configuration file:
```bash
php artisan vendor:publish --provider="Diffhead\PHP\LaravelRabbitMQ\ServiceProvider"
```
ThisEnvironment Variables
```env
# RabbitMQ Connection
RABBITMQ_HOST=localhost
RABBITMQ_PORT=5672
RABBITMQ_USER=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_VHOST=/
# Default event parameters
RABBITMQ_EVENT_CONNECTION=default
RABBITMQ_EVENT_QUEUE=default
RABBITMQ_EVENT_EXCHANGE=amq.direct
RABBITMQ_EVENT_EXCHANGE_TYPE=direct
RABBITMQ_EVENT_EXCHANGE_IS_DEFAULT=true
RABBITMQ_EVENT_ROUTING_KEY=
```
### Configuration Example
2. Register package's service provider
```php
// app/providers.php
return [
'connections' => [
'default' => [
'host' => env('RABBITMQ_HOST', 'localhost'),
'port' => env('RABBITMQ_PORT', 5672),
'user' => env('RABBITMQ_USER', 'guest'),
'password' => env('RABBITMQ_PASSWORD', 'guest'),
'vhost' => env('RABBITMQ_VHOST', '/'),
],
'secondary' => [
'host' => env('RABBITMQ_SECONDARY_HOST', 'localhost'),
'port' => env('RABBITMQ_SECONDARY_PORT', 5672),
'user' => env('RABBITMQ_SECONDARY_USER', 'guest'),
'password' => env('RABBITMQ_SECONDARY_PASSWORD', 'guest'),
'vhost' => env('RABBITMQ_SECONDARY_VHOST', '/'),
]
],
'message' => [
'serializer' => \Diffhead\PHP\LaravelRabbitMQ\Service\Serializer::class,
'unserializer' => \Diffhead\PHP\LaravelRabbitMQ\Service\Unserializer::class,
],
'event' => [
'defaults' => [
'connection' => env('RABBITMQ_EVENT_CONNECTION', 'default'),
'queue' => env('RABBITMQ_EVENT_QUEUE', 'default'),
'exchange' => env('RABBITMQ_EVENT_EXCHANGE', 'amq.direct'),
'exchange_type' => env('RABBITMQ_EVENT_EXCHANGE_TYPE', 'direct'),
'exchange_is_default' => (bool) env('RABBITMQ_EVENT_EXCHANGE_IS_DEFAULT', true),
'routing_key' => env('RABBITMQ_EVENT_ROUTING_KEY', ''),
],
'mapper' => \Diffhead\PHP\LaravelRabbitMQ\Service\EventMapper::class,
'map' => [
/**
* Map events to queues and routing keys
*/
\App\Events\User\UserCreated::class => [
'queues' => ['portal.users'],
'routing_keys' => ['user.created'],
],
\App\Events\Meeting\MeetingCreated::class => [
'queues' => ['portal.meetings'],
'routing_keys' => ['meeting.created'],
],
],
]
\Diffhead\PHP\LaravelRabbitMQ\ServiceProvider::class
];
```
## Usage
### 1. Creating an Event for Publishing to RabbitMQ
### 1. Creating regular events for publishing to a RabbitMQ
Create an event that implements the `Broadcast` interface:
Create an event that implements the `Broadcast` interface. `Broadcast` interface
extends `JsonSerializable` then your event should implements `jsonSerialize` method.
```php
namespace App\Events;
@@ -140,18 +66,28 @@ class UserCreated implements Broadcast
#### Event Parameters
When implementing the `Broadcast` interface, you must define the following methods:
When implementing the `Broadcast` interface, you must define the following methods
for targeting RabbitMQ message:
- `getConnection(): string` — RabbitMQ connection name
- `getConnection(): string` — connection name defined in configuration
- `getQueue(): string` — Queue name
- `getExchange(): string` — Exchange name
- `getExchangeType(): string` — Exchange type (direct, topic, fanout, headers)
- `getExchangeIsDefault(): bool` — Whether to use the default exchange
- `getRoutingKey(): string` — Routing key for the message
#### Using the BroadcastEvent Trait
#### Using the BroadcastEvent trait
The `BroadcastEvent` trait provides implementations of all methods using default parameters from configuration:
The `BroadcastEvent` trait provides implementations of all methods using default parameters from configuration. You can directly override method's return value for
customizing concrete event or override this behaviour globally by changing
following env variables:
* `RABBITMQ_EVENT_CONNECTION=default`
* `RABBITMQ_EVENT_QUEUE=default`
* `RABBITMQ_EVENT_EXCHANGE=amq.direct`
* `RABBITMQ_EVENT_EXCHANGE_TYPE=direct`
* `RABBITMQ_EVENT_EXCHANGE_IS_DEFAULT=true`
* `RABBITMQ_EVENT_ROUTING_KEY=`
```php
namespace App\Events;
@@ -162,19 +98,6 @@ use Diffhead\PHP\LaravelRabbitMQ\Trait\BroadcastEvent;
class UserCreated implements Broadcast
{
use BroadcastEvent;
public function __construct(
public int $userId,
public string $email,
) {}
public function jsonSerialize(): array
{
return [
'userId' => $this->userId,
'email' => $this->email,
];
}
}
```
@@ -190,10 +113,6 @@ class CriticalAlert implements Broadcast
{
use BroadcastEvent;
public function __construct(
public string $message,
) {}
public function getRoutingKey(): string
{
return 'alert.critical';
@@ -208,32 +127,76 @@ class CriticalAlert implements Broadcast
{
return 'topic';
}
public function jsonSerialize(): array
{
return [
'message' => $this->message,
];
}
}
```
### 2. Publishing Events
#### Publishing Events
Events are automatically published when dispatched:
Events are automatically published when dispatched.
`\Diffhead\PHP\LaravelRabbitMQ\Listener\PublishEvent`
This listener implements `ShouldQueue` interface then
you can setup his behaviour by chaning following env variables:
* `RABBITMQ_EVENT_PUBLISHING_CONNECTION=sync`
* `RABBITMQ_EVENT_PUBLISHING_QUEUE=default`
Example:
```php
use App\Events\UserCreated;
/**
* Events implementing Broadcast are automatically sent to RabbitMQ
*/
UserCreated::dispatch(userId: 1, email: 'user@example.com', name: 'John Doe');
App\Events\UserCreated::dispatch(
userId: 1,
email: 'user@example.com',
name: 'John Doe'
);
```
### 3. Consuming Messages from RabbitMQ
### 2. Consuming Messages from RabbitMQ
Use the `rabbitmq:consume` command to listen for messages:
#### Map events
Add regular laravel's application event classes mapping to
RabbitMQ queues and routing keys using `rabbitmq.event.map` section.
Event should implement `\Diffhead\PHP\LaravelRabbitMQ\Interface\Event`
with no methods if you are using default library mapper.
```php
// config/rabbitmq.php
'map' => [
\App\Events\User\UserCreated::class => [
'queues' => ['portal.users'],
'routing_keys' => ['user.created'],
],
\App\Events\Meeting\MeetingCreated::class => [
'queues' => ['portal.meetings'],
'routing_keys' => ['meeting.created'],
],
],
```
You can set empty array for matching event of all queues or routing keys.
Follow this example for mapping event with `amq.direct` exchange which not
uses routing keys:
```php
// config/rabbitmq.php
'map' => [
\App\Events\User\UserCreated::class => [
'queues' => ['portal.users'],
'routing_keys' => [],
],
],
```
#### Consume RabbitMQ queues
Use the `rabbitmq:consume` command to listen for messages.
Consumer will listen rabbitmq bus and emit mapped event from as regular laravel event.
```bash
#####################################################################################
@@ -252,7 +215,7 @@ Use the `rabbitmq:consume` command to listen for messages:
php artisan rabbitmq:consume
```
#### Full Consumer Startup Example
##### Full Consumer Startup Example
```bash
php artisan rabbitmq:consume \
@@ -264,7 +227,7 @@ php artisan rabbitmq:consume \
--tag=service-users-consumer
```
### 4. Handling Received Events
##### Handling Received Events
When a message is received from RabbitMQ, it is automatically deserialized and dispatched as a Laravel event. You can listen to these events normally:
@@ -274,7 +237,7 @@ namespace App\Listeners;
use App\Events\UserCreated;
use Illuminate\Support\Log;
class SendWelcomeEmail
class LogUserCreation
{
public function handle(UserCreated $event): void
{
@@ -288,139 +251,24 @@ Register the listener in `app/Providers/EventServiceProvider.php`:
```php
protected $listen = [
\App\Events\UserCreated::class => [
\App\Listeners\SendWelcomeEmail::class,
\App\Listeners\LogUserCreation::class,
],
];
```
or using Laravel's `\Illuminate\Support\Facades\Event` facade.
## Architecture
### 3. Configure and customize logics
### Event Publishing Flow
#### Serialization
```
Laravel Event (Broadcast)
↓
PublishEvent (Listener)
↓
Serializer (JSON)
↓
RabbitMQ Exchange
↓
Queue
```
The package does serialization/deserialization of message data via following interfaces:
### Message Consumption Flow
* `\Diffhead\PHP\LaravelRabbitMQ\Interface\Serializer`
* `\Diffhead\PHP\LaravelRabbitMQ\Interface\Unserializer`
```
RabbitMQ Queue
↓
Message Consumer
↓
Unserializer (JSON)
↓
EventMapper (Event)
↓
EventEmitter (Service)
↓
Event Listeners
```
By default implementations library interacts with JSON data.
## Microservices Architecture Example
### Service 1: Publishes event
```php
namespace App\Events;
use Diffhead\PHP\LaravelRabbitMQ\Event\Broadcast;
use Diffhead\PHP\LaravelRabbitMQ\Trait\BroadcastEvent;
class UserCreated implements Broadcast
{
use BroadcastEvent;
public function __construct(
public int $id,
public string $email,
public string $name,
) {}
public function getRoutingKey(): string
{
return 'user.created';
}
public function jsonSerialize(): array
{
return [
'id' => $this->id,
'email' => $this->email,
'name' => $this->name,
];
}
}
```
```php
use App\Events\UserCreated;
/**
* Controller method
*/
public function store(Request $request)
{
$user = User::create($request->validated());
UserCreated::dispatch($user->id, $user->email, $user->name);
return response()->json($user, 201);
}
```
### Service 2: Receives event
Map event using configuration:
```php
'map' => [
\App\Events\UserCreated::class => [
'queues' => ['service2.users'],
'routing_keys' => ['user.created']
]
]
```
Then implement and register event listener:
```php
namespace App\Listeners;
use App\Events\UserCreated;
class SyncUserToCalendar
{
public function handle(UserCreated $event): void
{
CalendarUser::create([
'external_id' => $event->id,
'email' => $event->email,
'name' => $event->name,
]);
}
}
```
Start consumer
```bash
php artisan rabbitmq:consume --queue=service2.users --routing-key=user.* --exchange=amq.topic --exchange-type=topic --exchange-is-default
```
## Serialization
The package uses JSON for serialization/deserialization of data via `Serializer` and `Unserializer` interfaces.
### Custom Serialization
##### Custom Serialization
You can use your own serialization classes by implementing
interfaces and overriding following configuration entities:
@@ -432,11 +280,7 @@ interfaces and overriding following configuration entities:
],
```
## Mapping
The package maps rabbitmq message to application events
### Custom mapping
##### Custom mapping
You can use your own mapping logic by implementing EventMapper
interface and overriding the following configuration entity:
@@ -446,3 +290,91 @@ interface and overriding the following configuration entity:
'mapper' => \App\Services\CustomEventMapper::class,
]
```
#### Configuration
Publish the configuration file if it not exists:
```bash
php artisan vendor:publish --provider="Diffhead\PHP\LaravelRabbitMQ\ServiceProvider"
```
Used environment variables:
```env
# RabbitMQ Connection
RABBITMQ_HOST=localhost
RABBITMQ_PORT=5672
RABBITMQ_USER=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_VHOST=/
# Events publishing listener parameters
RABBITMQ_EVENT_PUBLISHING_CONNECTION=sync
RABBITMQ_EVENT_PUBLISHING_QUEUE=default
# Default event parameters
RABBITMQ_EVENT_CONNECTION=default
RABBITMQ_EVENT_QUEUE=default
RABBITMQ_EVENT_EXCHANGE=amq.direct
RABBITMQ_EVENT_EXCHANGE_TYPE=direct
RABBITMQ_EVENT_EXCHANGE_IS_DEFAULT=true
RABBITMQ_EVENT_ROUTING_KEY=
```
##### Example of `config/rabbitmq.php`
```php
// config/rabbitmq.php
return [
'connections' => [
'default' => [
'host' => env('RABBITMQ_HOST', 'localhost'),
'port' => env('RABBITMQ_PORT', 5672),
'user' => env('RABBITMQ_USER', 'guest'),
'password' => env('RABBITMQ_PASSWORD', 'guest'),
'vhost' => env('RABBITMQ_VHOST', '/'),
],
'secondary' => [
'host' => env('RABBITMQ_SECONDARY_HOST', 'localhost'),
'port' => env('RABBITMQ_SECONDARY_PORT', 5672),
'user' => env('RABBITMQ_SECONDARY_USER', 'guest'),
'password' => env('RABBITMQ_SECONDARY_PASSWORD', 'guest'),
'vhost' => env('RABBITMQ_SECONDARY_VHOST', '/'),
]
],
'message' => [
'serializer' => \Diffhead\PHP\LaravelRabbitMQ\Service\Serializer::class,
'unserializer' => \Diffhead\PHP\LaravelRabbitMQ\Service\Unserializer::class,
],
'event' => [
'defaults' => [
'connection' => env('RABBITMQ_EVENT_CONNECTION', 'default'),
'queue' => env('RABBITMQ_EVENT_QUEUE', 'default'),
'exchange' => env('RABBITMQ_EVENT_EXCHANGE', 'amq.direct'),
'exchange_type' => env('RABBITMQ_EVENT_EXCHANGE_TYPE', 'direct'),
'exchange_is_default' => (bool) env('RABBITMQ_EVENT_EXCHANGE_IS_DEFAULT', true),
'routing_key' => env('RABBITMQ_EVENT_ROUTING_KEY', ''),
],
'publishing' => [
'connection' => env('RABBITMQ_EVENT_PUBLISHING_CONNECTION', 'sync'),
'queue' => env('RABBITMQ_EVENT_PUBLISHING_QUEUE', 'default'),
],
'mapper' => \Diffhead\PHP\LaravelRabbitMQ\Service\EventMapper::class,
'map' => [
/**
* Map events to queues and routing keys
*/
\App\Events\User\UserCreated::class => [
'queues' => ['portal.users'],
'routing_keys' => ['user.created'],
],
\App\Events\Meeting\MeetingCreated::class => [
'queues' => ['portal.meetings'],
'routing_keys' => ['meeting.created'],
],
],
]
];
```

5
composer.json
View File

@@ -3,10 +3,11 @@
"description": "A laravel package for events emitting between services using RabbitMQ as message broker.",
"type": "library",
"license": "MIT",
"version": "1.0.1",
"version": "1.1.0",
"keywords": [
"laravel", "rabbitmq", "event", "emit", "microservice",
"pipeline", "data exchanging", "message broker"
"pipeline", "data exchanging", "message", "broker", "php8",
"php", "amqp"
],
"autoload": {
"psr-4": {

40
config/rabbitmq.php
View File

@@ -2,6 +2,11 @@
return [
'connections' => [
/**
* This section helps you to configure
* RabbitMQ connections for consumers or events
* publishing.
*/
'default' => [
'host' => env('RABBITMQ_HOST', 'localhost'),
'port' => env('RABBITMQ_PORT', 5672),
@@ -11,10 +16,23 @@ return [
]
],
'message' => [
/**
* Serializer converts a PHP object into \PhpAmqpLib\Message\AMQPMessage
*/
'serializer' => \Diffhead\PHP\LaravelRabbitMQ\Service\Serializer::class,
/**
* Unserializer converts \PhpAmqpLib\Message\AMQPMessage into a PHP array
*/
'unserializer' => \Diffhead\PHP\LaravelRabbitMQ\Service\Unserializer::class,
],
'event' => [
/**
* \Diffhead\PHP\LaravelRabbitMQ\Trait\BroadcastEvent
*
* This trait helps you to configure where the event should be broadcasted
* by default. You can override these settings directly in the event class
* or globally using environment variables.
*/
'defaults' => [
'connection' => env('RABBITMQ_EVENT_CONNECTION', 'default'),
'queue' => env('RABBITMQ_EVENT_QUEUE', 'default'),
@@ -23,11 +41,30 @@ return [
'exchange_is_default' => (bool) env('RABBITMQ_EVENT_EXCHANGE_IS_DEFAULT', true),
'routing_key' => (string) env('RABBITMQ_EVENT_ROUTING_KEY', ''),
],
/**
* \Diffhead\PHP\LaravelRabbitMQ\Listener\PublishEvent
*
* This listener publishes events to a rabbitmq bus and
* does that using laravel queues.
*
* You can configure which connection and queue to use for.
*/
'publishing' => [
'connection' => env('RABBITMQ_EVENT_PUBLISHING_CONNECTION', 'sync'),
'queue' => env('RABBITMQ_EVENT_PUBLISHING_QUEUE', 'default'),
],
/**
* This mapper detects event which should
* be emitted when RabbitMQ message received.
*/
'mapper' => \Diffhead\PHP\LaravelRabbitMQ\Service\EventMapper::class,
/**
* Map events with their queues and routing keys.
*/
'map' => [
/**
* Example:
*
*
* \App\Shared\Event\User\UserCreated::class => [
* 'queues' => ['portal.calendar.users'],
* 'routing_keys' => ['user.created'],
@@ -38,6 +75,5 @@ return [
* ],
*/
],
]
];

9
src/Command/Consume.php
View File

@@ -5,7 +5,7 @@ declare(strict_types=1);
namespace Diffhead\PHP\LaravelRabbitMQ\Command;
use Diffhead\PHP\LaravelRabbitMQ\Dto\ConsumerParameters;
use Diffhead\PHP\LaravelRabbitMQ\Object\Connection;
use Diffhead\PHP\LaravelRabbitMQ\Object\ConnectionWithChannel;
use Diffhead\PHP\LaravelRabbitMQ\Object\Exchange;
use Diffhead\PHP\LaravelRabbitMQ\Object\ExchangeDeclaration;
use Diffhead\PHP\LaravelRabbitMQ\Object\Queue;
@@ -31,7 +31,7 @@ class Consume extends Command
*/
protected $description = 'Consume messages from RabbitMQ';
private ?Connection $connection = null;
private ?ConnectionWithChannel $connection = null;
public function handle(
Configuration $configuration,
@@ -40,7 +40,10 @@ class Consume extends Command
LoggerInterface $logger
): void {
$params = ConsumerParameters::fromArray($this->options());
$config = $configuration->get($params->connection->value() ?? 'default');
$config = $configuration->getConnection(
$params->connection->value() ?? 'default'
);
$queue = $this->getQueue($params);

26
src/Listener/PublishEvent.php
View File

@@ -13,11 +13,15 @@ use Diffhead\PHP\LaravelRabbitMQ\Object\QueueDeclaration;
use Diffhead\PHP\LaravelRabbitMQ\Service\Configuration;
use Diffhead\PHP\LaravelRabbitMQ\Service\Connector;
use Diffhead\PHP\LaravelRabbitMQ\Event\Broadcast;
use Diffhead\PHP\LaravelRabbitMQ\Object\Publishing;
use Exception;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Support\Facades\App;
class PublishEvent implements ShouldQueue
{
private Publishing $publishing;
public function __construct(
private Configuration $configuration,
private Connector $connector,
@@ -26,7 +30,7 @@ class PublishEvent implements ShouldQueue
public function handle(Broadcast $event): void
{
$config = $this->configuration->get($event->getConnection());
$config = $this->configuration->getConnection($event->getConnection());
$queue = new Queue(
name: $event->getQueue(),
@@ -59,4 +63,24 @@ class PublishEvent implements ShouldQueue
$connection->connection()->close();
}
}
public function viaConnection(): string
{
return $this->getOrCreatePublishing()->connection();
}
public function viaQueue(): string
{
return $this->getOrCreatePublishing()->queue();
}
private function getOrCreatePublishing(): Publishing
{
if (! isset($this->publishing)) {
$configuration = App::make(Configuration::class);
$this->publishing = $configuration->getPublishing();
}
return $this->publishing;
}
}

41
src/Object/Configuration.php
View File

@@ -1,41 +0,0 @@
<?php
declare(strict_types=1);
namespace Diffhead\PHP\LaravelRabbitMQ\Object;
class Configuration
{
public function __construct(
private string $host,
private int $port,
private string $user,
private string $password,
private string $vhost,
) {}
public function host(): string
{
return $this->host;
}
public function port(): int
{
return $this->port;
}
public function user(): string
{
return $this->user;
}
public function password(): string
{
return $this->password;
}
public function vhost(): string
{
return $this->vhost;
}
}

33
src/Object/Connection.php
View File

@@ -4,23 +4,38 @@ declare(strict_types=1);
namespace Diffhead\PHP\LaravelRabbitMQ\Object;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
class Connection
{
public function __construct(
private AMQPStreamConnection $connection,
private AMQPChannel $channel
private string $host,
private int $port,
private string $user,
private string $password,
private string $vhost,
) {}
public function connection(): AMQPStreamConnection
public function host(): string
{
return $this->connection;
return $this->host;
}
public function channel(): AMQPChannel
public function port(): int
{
return $this->channel;
return $this->port;
}
public function user(): string
{
return $this->user;
}
public function password(): string
{
return $this->password;
}
public function vhost(): string
{
return $this->vhost;
}
}

26
src/Object/ConnectionWithChannel.php Normal file
View File

@@ -0,0 +1,26 @@
<?php
declare(strict_types=1);
namespace Diffhead\PHP\LaravelRabbitMQ\Object;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
class ConnectionWithChannel
{
public function __construct(
private AMQPStreamConnection $connection,
private AMQPChannel $channel
) {}
public function connection(): AMQPStreamConnection
{
return $this->connection;
}
public function channel(): AMQPChannel
{
return $this->channel;
}
}

23
src/Object/Publishing.php Normal file
View File

@@ -0,0 +1,23 @@
<?php
declare(strict_types=1);
namespace Diffhead\PHP\LaravelRabbitMQ\Object;
class Publishing
{
public function __construct(
private string $connection,
private string $queue,
) {}
public function connection(): string
{
return $this->connection;
}
public function queue(): string
{
return $this->queue;
}
}

15
src/Service/Configuration.php
View File

@@ -4,12 +4,13 @@ declare(strict_types=1);
namespace Diffhead\PHP\LaravelRabbitMQ\Service;
use Diffhead\PHP\LaravelRabbitMQ\Object\Configuration as ConfigurationObject;
use Diffhead\PHP\LaravelRabbitMQ\Object\Connection;
use Diffhead\PHP\LaravelRabbitMQ\Object\Publishing;
use RuntimeException;
class Configuration
{
public function get(string $connection = 'default'): ConfigurationObject
public function getConnection(string $connection = 'default'): Connection
{
$config = config(sprintf('rabbitmq.connections.%s', $connection));
@@ -19,7 +20,7 @@ class Configuration
);
}
return new ConfigurationObject(
return new Connection(
$config['host'],
(int) $config['port'],
$config['user'],
@@ -27,4 +28,12 @@ class Configuration
$config['vhost'],
);
}
public function getPublishing(): Publishing
{
return new Publishing(
config('rabbitmq.event.publishing.connection', 'sync'),
config('rabbitmq.event.publishing.queue', 'default'),
);
}
}

6
src/Service/Connector.php
View File

@@ -4,14 +4,14 @@ declare(strict_types=1);
namespace Diffhead\PHP\LaravelRabbitMQ\Service;
use Diffhead\PHP\LaravelRabbitMQ\Object\Configuration;
use Diffhead\PHP\LaravelRabbitMQ\Object\Connection;
use Diffhead\PHP\LaravelRabbitMQ\Object\ConnectionWithChannel;
use Diffhead\PHP\LaravelRabbitMQ\Object\Queue;
use PhpAmqpLib\Connection\AMQPStreamConnection;
class Connector
{
public function connect(Configuration $config, Queue $queue): Connection
public function connect(Connection $config, Queue $queue): ConnectionWithChannel
{
$connection = new AMQPStreamConnection(
$config->host(),
@@ -57,6 +57,6 @@ class Connector
$queue->bindings()->ticket()
);
return new Connection($connection, $channel);
return new ConnectionWithChannel($connection, $channel);
}
}