Migrating the PHP backend to Redis streams and selecting a framework-independent library

Preface

My website, which I manage as a hobby, is intended for storing interesting personal pages and websites. This topic intrigued me at the very beginning of my programming journey, at that time I was fascinated by finding great professionals who write about themselves, their interests, and projects. The habit of discovering them for myself remains to this day: almost on every commercial and non-commercial site, I continue to peek into the footer in search of links to the authors.

Implementation of the idea

The first version was just an HTML page on my personal website, where I collected links with descriptions in an unordered list. After gathering about 20 pages over some time, I started thinking that this was not very efficient and decided to try to automate the process. I noticed on Stack Overflow that many indicate websites in their profiles, so I wrote a parser in PHP that simply traversed the profiles, starting from the first one (the URLs on SO to this day look like: "/users/1"), extracted links from the necessary tag, and stored them in SQLite.

This could be called the second version: a collection of tens of thousands of URLs in an SQLite table that replaced the static list in HTML. I created a simple search for this list. Since there were only URLs, the search was just by them.

At this stage, I abandoned the project and returned to it after a long time. At this stage, my work experience was over three years, and I felt I could do something more serious. Additionally, I had a strong desire to learn relatively new technologies.

Modern version

Project deployed in Docker, the database has been migrated to MongoDB, and relatively recently, Redis was added, which initially was just for caching. A microframework for PHP serves as the foundation.

The Problem

New websites are added via a console command that synchronously does the following:

• Downloads content by URL
• Sets a flag indicating whether HTTPS was accessible
• Saves the web entity
• Stores the original HTML and headers in the 'indexing' history
• Parses the content, extracting title and description
• Saves the data in a separate collection

This was sufficient to simply store websites and display them in a list:

However, the idea of automatically indexing, categorizing, and ranking everything while keeping it up to date fits poorly into this paradigm. Even simply adding a web method to include pages required code duplication and locks to avoid potential DDoS attacks.

In general, of course, everything can be done synchronously, and in the web method, we could just save the URL for a monstrous daemon to perform all tasks for the URLs in the list. Yet even here, the word 'queue' comes to mind. If we implement a queue, we can divide all tasks and process them at least asynchronously.

Solution

Implement queues and create an event-driven system for processing all tasks. I've long wanted to try Redis Streams.

Using Redis streams in PHP

Since my framework is not one of the three giants—Symfony, Laravel, Yii—I'd prefer to find an independent library. However, as it turned out (upon initial review), it was impossible to find any serious standalone libraries. Everything related to queues is either a five-year-old project with three commits or tied to a framework.

I’ve heard about Symfony as a supplier of useful standalone components, and I already use some. Also, there are things from Laravel that can be utilized, such as their ORM, without the need for the framework itself.

symfony/messenger

The first candidate immediately seemed perfect, and without any doubts, I installed it. However, it turned out to be more challenging to find examples of usage outside of Symfony. How do I assemble a bus for message transmission from a pile of classes with generic, meaningless names, and still use Redis?

The documentation on the official website was quite detailed, but the initialization was only described for Symfony using their beloved YML and other magical methods for non-Symfony users. I had no interest in the installation process, especially during the New Year holidays. Yet, I had to deal with it, and unexpectedly it took a long time.

Trying to figure out the instantiation of the system based on Symfony's source code is also not a trivial task given the tight deadlines:

After digging through all of this and attempting to do something manually, I realized I was working with hacks and decided to try something else.

illuminate/queue

It turned out that this library is tightly coupled to the Laravel infrastructure and a bunch of other dependencies, so I didn't spend much time on it: I installed it, looked at it, saw the dependencies, and removed it.

yiisoft/yii2-queue

Well, the name immediately suggested a strong coupling to Yii2. I had to use this library, and it was decent, but I didn't think about the fact that it was entirely dependent on Yii2.

Others

Everything else I found on GitHub was unreliable outdated and abandoned projects with no stars, forks, or many commits.

Returning to symfony/messenger, technical details

I had to figure out this library, and after spending some more time on it, I managed to do so. It turned out everything was quite concise and simple. For instantiating the bus, I made a small factory since I planned to have several buses with different handlers.

Just a few steps:

• We create message handlers, which should simply be callable
• We wrap them in HandlerDescriptor (a class from the library)
• These 'Descriptors' are wrapped in an instance of HandlersLocator
• We add HandlersLocator to an instance of MessageBus
• We pass a set of `SenderInterface` to SendersLocator, in my case instances of `RedisTransport`, which are configured in an obvious way
• We add SendersLocator to an instance of MessageBus

MessageBus has a method `->dispatch()`, which looks for corresponding handlers in HandlersLocator and passes the message to them, using the appropriate `SenderInterface` to send it through the bus (Redis streams).

