xyzshop/helper/room/room_processor.php

<?php
/**
 * Created by PhpStorm.
 * User: stanley-king
 * Date: 2017/12/14
 * Time: 下午12:07
 */

namespace room;

use event\IProcessor;
use process_looper;
use errcode;
use algorithm;
use QueueClient;
use scope_trace;
use Log;

class room_processor implements IProcessor
{
    private $mAccConnes;
    private $mFactory;
    private $mRooms;
    private $mChatwo;

    public function __construct()
    {
        $this->mAccConnes = [];
        $this->mFactory = new factory();
        $this->mRooms = [];
        $this->mChatwo = new chatwo();
    }
    
    public function onStart()
    {
    }
    public function onConnected($bufid,$stream,$host,$port,$args)
    {
    }
    public function onClose($bufid)
    {
        $this->remove_acc($bufid);
    }
    public function onRequest($bufid, $body)
    {
        $trace = new scope_trace(__METHOD__);
        $input = json_decode($body,true);
        if($input == false) {
            process_looper::instance()->close($bufid);
            return false;
        }

        $act = $input['act'];
        if(empty($act)) return false;

        if($act == proto_type::act_factory) {
            $ret = $this->onFactory($bufid,$input);
            process_looper::instance()->write($bufid,$ret);
            return true;
        }
        elseif($act == proto_type::act_access) {
            $ret = $this->onAccess($bufid,$input);
            process_looper::instance()->write($bufid,$ret);
            return true;
        }
        elseif($act == proto_type::act_room) {
            $ret = $this->onRoom($bufid,$input);
            process_looper::instance()->write($bufid,$ret);
            return true;
        }
        elseif($act == proto_type::act_chatwo) {
            $ret = $this->onChatwo($bufid,$input);
            return true;
        }
        else {
            return false;
        }
    }
    ////////////////////////////////////////Factory Message Handler/////////////////////////////////////////////////////
    private function onFactory($bufid,$input)
    {
        $op = $input['op'];
        if($op == 'build') {
            return $this->factory_build($bufid,$input);
        }
        elseif($op == 'list_room') {
            return $this->success(['rooms' => $this->mRooms]);
        }
        elseif($op == 'invite') {
            return $this->factory_invite($bufid,$input);
        }
        elseif($op == 'change') {
            return $this->factory_change($bufid,$input);
        }
        elseif($op == 'notice_room') {
            return $this->factory_notice_room($bufid,$input);
        }
        elseif($op == 'notice_users') {
            return $this->factory_notice_users($bufid,$input);
        }
        elseif($op == 'notice_all') {
            return $this->factory_notice_all($bufid,$input);
        }
        elseif($op == 'leave') {
            return $this->factory_leave($bufid,$input);
        }
        elseif($op == 'kickout') {
            return $this->factory_kickout($bufid,$input);
        }
        else {
            return $this->error(errcode::ErrRoomFactoryOp);
        }
    }
    /**
     * @param $input
     * @return mixed|string
     */
    private function factory_build($bufid,$input)
    {
        $roomid = intval($input['room']);
        $newroom = boolval($input['newroom']);

        if ($roomid <= 0) {
            return $this->error(errcode::ErrRoomParam);
        }
        if (array_key_exists($roomid, $this->mRooms)) {
            return $this->success(['room' => $roomid]);
        }

        $room = $this->mFactory->build($roomid);
        if ($room == false) {
            return $this->error(errcode::ErrRoomBuild);
        }
        else {
            $this->broad_access(msg_builder::build_message($roomid, $room->room_type()));
            $this->mRooms[$roomid] = $room;
            if ($newroom) {
                $this->broad_access(msg_builder::create_push($room->creator(), $roomid, $room->room_info()));
            }
            return $this->success(['room' => $room->room_id()]);
        }
    }
    /**
     * @param $input
     * @return mixed|string
     */
    private function factory_invite($bufid,$input)
    {
        $roomid  = intval($input['room']);
        $inviter = intval($input['inviter']);

        if ($roomid <= 0) {
            return $this->error(errcode::ErrRoomParam);
        }

        $room = $this->room($roomid);
        if ($room != false)
        {
            $invitees = $input['invitees'];
            if ($inviter > 0 && !empty($invitees))
            {
                $result = $room->invite($inviter, $invitees,$newusers);
                if ($result != false)
                {
                    if (!empty($newusers))
                    {
                        $room_info = $room->room_info();
                        $this->broad_access(msg_builder::invite_message($roomid, $room->room_type(), $newusers));

                        $inviter = $room->userinfos($inviter);
                        $this->broad_access(msg_builder::invited_push($roomid, $newusers,$inviter,$room_info));
                        $this->broadcast_roomsg($room);
                        $left = $room->usercount(); $right = count($newusers);
                        $last_count = $left - $right;
                        Log::record("ready to update -------$last_count||$left||$right|| ",Log::DEBUG);
                        if($last_count >= 0 && $last_count < 10) {
                            QueueClient::push("OnUpdateRoom",['room_id' => $roomid]);
                        } else {
                            $count = count($newusers);
                            Model('room')->editRoom(['room_id' => $roomid],['users' => ['exp',"users+{$count}"]]);
                        }
                    }
                    return $this->success($result);
                }
            }
        }

        return $this->error(errcode::ErrRoomInvite);
    }

