infantServiceApi/app/services/workerman/chat/ChatService.php

<?php
// +----------------------------------------------------------------------
// | Chat 客服聊天服务
// +----------------------------------------------------------------------
namespace app\services\workerman\chat;

use Channel\Client;
use app\services\workerman\ChannelService;
use app\services\workerman\Response;
use Workerman\Connection\TcpConnection;
use Workerman\Lib\Timer;
use Workerman\Worker;

class ChatService
{
    protected $worker;
    protected $connections = [];
    protected $user = [];
    protected $kefuUser = [];
    protected $handle;
    protected $response;
    protected $timer;

    public function __construct(Worker $worker)
    {
        $this->worker = $worker;
        $this->handle = new ChatHandle($this);
        $this->response = new Response();
    }

    public function setUser(TcpConnection $connection)
    {
        $this->user[$connection->user->uid] = $connection;
    }

    public function kefuUser()
    {
        return $this->kefuUser;
    }

    public function setKefuUser(TcpConnection $connection, bool $isUser = true)
    {
        $this->kefuUser[$connection->kefuUser->uid] = $connection;
        if ($isUser) {
            $this->user[$connection->user->uid] = $connection;
        }
    }

    public function user($key = null)
    {
        return $key ? ($this->user[$key] ?? false) : $this->user;
    }

    public function onConnect(TcpConnection $connection)
    {
        $this->connections[$connection->id] = $connection;
        $connection->lastMessageTime = time();
    }

    public function onMessage(TcpConnection $connection, $res)
    {
        $connection->lastMessageTime = time();
        $res = json_decode($res, true);
        if (!$res || !isset($res['type']) || !$res['type'] || $res['type'] == 'ping') {
            return $this->response->connection($connection)->success('ping', ['now' => time()]);
        }
        if (!method_exists($this->handle, $res['type'])) return;
        try {
            $this->handle->{$res['type']}($connection, $res + ['data' => []], $this->response->connection($connection));
        } catch (\Throwable $e) {
            // 忽略异常
        }
    }

    public function onWorkerStart(Worker $worker)
    {
        ChannelService::connet();

        Client::on('muyinjie_chat', function ($eventData) use ($worker) {
            if (!isset($eventData['type']) || !$eventData['type']) return;
            $ids = isset($eventData['ids']) && count($eventData['ids']) ? $eventData['ids'] : array_keys($this->user);
            foreach ($ids as $id) {
                if (isset($this->user[$id])) {
                    $this->response->connection($this->user[$id])->success($eventData['type'], $eventData['data'] ?? null);
                }
            }
        });

        $this->timer = Timer::add(15, function () use (&$worker) {
            $time_now = time();
            foreach ($worker->connections as $connection) {
                if ($time_now - $connection->lastMessageTime > 12) {
                    $this->response->connection($connection)->close('timeout');
                }
            }
        });
    }

    public function onClose(TcpConnection $connection)
    {
        unset($this->connections[$connection->id]);
        if (isset($connection->user->uid)) {
            unset($this->user[$connection->user->uid]);
        }
        if (isset($connection->kefuUser->uid)) {
            unset($this->kefuUser[$connection->kefuUser->uid]);
        }
    }
}