Transferring data between workers about workerman HOT 5 OPEN

<?php

require_once __DIR__ . '/vendor/autoload.php';

//php /var/www/html/chat.php start

use Channel\Client;
use Channel\Server;
use modules\chat\models\Chat;
use modules\chat\models\ChatMember;
use modules\chat\models\ChatMessage;
use modules\homeCall\models\Users;
use Workerman\Connection\TcpConnection;
use Workerman\Worker;
use yii\helpers\ArrayHelper;

const ACTION_CONNECT = 'connect';
const ACTION_SEND_MESSAGE = 'send_message';
const ACTION_CONNECT_TO_CHAT = 'connect_to_chat';

const AUTH_TYPE_KEY = 'key';
const AUTH_TYPE_TOKEN = 'token';

$websocketUrl = '127.0.0.1';
$websocketHost = '127.0.0.1';
$websocketPort = '2346';
$websocketDomain = 'homecall_php_new';
$websocketSsl = false;

$channel_server = new Server(getWebsocketHost());

$users = [];
$connectionsUsers = [];
$chatsLoggedUsers = [];
$chatsUsers = [];
$connections = [];

$dateTimeFormat = 'php:d M Y, H:i';

$debug = true;

$context = [];

if (getWebsocketSsl()) {
    $context = [
        'ssl' => [
//            'local_cert'  => Yii::$app->params['server.chat.ssl.cert'] ?? '',
//            'local_pk'    => Yii::$app->params['server.chat.ssl.pkey'] ?? '',
            'verify_peer' => false,
        ]
    ];
}

$worker = new Worker(getWebsocketAddressWorker(), $context);

if (getWebsocketSsl()) {
    $worker->transport = 'ssl';
}

$worker->count = 4;

// При запуске соединение процесса с каналами
$worker->onWorkerStart = function () {
    Client::connect(getWebsocketHost());
    Client::on('publish', function($eventData){
        onPublish($eventData);
    });
};

// Обработка события нового подключения
$worker->onConnect = function(TcpConnection $connection) {
    global $debug;
    if ($debug)
        echo 'New connection '.$connection->id. ' worker ' . $connection->worker->id . "\n";
};

// Обработка события закрытия соединения
$worker->onClose = function (TcpConnection $connection) {
    global $debug;

    $userID = getUserIDByConnection($connection);

    if(!$userID) return;

    deleteClient($connection, $userID);
    if ($debug)
        echo 'Connection closed ' . $connection->id . " worker " . $connection->worker->id . "\n";
};

// Обработка события получения данных
$worker->onMessage = function (TcpConnection $connection, $data) {
    onMessage($connection, $data);
};

// Запуск
Worker::runAll();

function onMessage(TcpConnection $connection, string $message): void
{
    global $debug;

    $data = json_decode($message, true);

    $action = $data['action'] ?? '';

    $userID = getUserIDByConnection($connection);

    if ($action !== ACTION_CONNECT && is_null($userID)) {
        $connection->send(toJson(['error' => 'У вас нет прав для доступа к чату']));
        $connection->close();
        return;
    }

    switch ($action) {
        case ACTION_CONNECT: {

            $userID = $data['user_id'];

            addClient($connection, $userID, []);

            $connection->send(toJson([
                'action' => 'init',
                'chats' => [],
            ]));

            break;
        }
        case ACTION_CONNECT_TO_CHAT: {
            $chatID = $data['chat_id'] ?? null;

            if (!Chat::checkAccess($chatID, $userID)) {
                $connection->send(toJson(['error' => 'Нет доступа к чату, попробуйте обновить страницу']));
                return;
            }

            $messageList = formatMessages(ChatMessage::find()->where(['chat_id' => $chatID])->all());

            $connection->send(toJson([
                'action' => 'init_chat',
                'messages' => $messageList,
            ]));
        }
        case ACTION_SEND_MESSAGE: {
            $chatID = $data['chat_id'] ?? null;
            $message = $data['message'] ?? null;
            $message_type_id = $data['message_type_id'] ?? null;

            if ($debug) echo 'Received message ' . print_r($data, true)."\n";

            //$count_messages = ChatMessage::find()->where(['chat_id' => $chatID])->count();
            $chat = Chat::findOne($chatID);

            if($chat->status != Chat::STATUS_ACTIVE) {
                $connection->send(toJson(['error' => 'Чат закрыт']));
                $connection->close();
                break;
            }

            $chatMessage = new ChatMessage();
            $chatMessage->user_id = $userID;
            $chatMessage->chat_id = $chatID;
            $chatMessage->message = $message;
            $chatMessage->type_id = $message_type_id;

            if (!$chatMessage->save()) {
                $connection->send(toJson($chatMessage->getErrors()));
                break;
            }

            if ($debug) echo 'Chat ID ' . $chatID . "\n" . 'Message ' . $message . "\n";

            Client::publish('publish', array(
                'chat_id' => $chatID,
                'message' => toJson([
                    'action' => 'add_message',
                    'chat_id' => $chatID,
                    'user_id' => $userID,
                    'message' => $message,
                    'message_type_id' => $message_type_id
                ]),
            ));

            break;
        }

        default: {
            $connection->send(toJson(['error' => 'Команда не найдена']));
        }
    }
}

function formatMessages($messageList)
{
    global $dateTimeFormat;

    foreach ($messageList as &$item) {
        $item['created_at'] = '';
    }

    return $messageList;
}

function formatChats($chatList, $current_user_id)
{
    global $dateTimeFormat;

    $chatListArray = [];

    foreach ($chatList as $key => $chat) {
        /** @var $chat Chat */

        $chatListArray[$key]['id'] = $chat->id;
        $chatListArray[$key]['type_id'] = $chat->type_id;
        $chatListArray[$key]['created_at'] = '';
        if($chat->type_id == Chat::TYPE_SIMPLE) {
            foreach ($chat->users as $user) {
                /** @var $user Users */
                if($user->id != $current_user_id) {
                    $chatListArray[$key]['name'] = $user->getFullName();
                    break;
                }
            }
        }
    }

    return $chatListArray;
}