    private function factory_leave($bufid,$input)
    {
        $roomid = intval($input['room']);
        if ($roomid <= 0) {
            return $this->error(errcode::ErrRoomParam);
        }

        $room = $this->room($roomid);
        if ($room != false) {
            $userid = intval($input['user']);
            if ($userid > 0) {
                $result = $room->leave($userid);
                if ($result != false)
                {
                    $this->broad_access(msg_builder::leave_message($roomid, $room->room_type(), [$userid]));
                    $this->reply_roomsg($bufid,$room);
                    $this->broadcast_roomsg($room);

                    $user_count = $room->usercount();
                    if($user_count >= 0 && $user_count < 10) {
                        QueueClient::push("OnUpdateRoom",['room_id' => $roomid]);
                    }

                    return $this->success($result);
                }
            }
        }

        return $this->error(errcode::ErrRoomLeave);
    }

    private function factory_kickout($bufid,$input)
    {
        $roomid = intval($input['room']);
        if ($roomid <= 0) {
            return $this->error(errcode::ErrRoomParam);
        }
        $room = $this->room($roomid);

        if ($room != false)
        {
            $user =  intval($input['user']);
            $kicks = $input['kicks'];

            if ($user > 0)
            {
                $users = $room->kickout($user,$kicks);
                if ($users != false)
                {
                    $this->broad_access(msg_builder::leave_message($roomid, $room->room_type(), $users));
                    $this->broadcast_roomsg($room);

                    $user_count = $room->usercount();
                    if($user_count >= 0 && $user_count < 10) {
                        QueueClient::push("OnUpdateRoom",['room_id' => $roomid]);
                    }
                    return $this->success(['users' => $users]);
                }
                else {
                    return $this->success(['users' => []]);
                }
            }
        }

        return $this->error(errcode::ErrRoomLeave);
    }

    private function factory_change($bufid,$input)
    {
        $roomid = intval($input['room']);
        if ($roomid <= 0) {
            return $this->error(errcode::ErrRoomParam);
        }

        $room = $this->room($roomid);
        if ($room != false)
        {
            $result = $room->change();
            if ($result != false) {
                $this->broad_access(msg_builder::change_push($roomid, $room->room_info()));
                return $this->success($result);
            }
        }

        return $this->error(errcode::ErrRoomChange);
    }

