zhishifufei_php/application/push/controller/Events.php

<?php
// +----------------------------------------------------------------------
// | 天诚科技 [ 刘海东 17600099397赋能开发者，助力企业发展 ]
// +----------------------------------------------------------------------
// | Copyright (c) 2016~2020 https://www.tczxkj.com All rights reserved.
// +----------------------------------------------------------------------
// | Licensed 该系统并不是自由软件，未经许可不能去掉相关版权
// +----------------------------------------------------------------------
// | Author:甘肃天诚志信电子商务有限公司 刘海东 联系电话维系17600099397
// +----------------------------------------------------------------------

namespace app\push\controller;

use app\kefu\model\KefuModel;
use app\kefu\model\KefuUserRecordModel;
use GatewayWorker\Lib\Gateway;
use think\Hook;
use think\Request;
use Workerman\Lib\Timer;
use app\wap\model\live\LiveUser;
use think\Log;
use think\Session;

class Events
{

    //定时器间隔
    protected static $interval = 2;
    //定时器
    protected static $timer = null;

    /**
     *
     * @var object
     */
    protected static $worker;

    /**
     * 事件处理类
     * @var string
     */
    protected static $evevtRunClass = \app\push\controller\EvevtRun::class;

    /**
     * 消息事件回调
     * @var string
     */
    protected static $eventClassName = [
        'kefu' => \app\push\controller\KefuHandler::class,
        'live' =>  \app\push\controller\Push::class
    ];

    /**
     * 当客户端发来消息时触发
     * @param int $client_id 连接id
     * @param mixed $message 具体消息
     */
    public static function onMessage($client_id, $message)
    {
        $message_data = json_decode($message, true);
        if (!$message_data) return;
        try {
            if (!isset($message_data['type'])) throw new \Exception('缺少消息参数类型');
            //消息回調处理

            $channel = isset($message_data['channel']) ? $message_data['channel'] : 'live';

            $className = self::$eventClassName[$channel];

            $evevtName = $className . '::instance';
            if (is_callable($evevtName))
                $evevtName()->start($message_data['type'], $client_id, $message_data);
            else
                throw new \Exception('消息处理回调不存在。[' + $evevtName + ']');
        } catch (\Exception $e) {
            var_dump([
                'file' => $e->getFile(),
                'code' => $e->getCode(),
                'msg' => $e->getMessage(),
                'line' => $e->getLine()
            ]);
        }
    }

    /**
     * 当用户连接时触发的方法
     * @param integer $client_id 连接的客户端
     * @return void
     */
    public static function onConnect($client_id)
    {

        Gateway::sendToClient($client_id, json_encode(array(
            'type' => 'init',
            'client_id' => $client_id
        )));
    }

    /**用户进入时连接
     * @param $client_id
     * @param $data
     */
    public static function onWebSocketConnect($client_id, $data)
    {
        $channel = array_key_exists('channel', $data['get']) ? $data['get']['channel'] : "";
        if ($channel === "kefu") {
            // 客服
            // Log::write($kefuId);
        } else {
            // live
            Gateway::bindUid($client_id, $data['get']['uid']);
            LiveUser::setLiveUserOnline($data['get']['room'], $data['get']['uid'], 1);
        }
    }

    /**
     * 当用户断开连接时触发的方法
     * @param integer $client_id 断开连接的客户端
     * @return void
     */
    public static function onClose($client_id)
    {
        if (isset($_SESSION['is_kefu']) && isset($_SESSION['uid'])) {
            $is_kefu = $_SESSION['is_kefu'];
            $uid = $_SESSION['uid'];

            if ($is_kefu) {
                KefuUserRecordModel::update([
                    'online' => 0
                ], [
                    'is_kefu_send' => 0,
                    'to_uid' => $uid
                ]);

                KefuModel::update([
                    'online' => 0
                ], [
                    'id' => $uid
                ]);
            } else {
                KefuUserRecordModel::update([
                    'online' => 0
                ], [
                    'is_kefu_send' => 1,
                    'to_uid' => $uid
                ]);

                if (isset($_SESSION['chat_kefu_uid']) && count($_SESSION['chat_kefu_uid'])) {
                    
                    Gateway::sendToUid($_SESSION['chat_kefu_uid'], json_encode([
                        'type' => 'offline',
                        'data' => [
                            'self' => false,
                            'uid' => $uid
                        ]
                    ]));
                }
            }
        }

        Gateway::sendToClient($client_id, json_encode([
            'type' => 'logout',
            'message' => "client[$client_id]"
        ]));
    }

    /**
     * 当进程启动时
     * @param integer $businessWorker 进程实例
     */
    public static function onWorkerStart($worker)
    {
        //在进程1上开启定时器 每self::$interval秒执行
        self::$worker = $worker;
        if ($worker->id === 0) {
            $last = time();
            $task = [6 => $last, 10 => $last, 30 => $last, 60 => $last, 180 => $last, 300 => $last];
            self::$timer = Timer::add(self::$interval, function () use (&$task) {
                try {
                    $now = time();
                    Hook::exec(self::$evevtRunClass);
                    foreach ($task as $sec => &$time) {
                        if (($now - $time) >= $sec) {
                            $time = $now;
                            Hook::exec(self::$evevtRunClass, 'task_' . $sec);
                        }
                    }
                } catch (\Throwable $e) {
                }
            });
        }
    }

    /**
     * 当进程关闭时
     * @param integer $businessWorker 进程实例
     */
    public static function onWorkerStop($worker)
    {
        if ($worker->id === 0) Timer::del(self::$timer);
    }
}