Download the PHP package prwnr/laravel-streamer without Composer

Laravel Streamer

Streamer is a Laravel package for events functionality between different applications, powered by Redis Streams. This package utilizes all main commands of Redis 5.0 Streams providing a simple usage of Streams as Events.

Main concept of this package is to provide easy way of emitting new events from your application and to allow listening to them in your other applications that are using same Redis server.

Installation

1. Install package via composer command composer require prwnr/laravel-streamer or by adding it to your composer.json file with version.
2. Discover the package
3. Publish configuration with vendor:publish command.
4. Make sure that you have running Redis 5.0 instance and that Laravel is configured to use it
5. Make sure that you have PHPRedis extension installed.

   Usage

   There are two main ends of this package usage - emiting new event and listening to events. Whereas emiting requires a bit more work to get it used, such as creating own Event classes, then listening to events is available with artisan command and is working without much work needed.

Version Compatibility

Emiting new events

In order to emit new event few things needs to be done. First of all, you will need to have a valid class that implements Prwnr\Streamer\Contracts\Event like this:

Then, at any point in your application all you need to do is to emit that event by using either Streamer instance or Streamer facade.

This will create a message on a stream named (if such does not exist): example.streamer.event. Emit method will return an ID if emitting your event ended up with success.

Listening for new messages on events

In order to listen to event you will have to properly configure config/streamer.php (or use ListenerStack::add method) and run php artisan streamer:listen command. At config file you will find Application listeners configuration with default values for it, that should be changed if you want to start listening with streamer listen command. Other way to add listeners for events is to use ListenersStack static class. This class is being booted with listeners from configuration file and is then used by command to get all of them. So, the addition of this class is that it allows adding listeners not only via configuration file, but also programmatically.

Remember that local listener should implement MessageReceiver contract to ensure that it has handle method which accepts ReceivedMessage as an argument.

Above configuration is an array of Streamer events with an array of local listeners related to Streamer event. When listening to example.streamer.event all local listeners from its config definition will be created and their handle method fired with message received from Stream. For listener instance creation this package uses Laravel Container, therefore you can type hint anything into your listener constructor to use Laravel dependency injection.

To start listening for an event, use listen command.

Commands

Listen

This command will start listening on a given stream (or streams separated by comma) starting from "now". It will be listening in a blocking way, meaning that it will run until Redis will time out or crash. All listener related errors are being caught and logged into console as well as stored in Failed Messages list for later debugging and/or retrying.

That's a basic usage of this command, where event name is a required argument (unless --all option is provided). So in this case it simply starts listening for only new events. This command however has few options that are extending its usage, those are:

When consumer and group options are being in use, every message on a stream will be marked as acknowledged for the given consumer, thus it will not be processed by consequent streamer:listen command call with the same options. Note that listening from a specific ID without consumer and group being set will ignore acknowledgments.