    private function factory_notice_room($bufid, $input)
    {
        $roomid = intval($input['room']);
        $type = $input['type'];
        $content = $input['content'];

        if ($roomid <= 0 || empty($type) || empty($content)) {
            return $this->error(errcode::ErrRoomParam);
        }

        $room = $this->room($roomid);
        if ($room != false)
        {
            $result = $room->notice($type,$content);
            if($result) {
                $this->broadcast_roomsg($room);
                return $this->success($result);
            } else {
                return $this->error(errcode::ErrRoomNotice);
            }
        }

        return $this->error(errcode::ErrRoomChange);
    }

    ///Push消息处理区/////////////////////////////////////////////////////////////////////////////////////////////////////
    private function factory_notice_users($bufid, $input)
    {
        $content = $input['content'];
        $msg = $input['msg'];
        $users = $input['users'];

        $type = $input['type'];
        if($type === proto_type::push_command) {
            $pushid = 1;
            $msg_type = proto_type::msg_type_command;
            $send = $content;
        }
        elseif($type === proto_type::push_notify) {
            $pushid = 2;
            $msg_type = proto_type::msg_type_nofity;
            $send[] = $msg;
        }
        elseif($type === proto_type::push_apply) {
            $pushid = 3;
            $msg_type = proto_type::msg_type_apply;
            $send[] = $msg;
        }
        else {
            return $this->error(errcode::ErrParamter);
        }

        if($msg_type != proto_type::msg_type_command)
        {
            $mod_room = Model('room');

            $orgmsg = json_encode($content);
            foreach ($users as $user) {
                $msgid = $mod_room->addRoomMsg(['room_id' => 0,'member_id' => $user, 'type' => $msg_type,'msg' => $msg, 'orgmsg' => $orgmsg, 'add_time' => time(),'msg_type' => 1]);
                $send['msgid'] = $msgid;
                $send['send_time'] = time();
                $this->write_push_users($pushid,$users,$send);
            }
        }

        return $this->success(NULL);
    }

