LaravelとWorkermanの統合について、公式ドキュメントではこれらを分離することを推奨していますが、実際の業務では完全に分離するのは現実的ではありません。

Gatewayに一部のロジックを追加することは可能ですが、ソースコードを見た結果、独自の実装が必要でした。

まず、composerを使用してWorkermanをインストールします。

composer require walkor/workerman

その後、Packagistでパッケージの存在を確認してください。

Consoleコマンドの作成

新しいコマンドを作成します。

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Workerman\Worker;
use App\Services\MessagePushService;

class MessagePushCommand extends Command
{
    protected $actions = ['start', 'stop', 'reload', 'status', 'connections'];

    protected $signature = 'message-push {action}';

    protected $description = 'Workermanを使ったメッセージプッシュサービス';

    public function handle()
    {
        $action = $this->argument('action');

        if (!in_array($action, $this->actions)) {
            $this->error('無効なアクションです');
            return;
        }

        MessagePushService::initialize($action);
    }
}

このコマンドをKernelに登録します。

class Kernel extends ConsoleKernel
{
    protected $commands = [
        \App\Console\Commands\MessagePushCommand::class,
    ];
}

Workermanの初期化

MessagePushServiceクラスを作成し、Workermanの初期設定を行います。

<?php

namespace App\Services;

use Illuminate\Support\Facades\DB;
use Workerman\Worker;
use Workerman\Lib\Timer;
use App\Models\User;
use App\Models\MessageLog;

class MessagePushService
{
    private static $connectionCount = 0;

    public static function initialize($action)
    {
        global $argv;

        $argv[0] = 'message-push:websocket';
        $argv[1] = $action;
        $argv[2] = '-d';

        define('HEARTBEAT_INTERVAL', 30);

        $worker = new Worker("websocket://172.17.1.247:9099");
        $worker->name = 'MessagePushWorker';
        $worker->count = 4;

        $worker->onConnect = function($connection) {
            self::$connectionCount++;
            self::handleConnection($connection);
        };

        $worker->onMessage = function($connection, $data) {
            self::handleMessage($connection, $data);
        };

        $worker->onClose = function($connection) {
            self::$connectionCount--;
            self::handleClose($connection);
        };

        Worker::runAll();
    }

    private static function handleConnection($connection)
    {
        Timer::add(10, function() use($connection) {
            if (isset($_SESSION['user_id'])) {
                $user = User::find($_SESSION['user_id']);
                if ($user) {
                    $unreadMessages = MessageLog::where('user_id', $_SESSION['user_id'])
                                                ->where('is_read', false)
                                                ->count();
                    $messages = MessageLog::where('user_id', $_SESSION['user_id'])
                                            ->where('is_read', false)
                                            ->orderBy('created_at', 'desc')
                                            ->get(['content', 'id', 'created_at'])
                                            ->toArray();

                    if ($messages) {
                        $connection->send(json_encode([
                            'code' => 200,
                            'message' => '成功',
                            'data' => $messages,
                            'connections' => self::$connectionCount,
                            'unread_messages' => $unreadMessages
                        ]));
                    }
                }
            }
        });
    }

    private static function handleMessage($connection, $data)
    {
        if ($data && is_json($data)) {
            $payload = json_decode($data, true);
            switch ($payload['type']) {
                case 'ping':
                    $connection->send(json_encode([
                        'code' => 200,
                        'message' => 'サービス稼働中',
                        'data' => [],
                        'connections' => self::$connectionCount
                    ]));
                    break;
                case 'login':
                    $user = User::find($payload['user_id']);
                    if (!$user) {
                        $connection->send(json_encode([
                            'code' => 201,
                            'message' => 'ユーザーIDが無効です',
                            'data' => [],
                            'connections' => self::$connectionCount
                        ]));
                    } else {
                        $_SESSION['user_id'] = $payload['user_id'];
                        $connection->send(json_encode([
                            'code' => 200,
                            'message' => 'ログイン成功',
                            'data' => [],
                            'connections' => self::$connectionCount
                        ]));
                    }
                    break;
            }
        } else {
            $connection->send(json_encode([
                'code' => 201,
                'message' => 'データが空です',
                'data' => [],
                'connections' => self::$connectionCount
            ]));
        }
    }

    private static function handleClose($connection)
    {
        // クローズ時の処理
    }
}

クライアントからのログインリクエストは次の形式で送ります。

{"type":"login","user_id":"24"}

ログイン後、指定されたユーザーの未読メッセージを10秒ごとにプッシュします。

サービスを起動するには以下のコマンドを使用します。

php artisan message-push start

$argv[2] = '-d';のコメントアウトを外すことでデーモンモードで動作します。