function onPublish($eventData)
{
    global $debug;
    global $chatsLoggedUsers;
    global $connections;

    $chatID = $eventData['chat_id'];
    $message = $eventData['message'];

    //Получаем список ID пользователей, которые зашли в данный чат
    $userIDs = $chatsLoggedUsers[$chatID];

    foreach ($userIDs as $userID) {
        if (isset($users[$userID])) {
            foreach ($users[$userID]['connections'] as $connectionID) {

                /** @var  $connection TcpConnection */

                $connection = $connections[$connectionID];

                if($connection) $connection->send($message);

                if ($debug) echo 'Send to connection[' . $connectionID . '] - user[' . $userID . ']' . "\n";
                if ($debug) echo 'Message ' . print_r(json_decode($message), true) . "\n";
            }
        }
    }
}

/**
 * Авторизован ли клиент
 * @param TcpConnection $connection
 * @return int|null
 */
function getUserIDByConnection(TcpConnection $connection): ?int
{
    global $connectionsUsers;
    return $connectionsUsers[$connection->id . '-' . $connection->worker->id] ?? null;
}

/**
 * Добавление соединения
 * @param TcpConnection $connection
 * @param int $userID
 * @param array $chatIDs
 * @return void
 */
function addClient(TcpConnection $connection, int $userID, array $chatIDs)
{
    global $connections;
    global $users;
    global $connectionsUsers;

    // Добавляем клиента и его соединение
    $connections[$connection->id . '-' . $connection->worker->id] = $connection;

    $users[$userID]['connections'][$connection->id . '-' . $connection->worker->id] = $connection->id . '-' . $connection->worker->id;

    $users[$userID]['chats'] = $chatIDs;

    $connectionsUsers[$connection->id . '-' . $connection->worker->id] = $userID;

    foreach ($chatIDs as $chatID) {
        $chatsUsers[$chatID][$userID] = $userID;
    }

    printConnectionList();
    printClientList();
}

/**
 * Удаление соединения
 * @param TcpConnection $connection
 * @param int $userID
 * @return void
 */
function deleteClient($connection, $userID)
{
    global $connections;
    global $users;
    global $connectionsUsers;

    //Удаление соединения из списка соединений
    if($connections[$connection->id . '-' . $connection->worker->id]) unset($connections[$connection->id . '-' . $connection->worker->id]);

    unset($connectionsUsers[$connection->id . '-' . $connection->worker->id]);

    //Удаление соединения из списка пользователей
    if (isset($users[$userID]) && $users[$userID]['connections'][$connection->id . '-' . $connection->worker->id]) {
        unset($users[$userID]['connections'][$connection->id . '-' . $connection->worker->id]);
    }

    //Если соединений у пользователя больше не осталось то удаляем пользователя
    if(empty($users[$userID]['connections'])) {
        foreach ($users[$userID]['chats'] as $chatID) {
            if(!empty($chatsUsers[$chatID][$userID])) unset($chatsUsers[$chatID][$userID]);
            if(!empty($chatsLoggedUsers[$chatID][$userID])) unset($chatsLoggedUsers[$chatID][$userID]);
        }

        unset($users[$userID]);
    }

    printConnectionList();
    printClientList();
}

/**
 * Вывод пользователей в консоль
 * @return void
 */
function printClientList()
{
    global $debug;
    global $users;

    if (!$debug) return;

    echo "\n------------------------------------------\nClients: \n";
    foreach ($users as $userID => $userItem) {
        echo 'User ID: ' . $userID . " \n";
        foreach ($userItem['connections'] as $connectionID => $connectionItem) {
            echo ' - Connection ID: ' . $connectionID . " \n";
        }
        foreach ($userItem['chats'] as $chat) {
            echo ' - Chat ID: ' . $chat . " \n";
        }
    }
    echo "---------------- " . memory_get_usage() . " ---------------\n";
}

/**
 * Вывод соединений в консоль
 * @return void
 */
function printConnectionList()
{
    global $debug;
    global $connectionsUsers;

    if (!$debug) return;

    echo "\n------------------------------------------\nConnections: \n";

    foreach ($connectionsUsers as $connectionID => $userID) {
        echo 'Connection ID ' . $connectionID . ' User ID ' . $userID . " \n";
    }

    echo "---------------- " . memory_get_usage() . " ---------------\n";
}

/**
 * Формирование ответа в JSON
 * @param array $data
 * @return false|string
 */
function toJson(array $data) {
    return json_encode($data, JSON_UNESCAPED_UNICODE);
}

function getWebsocketHost()
{
    global $websocketHost;
    return $websocketHost;
}

function getWebsocketAddressWorker(bool $cli = false): string
{
    if ($cli) {
        return (getWebsocketSsl() ? 'wss://' : 'ws://').getWebsocketDomainWorker().':'.getWebsocketPort();
    }
    return 'websocket://'.getWebsocketDomainWorker().':'.getWebsocketPort();
}

function getWebsocketSsl(): bool
{
    global $websocketSsl;
    return $websocketSsl;
}

function getWebsocketDomain(): string
{
    global $websocketDomain;
    return $websocketDomain;
}

function getWebsocketPort(): string
{
    global $websocketPort;
    return $websocketPort;
}

function getWebsocketDomainWorker(): string
{
    global $websocketDomain;
    return $websocketDomain;
}

$this->worker->onMessage = function (TcpConnection $connection, $data) {
            $this->onMessage($connection, $data);
        };