The purge and archive options (available since v2.6) are designed to be used to release memory or storage of the Redis instance (in a cases when there are tons of streamed messages or the payloads are big and Redis runs out of memory/storage). When using those options, keep in mind, that they are not going to take into account listeners running in other instances or other servers - meaning, that when first listener hooked to specific event will process its messages, the purge and archive options will delete the message not waiting for other listeners to finish. To fully use archive option see [Stream Archive][#stream-archive] for more details and instructions.

Using multiple events in argument or the --all option with any other option (like group, consumer, last_id) will apply those options to every stream event that is being in use.

Failed List

This command will show list of stream messages that failed to be handled by their listeners. It will yield all the important information about them like: ID, stream name, listener class, error message that cause it to fail, and a date when that happened.

Table example:

There's one addition option for this command, called --compact which will limit the table output to only ID, Stream and Error columns.

Failed Retry

This command is meant to try again failed listening. It simply reads the message from a stream and attempts to handle it again by the listener that it was originally processed.

When the listener fails to process the message again, the message failed information will be re-stored (with a newer date and updated error message) and will be available to be retried again. There's no limit to how many times message can be processed. It will remain available after each fail unless flush command will be used.

This command has few options that are available:

At least one of those options is required to be used with the command to process failed messages. The all option can be only used solely, while the other three options can be used together or not. This means, that any combination of id , stream and receiver can be used to match any number of failed messages and retry them. So, for example a stream can be used together with id or in other case id can be used with receiver, or only one of them can be used, or all three at once, its all up to the use case.

Failed Flush

This command will remove all existing failed messages from the messages' repository. Can be used to prune entries that cannot be processed at all by listeners.

This command WILL NOT remove the message from the Stream itself - the message will remain there untouched, but acknowledged by its original consumer (if used).

List

This command will list all registered events, and their associated listeners. The option --compact will yield only a list of the events, skipping listeners column.

This command may be useful to see what events are being actually handled by a listener, what can help to find out what's missing. This list can be also used to start listening to available events by 3rd party app.

Table example:

Archive

This command will archive messages from a selected streams older than days/weeks or so. It will process all stream messages, verifying their created timestamp and will attempt to archive (deleting them from redis and attempting to store them in associated archive storage) each one of them.

This command has two required options:

Be aware of using this command, as it will not take into account whether listeners processed messages it tries to archive or not. This should be used with caution and only for older messages, so that it will be more certain, that all listeners processed their messages.

Purge

This command will purge messages from a selected streams older than days/weeks or so. It will process all stream messages, verifying their created timestamp and will attempt to purge them (deleting them from the redis entirely).

This command has two required options:

Be aware of using this command, as it will not take into account whether listeners processed messages it tries to purge or not. This should be used with caution and only for older messages, so that it will be more certain, that all listeners processed their messages.

Archive Restore

This command will restore archived stream messages from the associated archive storage. It will essentially fetch messages (all or selection) and will try to put them back onto the stream, while also deleting them from the archive. This action will trigger listeners that are hooked to the restored streams!

This command has few options that are available:

At least one of those options is required to attempt restoration of the messages. If any error occurs while restoring a message, it will be reported for that particular attempt not preventing other message from being processed.

Restoring message puts it back onto a stream with NEW ID - this is Redis requirement and limitation, that any message added to stream, needs to have ID higher than the last generated one. The original ID of the message that is being restored will be stored in meta information in original_id field.

Stream Archive

Stream Archive allows storing processed message in any kind of storage, to free up Redis memory and/or space since 2.6 version. Archive allows restoring those messages, releasing them back onto the stream.

To fully use archive storage, a new storage driver needs to be written and added to manager. To do so, few quick steps needs to be finished:

1) define your storage driver class and make it implement \Prwnr\Streamer\Contracts\ArchiveStorage contract. This interface is mandatory. 2) extend the manager with your driver like this:

3) define your driver as default in streamer config file

Replaying Events

Since 2.2 version, Stream events can be "replayed". This means, that the specific message (with a unique identifier) can be "reconstructed" until "now" (or until a selected date).

What "replaying" messages really means? It means, that all the messages that are in the stream, will be read from the very beginning, and payload of each single entry will be "combined" into a final version of the message - each filed will be replaced with its "newer" value, if such exists in the history.

This is going to be useful with events that don't hold all the information about the resource they may represent, but have only data about fields that changed.

So, for example having a resource with fields name and surname, we will emit 3 different events:

• first for its creation populating both fields with values (name: foo; surname: bar)
• second event that will change only name into foo bar
• third event that changes name again to bar foo.

While replaying this set of messages (remember that each one has the same unique identifier) our final replayed resource will be: name: bar foo; surname: bar. If we replay the event until the time before third change, we would have name: foo bar; surname: bar

Usage

To make Event replayable, it needs to implement the Prwnr\Streamer\Contracts\Replayable contract. This will enforce adding a getIdentifier method, that should return unique identifier for the resource (like UUID of the resource that this event represents). With this contract being fulfilled, all events that will go through Streamer emit method, will be also "marked" as available to be replayed.

To actually replay messages, Hsitory interface implementation needs to be used.

Method that should be used is: replay(string $event, string $identifier, Carbon $until = null): array. This method will return the "current state" of the event, rebuilding it from its history. As seen in method definition, it asks for event string name and resource identifier (that was applied by Replayable contract). Third parameter is optional and if used, it will stop replaying messages when first message with matching date will be encountered.

Eloquent Model Events

With use of a EmitsStreamerEvents trait you can easily make your Eloquent Models emit basic events. This trait will integrate your model with Streamer and will emit events on actions like: save, create and delete. It will emit an event of your model name with suffix of the action and a payload of what happened. In case of a create and save actions the payload will have a list of changed fields and a before/after for each of those fields (with create action fields before will basically have all values set to null), in case of a delete action, payload will simply state that the model has been deleted. Each payload includes a [key_name => key_value] pair of your model ID.

By default, events will take names from their models with a suffix of the action, but the name can be changed by assigning it to a baseEventName attribute. This name will replace the model name but will keep suffix of what action has been taken.

Check example's directory in this package to see how can you exactly use each command with package Stream and Consumer instances.