Download the PHP package makinacorpus/corebus without Composer
On this page you can find all versions of the php package makinacorpus/corebus. It is possible to download/install these versions without Composer. Possible dependencies are resolved automatically.
Download makinacorpus/corebus
More information about makinacorpus/corebus
Files in makinacorpus/corebus
Package corebus
Short Description Command and event buses interface and logic.
License GPL-2.0-or-later
Homepage http://github.com/makinacorpus/php-corebus
Informations about the package corebus
CoreBus - Command and Event buses interfaces
Discrete command bus and domain event dispatcher interfaces for message based architectured projects.
Discrete means that your domain code will not be tainted by this component hard-dependency, aside attributes used for targeting command handler methods and event listener methods. Your domain business code remains dependency-free.
Event bus features:
- Internal synchronous event dispatcher.
- Event listener locator based upon PHP attributes.
- Event listener locator fast dumped PHP cache.
- External event dispatcher (unplugged yet).
Command bus features:
- Transactional synchronous command bus that handles your transactions.
- Event buffering during transactions, which flushes events to the external event dispatcher only in case of transaction success.
- Command handler locator based upon PHP attributes.
- Command handler locator fast dumped PHP cache.
- Default transaction implementation using
makinacorpus/goat-query
. - Command asynchronous dispatcher with implementation that plugs to
makinacorpus/message-broker
message broker interface.
Other various features:
- Worker object for consuming asynchronous events in CLI.
- Symfony integration for everything, including console commands for the command bus worker.
- Global attributes for aspect-driven domain code configuration.
- Simple command bus interface.
Design
Basic design
Expected runtime flow of your application is the following:
- Commands may be dispatched to trigger writes in the system.
- Commands are always asynchronously handled, they may return a response.
- One command implies one transaction on your database backend.
- During a single command processing, the domain code may raise one or many domain events.
- Domain events are always dispatched synchronously within your domain code, within the triggering command transaction.
During the whole command processing, the database transaction will be isolated if the backend permits it. Commit is all or nothing, including events being emitted and listener execution during the process.
Transaction and event buffer
Transaction handling will be completely hidden in the implementations, your business code will never see it, here is how it works:
- Domain events while emitted and dispatched internally are stored along the way into a volatile in-memory temporary buffer.
- Once command is comsumed and task has ended, transaction will commit.
- In case of success, buffer is flushed and events may be sent to a bus for external application to listen to.
- In case of failure, transaction rollbacks, event buffer is emptied, events are discarded without further action.
Transactions can be disabled on a per-command basis, using PHP attributes on the command class.
Optional event store
If required for your project, you may plug an event store on the event dispatcher. Two options are possible:
- Plug in into the internal event dispatcher, events will be stored along the way, this requires that the event store works on the same database transaction, hence connection, than your domain repositories.
- Plug in into the event buffer output, which means events will be stored after commit, there is no consistency issues anymore, but if event storage procedure fails, you will loose history.
Internal workflow of commands
Command dispatcher
Command dispatcher is the user facing command bus for users. It has two existing variations:
-
MakinaCorpus\CoreBus\CommandBus\CommandBus
which is the default command bus interface. MakinaCorpus\CoreBus\CommandBus\SynchronousCommandbus
, which extends the former. It is the exact same interface, but exists for dependency injection purpose: using the synchronous command bus will always passthrough the command directly from the dispatcher to the consumer, which will make messages being consumed synchronously.
All implementations implement both interfaces, the dependency injection component has the reponsability to distinguish the synchronous from the asynchronous implementation.
It is also the place where command access control will happen throught a decorator, as documented later in this document.
Command consumer
Command consumer is the local implementation that from a given command will execute it. You can find multiple implementations:
-
MakinaCorpus\CoreBus\CommandBus\Consumer\DefaultCommandConsumer
is the real implementation, that does the handler function lookup and execute it. -
MakinaCorpus\CoreBus\CommandBus\Consumer\TransactionalCommandConsumer
is an implementation that will decorate the default consumer, and wrap handler execution into a single database transaction, raising domain events along the way, and rollback in case of any error. MakinaCorpus\CoreBus\CommandBus\Consumer\NullCommandConsumer
is only in use in unit tests, please ignore that implementation. It may come useful to to you if for any reason you need a dummy implementation.
For now there is a step which hasn't made generic yet, the command bus worker
which get messages from the queue and send those to the message consumer: it
is hard wired to the makinacorpus/message-broker
package.
Implementations
Two implementations are provided:
- In-memory bus, along with null transaction handling (no transaction at all) ideal for prototyping and unit-testing.
- PostgreSQL bus implementation using
makinacorpus/goat
for message broker, andmakinacorpus/goat-query
for transaction handling using the same database connection, reliable and guaranteing data consistency.
Everything is hidden behind interfaces and different implementations are easy to implement. Your projects are not required to choose either one of those implementations, in the opposite, is encouraged implementing its own.
Setup
Standalone
There is no standalone setup guide for now. Refer to provided Symfony configuration for a concrete example.
Symfony
Simply enable the bundle in your config/bundles.php
file:
Then cut and paste src/Bridge/Symfony/Resources/example/corebus.sample.yaml
file into your config/packages/
folder, and edit it.
Usage
Commands and events
Commands are plain PHP object and don't require any dependency.
Just write a Data Transport Object:
Same goes with events, so just write:
Register handlers using base class
Tie a single command handler:
Please note that using the AbstractCommandHandler
base class is purely
optional, it's simply an helper for being able to use the event dispatcher
and command bus from within your handlers.
Alternatively, if you don't require any of those, you may just:
- Either set the
#[MakinaCorpus\CoreBus\Attr\CommandHandler]
attribute on the class, case in which all of its methods will be considered as handlers. - Either set the
#[MakinaCorpus\CoreBus\Attr\CommandHandler]
attribute on each method that is an handler.
You may also write as many event listeners as you wish, then even may emit events themselves:
Same goes for event listeners, the base class is just here to help but is not required, you may just:
- Either set the
#[MakinaCorpus\CoreBus\Attr\EventListener]
attribute on the class, case in which all of its methods will be considered as listeners. - Either set the
#[MakinaCorpus\CoreBus\Attr\EventListener]
attribute on each method that is an listener.
This requires that your services are known by the container. You have three different options for this.
First one, which is Symfony's default, autoconfigure all your services:
Or if you wish to play it subtle:
Or if you want to do use the old ways:
In all cases, you don't require any tags or any other metadata as long as you either extend the base class, or use the attributes.
Register handlers using attributes
Tie a single command handler:
You may also write as many event listeners as you wish, then even may emit events themselves:
Using Symfony container machinery, no configuration is needed for this to work.
Symfony commands
Push a message into the bus
Pushing a message is as simple as:
Run worker process
Running the worker process is as simple as:
If you set -vv
you will obtain a very verbose output and is a very bad idea
to do in any other environment than your development machine.
Running using -v
will output a single line for every message being consumed
including some time and memory information. Exceptions traces when a message fail
will be displayed fully in output. This is a good setting for using it with
systemd
or a docker
container that will pipe the output into logs.
Not setting any -v
flag will be equivalent to -vv
but output will only
happen in monolog, under the corebus
channel.
Additionally, you may tweak using the following options:
--limit=X
: only process X messages and die,--routing-key=QUEUE_NAME
: only process messages in theQUEUE_NAME
queue,--memory-limit=128M
: when PHP memory limit exceeds the given limit, die, per default the process will use current PHP limit minus 16M, in order to avoid PHP memory limit reached errors during message processing,--memory-leak=512K
: warn in output when a single message consumption doesn't free completely memory once finished, with the given threshold,--sleep-time=X
: wait for X microseconds between two messages when there is none left to consume before retrying a bus fetch. This may be ignored by some implementations in the future.
Using attributes
This package comes with an attribute support for annotating commands and events in order to infer behaviors to the bus. This allows to declare commands or event behaviour without tainting the domain code.
Command attributes
-
#[MakinaCorpus\CoreBus\Attr\NoTransaction]
disables transaction handling for the command. Use it wisely. -
#[MakinaCorpus\CoreBus\Attr\RoutingKey(name: string)]
allows you to route the command via the given routing key (or queue name). Default when this attribute is not specified isdefault
. -
#[MakinaCorpus\CoreBus\Attr\Async]
forces the command to always be dispatched asynchronously. Warning, this is not implemented yet, and is an empty shell. #[MakinaCorpus\CoreBus\Attr\Retry(count: ?int)]
allows the command to be retried in case an error happen. First parameter is the number of retries allowed, default is3
. Warning, this is not implemented yet, and is an empty shell.
Domain event attributes.
#[MakinaCorpus\CoreBus\Attr\Aggregate(property: string, type: ?string)]
allows the developer to explicitely tell which aggregate (entity or model) this event targets. First argument must be a property name of the event that is the aggregate identifier, second argument is optional, and is the target aggregate class or logicial name. If you are using an event store, aggregate type is only mandatory for aggregate stream creation events, identifier will be enough for appending event in an existing stream.
Configuration attributes
-
#[MakinaCorpus\CoreBus\Attr\CommandHandler]
if set on a class, will force the bus to introspect all methods and register all its methods as command handlers, if on a single method, will register this explicit method as being a command handler. #[MakinaCorpus\CoreBus\Attr\EventListener]
if set on a class, will force the bus to introspect all methods and register all its methods as event listeners, if on a single method, will register this explicit method as being an event listener.
For all those attributes, parameters are optional, but you might set the
target
parameter to disambiguate which class the handler or listener catches.
Using this, you can use interfaces for matching instead of concrete classes.
Access control on command input
This API allows you to provide access control on command input. Warning, on input only, once a command is withing the bus, you can't access control it anymore.
You can do custom access control by implementing the
MakinaCorpus\CoreBus\CommandBus\CommandAuthorizationChecker
interface:
For Symfony bundle user, you need to set the corebus.authorization_checker
tag on your registered authorization checker services.
Moreover, if you use makinacorpus/access-control
Symfony bundle along this
API own bundle, it will be autoconfigured:
Access checks are done when executing the CommandBus::dispatch()
method using
the decorator pattern. This may prevent you from running arbitrary commands into
the bus so we will provide a fallback in the future in order to be able to use
an unprotected bus (for CLI commands, for example). For the time being, you need
to implement the authorization checker wisely to avoid unexpected behavior.
Exposing command bus HTTP endpoints
Provided controller
A working basic controller implementation is provided as the
MakinaCorpus\CoreBus\Bridge\Symfony\Controller\CommandController
class and
provides three methods.
Since exposing your bus directly as an HTTP endpoint poses important security concerns, it's up to you to configure and secure it. This is why this controller is not auto-configured.
Remember that all commands will pass throught the CommandAuthorizationChecker
in default configuration, you may configure access right using it.
Configuring the controller in Symfony
A sample src/Bridge/Symfony/Resources/example/corebus.routing.yaml
file is
provided but will not be configured per default: you must copy/paste it or its
contents into your project to make it work.
Dispatching a command
Simply make a POST
or PUT
HTTP request on the /api/command/dispatch
endpoint path (that you may have changed):
You should expect the following result:
Please note that if you pass the &async=1
GET parameter, command will be
queued into the bus, and "status": "queued"
will be present in the response.
If you don't specify the parameter, command will be consumed synchronously
"status": "ok"
will be returning instead, if message succeded.
The reply_to
parameter can be expressed as a GET parameter as well, if set
to any value, a reply to queue name will be generated on the server side
and returned in the "reply_to"
JSON property.
This allows you to use the consume
endpoint using this queue name in order
to fetch a potential reply.
Dispatching a command transaction
This is undocumented yet.
Consumming messages from a given queue
The consume endpoint was created mostly for implementing the RPC method call
via the command bus pattern, hence the following restriction: you can only
listen to queues you implicitely created by setting the reply_to
parameter
when calling the HTTP dispatch endpoint.
When sending a reply_to
parameter, a name will be generated on the server
side, stored into session, and returned by the dispatch command under the
reply_to
attribute.
Examples
Synchronous dispatch example
This would give you the following result:
Please note the lack of properties, since the message went throught a direct dispatch, there is no bus metadata attached to the message.
Request body response
property contains the handler response.
Asychronous dispatch example
For example, considering that the App\Command\Ping
command simply return
a App\Command\PingResponse
object with a few properties, let's dispatch it
into the bus:
This would give you the following result:
Queue response consumption
Let's consider the previous example, notice that you will find the reply-to
property into the properties
property of the response: this is the
automatically computed queue name for receiving the handler asynchronous
reponse.
Let's consume the queue:
If the command was not processed, you would get the following result:
As soon as the response is received and awaits into the queue:
Response is similar to synchronous dispatch, except that you will receive the additional properties metadata as well.
This feature is still experimental, there is no security attached to it yet.
Securing the consume endpoint
Per default, the consume endpoint won't allow any arbitrary queue name, since it was created in order to fullfil the need of implemeting RPC via the command bus, it will only allow a certain naming pattern.
Calling RPC methods via bus using HTTP
@todo
Overriding implementations
Any interface in this package is a service in the dependency injection container you will use. You may replace or decorate any of them.
Roadmap
- [ ] Implement profiling decorator for event bus using
makinacorpus/profiling
. - [ ] Implement profiling decorator for command bus using
makinacorpus/profiling
. - [ ] Allow multiple message brokers to co-exist, one for each queue.
- [ ] Implement dead letter queue routing.
- [ ] Create a retry strategy chain for having more than one instance.
- [ ] Implement retry strategy using our attributes.
- [ ] Configurable per-exception type retry strategry.
- [ ] Implement an argument resolver for command handlers and event listeners.
All versions of corebus with dependencies
psr/log Version ^1.0 || ^2.0 || ^3.0
makinacorpus/argument-resolver Version ^1.0.4
makinacorpus/message Version ^1.1