Превращение событий PostgreSQL в события Laravel05.03.2024 23:00

Пишем логику команды listen: notify

Пойдём от простого к сложному. Для начала просто получим сообщение от PostgreSQL.

Для этого дорабатываем метод handle, не забывая доработать поля $signature и $description

Содержимое файла app/Console/Commands/ListenNotifyCommand.php

getPdo();

        // Listen to the 'my_channel' notifications
        $pdo->exec("LISTEN my_event");
        $this->info('Starting');
        // Forever loop
        while (true) {
            $notification = $pdo->pgsqlGetNotify(PDO::FETCH_ASSOC, 10000);

            if ($notification) {
                $this->info('Received notification: ' . json_encode($notification, JSON_THROW_ON_ERROR));
            }
        }
    }
}

Проверяем, работает ли наша команда. Запускаем в терминале команду: php artisan listen: notify и в третьем терминале, (мы его не закрывали) вновь подаём NOTIFY my_event, 'Hello, PostgreSQL!';

Отлично! Работает!

Начало положено. Приложение на Laravel получило событие из PostgreSQL.

При изменении кода команды, не забывайте перезапускать процесс.

Добавляем обработку сигналов

Немного о сигналах, гуру могут пропустить

Добавляем два поля ($hasPcntl, $running) типа bool в класс ListenNotifyCommand и инициализируем их. Пишем метод — обработчик сигналов

    protected bool $hasPcntl = false;
    protected bool $running = true;

    private function handleSignal(int $signal): void
    {
        switch ($signal) {
            case SIGINT:
            case SIGTERM:
                $this->info( PHP_EOL . 'Received stop signal, shutting down...');
                $this->running = false;
                break;

            default:
        }
    }

Для обработки сигналов потребуется расширение pcntl. Данное расширение недоступно для Windows, тем не менее, вполне возможно написать кроссплатформенное решение.

Дорабатываем метод handle

    public function handle(): int
    {
        // Проверка, что модуль pcntl подключён
        $this->hasPcntl = extension_loaded('pcntl');

        if ($this->hasPcntl) {
            // Если модуль pcntl подключён, назначаем обработчики сигналов
            pcntl_signal(SIGINT, [$this, 'handleSignal']);
            pcntl_signal(SIGTERM, [$this, 'handleSignal']);
        }

        $pdo = DB::connection()->getPdo();
        $pdo->exec("LISTEN my_event");
        $this->info('Start listening');

        while ($this->running) {
            $notification = $pdo->pgsqlGetNotify(PDO::FETCH_ASSOC, 10000);
            $this->info('iter');
            if ($notification) {
                $this->info('Received notification: ' . json_encode($notification, JSON_THROW_ON_ERROR));
            }

            if ($this->hasPcntl) {
                // Если модуль pcntl подключён, вызываем обработчики сигналов
                pcntl_signal_dispatch();
            }
        }
        // Возвращаем 0, как код завершения
        return 0;
    }

Файл app/Console/Commands/ListenNotifyCommand.php

hasPcntl = extension_loaded('pcntl');

        if ($this->hasPcntl) {
            pcntl_signal(SIGINT, [$this, 'handleSignal']);
            pcntl_signal(SIGTERM, [$this, 'handleSignal']);
        }

        $pdo = DB::connection()->getPdo();
        $pdo->exec("LISTEN my_event");
        $this->info('Start listening');

        while ($this->running) {
            $notification = $pdo->pgsqlGetNotify(PDO::FETCH_ASSOC, 10000);
            $this->info('iter');
            if ($notification) {
                $this->info('Received notification: ' . json_encode($notification, JSON_THROW_ON_ERROR));
            }

            if ($this->hasPcntl) {
                pcntl_signal_dispatch();
            }
        }
        return 0;
    }

    private function handleSignal(int $signal): void
    {
        switch ($signal) {
            case SIGINT:
            case SIGTERM:
                $this->info( PHP_EOL . 'Received stop signal, shutting down...');
                $this->running = false;
                break;

            default:
        }
    }

}

Можете запустить команду, а также вызвать в соседней консоли NOTIFY, всё должно работать. Если команда запущена в Linux и модуль pcntl подключён, то при нажатии Ctrl+C будет выведено сообщение: Received stop signal, shutting down… Это значит, что скрипт корректно обрабатывает сигналы и останавливается, вместо того, чтобы быть принудительно остановленным.