    private function factory_notice_all($bufid, $input)
    {
        $content = $input['content'];
        $type = $input['type'];
        if($type === proto_type::push_command) {
            $pushid = 1;
        }
        elseif($type === proto_type::push_notify) {
            $pushid = 2;
        }
        elseif($type === proto_type::push_apply) {
            $pushid = 3;
        }
        else {
            return $this->error(errcode::ErrParamter);
        }

        $this->write_push($content);
        return $this->success(NULL);
    }
    ////////////////////////////////////////Access Message Handler//////////////////////////////////////////////////////
    private function onAccess($bufid,$input)
    {
        $op = $input['op'];
        if($op == 'list') {
            $this->add_acc($bufid);
            $reply = ['act' => 'access','op' => $op,'rooms' => $this->rooms()];
            return $this->success($reply);
        }
        else {
            return $this->error(errcode::ErrRoomAccessOp);
        }
    }
    private function rooms()
    {
        $result = [];
        foreach ($this->mRooms as $roomid => $room) {
            $result[] = $roomid;
        }
        return $result;
    }
    ////////////////////////////////////////Room Message Handler////////////////////////////////////////////////////////
    private function onRoom($bufid,$input)
    {
        $roomid = intval($input['room']);
        $room = $this->room($roomid);

        if($room != false)
        {
            $function = $input['op'] . 'Op';
            if (method_exists($room,$function))
            {
                $ret = $room->$function($input);
                if($ret != false) {
                    $this->reply_roomsg($bufid,$room);
                    $this->broadcast_roomsg($room);
                }
                return $this->success($ret);
            }
        }

        return $this->error(errcode::ErrRoom);
    }
    ////////////////////////////////////////Peer Message Handler////////////////////////////////////////////////////////
    private function onChatwo($bufid, $input)
    {
        $function = $input['op'] . 'Op';
        if (method_exists($this->mChatwo,$function))
        {
            $msg = $this->mChatwo->$function($input);
            if($msg != false)
            {
                $body = json_encode($msg);
                foreach ($this->mAccConnes as $conn) {
                    process_looper::instance()->write($conn,$body);
                }
            }

            return true;
        }
    }

    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    private function write_push_users($pushid,$users,$content)
    {
        $msg = [];

        $msg['act']  = "room";
        $msg['op']   = "relay";
        $msg['relay_type'] = "users";
        $msg['receivers'] = $users;
        $msg['room'] = 0;
        $msg['receiver_type'] = proto_type::sroom_push;
        $msg['msgtype'] = "message";


        $msg['body']['act']  = proto_type::act_push;
        $msg['body']['op']   = 'message';
        $msg['body']['push']   = $pushid;
        $msg['body']['content'] = $content;
        $msg['body']['msgtype'] = "message";

        $body = json_encode($msg);
        foreach ($this->mAccConnes as $conn) {
            process_looper::instance()->write($conn,$body);
        }
    }
    private function write_push($content)
    {
        $msg = [];

        $msg['act']  = "room";
        $msg['op']   = "relay";
        $msg['relay_type'] = "alluser";
        $msg['receivers'] = [];
        $msg['room'] = 0;
        $msg['receiver_type'] = proto_type::sroom_push;
        $msg['msgtype'] = "message";


        $msg['body']['act']  = proto_type::act_push;
        $msg['body']['op']   = 'message';
        $msg['body']['push']   = 0;
        $msg['body']['content'] = $content;
        $msg['body']['msgtype'] = "message";

        $body = $this->success($msg);
        foreach ($this->mAccConnes as $conn) {
            process_looper::instance()->write($conn,$body);
        }
    }
    ////////////////////////////////返回包///////////////////////////////////////////////////////////////////////////////
    private function success($val)
    {
        $code = errcode::Success;
        $data['code'] = $code;
        $data['message'] = errcode::msg($code);
        $data['data'] = $val;
        $data['msgtype'] = "reply";

        return json_encode($data);
    }
    private function error($code,$message = '')
    {
        if (empty($message)) {
            $message = errcode::msg($code);
        }

        $data['code'] = $code;
        $data['message'] = $message;
        $data['data'] = null;
        $data['msgtype'] = "reply";

        return json_encode($data);
    }
    private function reply_roomsg($bufid,base_room $room)
    {
        $retmsgs = $room->relay_reply_msgs();
        foreach ($retmsgs as $msg) {
            $body = json_encode($msg);
            process_looper::instance()->write($bufid,$body);
        }
    }
    private function broad_access($messages)
    {
        $body = json_encode($messages);
        foreach ($this->mAccConnes as $bufid) {
            process_looper::instance()->write($bufid,$body);
        }
    }
    private function broadcast_roomsg(base_room $room)
    {
        $broad_msgs = $room->relay_broadcast_msgs();
        foreach ($broad_msgs as $msg)
        {
            $body = json_encode($msg);
            foreach ($this->mAccConnes as $bufid) {
                process_looper::instance()->write($bufid,$body);
            }
        }
    }
    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    private function room($roomid)
    {
        if($roomid <= 0) return false;
        if(array_key_exists($roomid,$this->mRooms)) {
            return $this->mRooms[$roomid];
        } else {
            return false;
        }
    }

    private function add_acc($bufid) {
        $connid = intval($bufid);
        if($connid <= 0) return false;

        if(!algorithm::binary_search($this->mAccConnes,$connid)) {
            $pos = algorithm::lower_bonud($this->mAccConnes,$connid);
            algorithm::array_insert($this->mAccConnes,$pos,$connid);
        }
        return true;
    }
    private function remove_acc($bufid)
    {
        $connid = intval($bufid);
        if($connid <= 0) return false;

        if(algorithm::binary_search($this->mAccConnes,$connid)) {
            $pos = algorithm::lower_bonud($this->mAccConnes,$connid);
            algorithm::array_erase($this->mAccConnes,$pos);
        }
        return true;
    }
}