In the container configuration (in this case php-di), all this linkage can be configured as follows:

        CONTAINER_REDIS_TRANSPORT_SECRET => function (ContainerInterface $c) {
            return new RedisTransport(
                $c->get(CONTAINER_REDIS_STREAM_CONNECTION_SECRET),
                $c->get(CONTAINER_SERIALIZER))
            ;
        },
        CONTAINER_REDIS_TRANSPORT_LOG => function (ContainerInterface $c) {
            return new RedisTransport(
                $c->get(CONTAINER_REDIS_STREAM_CONNECTION_LOG),
                $c->get(CONTAINER_SERIALIZER))
            ;
        },
        CONTAINER_REDIS_STREAM_RECEIVER_SECRET => function (ContainerInterface $c) {
            return new RedisReceiver(
                $c->get(CONTAINER_REDIS_STREAM_CONNECTION_SECRET),
                $c->get(CONTAINER_SERIALIZER)
            );
        },
        CONTAINER_REDIS_STREAM_RECEIVER_LOG => function (ContainerInterface $c) {
            return new RedisReceiver(
                $c->get(CONTAINER_REDIS_STREAM_CONNECTION_LOG),
                $c->get(CONTAINER_SERIALIZER)
            );
        },
        CONTAINER_REDIS_STREAM_BUS => function (ContainerInterface $c) {
            $sendersLocator = new SendersLocator([
                AppMessagesSecretJsonMessages::class => [CONTAINER_REDIS_TRANSPORT_SECRET],
                AppMessagesDaemonLogMessage::class => [CONTAINER_REDIS_TRANSPORT_LOG],
            ], $c);
            $middleware[] = new SendMessageMiddleware($sendersLocator);

            return new MessageBus($middleware);
        },
        CONTAINER_REDIS_STREAM_CONNECTION_SECRET => function (ContainerInterface $c) {
            $host = 'bu-02-redis';
            $port = 6379;
            $dsn = "redis://$host:$port";
            $options = [
                'stream' => 'secret',
                'group' => 'default',
                'consumer' => 'default',
            ];

            return Connection::fromDsn($dsn, $options);
        },
        CONTAINER_REDIS_STREAM_CONNECTION_LOG => function (ContainerInterface $c) {
            $host = 'bu-02-redis';
            $port = 6379;
            $dsn = "redis://$host:$port";
            $options = [
                'stream' => 'log',
                'group' => 'default',
                'consumer' => 'default',
            ];

            return Connection::fromDsn($dsn, $options);
        },

Here, you can see that in the SendersLocator, we assigned different "transports" for two different messages, each with its corresponding connection to the respective streams.

I created a separate demo project demonstrating an application made up of three demons communicating with each other using such a bus: https://github.com/backend-university/products/tree/master/products/02-redis-streams-bus.

But I will show you how a consumer can be structured:

use AppMessagesDaemonLogMessage;
use SymfonyComponentMessengerHandlerHandlerDescriptor;
use SymfonyComponentMessengerHandlerHandlersLocator;
use SymfonyComponentMessengerMessageBus;
use SymfonyComponentMessengerMiddlewareHandleMessageMiddleware;
use SymfonyComponentMessengerMiddlewareSendMessageMiddleware;
use SymfonyComponentMessengerTransportSenderSendersLocator;

require_once __DIR__ . '/.. /vendor/autoload.php';
/** @var PsrContainerContainerInterface $container * / 
$container = require_once('config/container.php');

$handlers = [
    DaemonLogMessage::class => [
        new HandlerDescriptor(
            function (DaemonLogMessage $m) {
                error_log('DaemonLogHandler: message handled: / ' . $m->getMessage());
            },
            ['from_transport' => CONTAINER_REDIS_TRANSPORT_LOG]
        )
    ],
];
$middleware = [];
$middleware[] = new HandleMessageMiddleware(new HandlersLocator($handlers));
$sendersLocator = new SendersLocator(['*' => [CONTAINER_REDIS_TRANSPORT_LOG]], $container);
$middleware[] = new SendMessageMiddleware($sendersLocator);

$bus = new MessageBus($middleware);
$receivers = [
    CONTAINER_REDIS_TRANSPORT_LOG => $container->get(CONTAINER_REDIS_STREAM_RECEIVER_LOG),
];
$w = new SymfonyComponentMessengerWorker($receivers, $bus, $container->get(CONTAINER_EVENT_DISPATCHER));
$w->run();

Using this infrastructure in the application

Having implemented the bus in my backend, I separated distinct stages from the old synchronous command and created individual handlers, each focusing on its own task.

The pipeline for adding a new site to the database turned out to be as follows:

And immediately after that, it became much easier for me to add new functionality, for instance, extracting and parsing RSS. Since this process also requires the original content, the link extractor handler for RSS, just like the WebsiteIndexHistoryPersistor, subscribes to the "Content/HtmlContent" message, processes it, and passes the necessary message further along its pipeline.

Ultimately, there ended up being several daemons, each maintaining connections only to the necessary resources. For example, the daemon crawlers contains all the handlers that require fetching content from the internet, while the daemon persister holds the connection to the database.

Now, instead of selecting from the database, the necessary IDs are simply passed through the bus to all interested handlers after being inserted by the persister.

Source: habr.com