Настраиваем Supervisor для наблюдения за скриптом

Supervisor — это удобный инструмент для управления фоновыми процессами в операционных системах, похожих на Unix. Он отслеживает работу скрипта, автоматически перезапускает его при сбое и даёт возможность управлять его состоянием, такими как запуск, остановка и перезагрузка.

Supervisor также совместим с сигналами Unix, что дает возможность настраивать поведение процесса в зависимости от разных сигналов. Mы настроили скрипт на обработку сигналов SIGINT и SIGTERM, чтобы правильно завершить его, что хорошо согласуется с Supervisor.

Когда Supervisor отправляет сигнал SIGTERM процессу, он ожидает, что процесс завершит свою работу и передаст управление обратно системе.

Если процесс успешно обработал SIGTERM и корректно завершил свою работу, обычно возвращается код завершения 0.

Если процесс не вернул управление за разумное время (по умолчанию 10 секунд), Supervisor посылает SIGKILL. Это время можно изменить в настройках, опция stopwaitsecs.

Примерный файл конфигурации Supervisor

[program:postrgres_laravel]
process_name=%(program_name)s_%(process_num)02d
command=php /path/to/your/laravel/artisan listen:notify
autostart=true
autorestart=true
user=www-data
numprocs=1
redirect_stderr=true
stdout_logfile=/var/log/postrgres_laravel.log

Подробнее про Supervisor в документации Laravel.

Сериализуем полезную нагрузку

Аргумент для NOTIFY — всегда строка. Т.е. если мы хотим передать что-либо cложное, нам нужно это сложное — сериализовать. PostgreSQL умеет работать с JSON, давайте используем это умение.

Создадим хранимую функцию, которая принимает на вход json, и отправляет его в канал my_event.

Файл миграции 2024_03_05_125805_create_send_notify_function.php

php artisan migrate

            if ($notification) {
                $this->info('Received notification: ' . json_encode($notification, JSON_THROW_ON_ERROR));
                $payload = json_decode($notification['payload'], true, 512, JSON_THROW_ON_ERROR);
                $this->info('Decoded payload: ' . print_r($payload, true));
            }

select send_notify(json_build_object('key1', 'Hello, PostgreSQL!'));

select send_notify(json_build_object('key1', 'Hello, PostgreSQL!', 'key2', json_build_object('key2_inner', 2, 'key3_inner', 3)))

            if ($notification) {
                $this->info('Received notification: ' . json_encode($notification, JSON_THROW_ON_ERROR));
                $payload = json_decode($notification['payload'], true, 512, JSON_THROW_ON_ERROR);
                $this->info('Decoded payload: ' . print_r($payload, true));
                // Новая строка
                Event::dispatch(new PostgresNotificationReceived($payload));
            }

hasPcntl = extension_loaded('pcntl');

        if ($this->hasPcntl) {
            pcntl_signal(SIGINT, [$this, 'handleSignal']);
            pcntl_signal(SIGTERM, [$this, 'handleSignal']);
        }

        $pdo = DB::connection()->getPdo();
        $pdo->exec("LISTEN my_event");
        $this->info('Start listening');

        while ($this->running) {
            $notification = $pdo->pgsqlGetNotify(PDO::FETCH_ASSOC, 10000);

            if ($notification) {
                $this->info('Received notification: ' . json_encode($notification, JSON_THROW_ON_ERROR));
                $payload = json_decode($notification['payload'], true, 512, JSON_THROW_ON_ERROR);
                $this->info('Decoded payload: ' . print_r($payload, true));
                Event::dispatch(new PostgresNotificationReceived($payload));
            }

            if ($this->hasPcntl) {
                pcntl_signal_dispatch();
            }
        }

        return 0;

    }

    private function handleSignal(int $signal): void
    {
        switch ($signal) {
            case SIGINT:
            case SIGTERM:
                $this->info( PHP_EOL . 'Received stop signal, shutting down...');
                $this->running = false;
                break;

            default:
        }
    }

}

select send_notify(json_build_object('key1', 'Hello, PostgreSQL!', 'key2', json_build_object('key2_inner', 2, 'key3_inner', 3)));