src/SMProxyServer.php
public function onReceive(\swoole_server $server, int $fd, int $reactor_id, string $data)
    {
        self::go(function () use ($server, $fd, $reactor_id, $data) {
            if (!isset($this->source[$fd]->auth)) {
                throw new SMProxyException('Must be connected before sending data!');
            }
            if (!isset($this->halfPack[$fd])) {
                $this->halfPack[$fd] = '';
            }
            if ($this->source[$fd]->auth) {
                $headerLength = 4;
            } else {
                $headerLength = 3;
            }
            $packages = packageSplit($data, $this->source[$fd]->auth ?: false, $headerLength, false, $this->halfPack[$fd]);
            foreach ($packages as $package) {
                $data = $package;
                self::go(function () use ($server, $fd, $reactor_id, $data) {
                    $bin = (new MySqlPacketDecoder())->decode($data);
                    if (!$this->source[$fd]->auth) {
                        $this->auth($bin, $server, $fd);
                    } else {
                        $this->query($bin, $data, $fd);
                        if (isset($this->connectReadState[$fd]) && true === $this->connectReadState[$fd]) {
                            $model = 'read';
                        } else {
                            $model = 'write';
                        }
                        $key = $this ->compareModel($model, $server, $fd);
                        if ($data) {
                            if (isset($this->mysqlClient[$fd][$key])) {
                                $this->mysqlClient[$fd][$key]->send($data);
                            } else {
                                $client = MySQLPool::fetch($key, $server, $fd);
                                $result = $client->send($data);
                                if ($result) {
                                    $this->mysqlClient[$fd][$key] = $client;
                                }
                            }
                        }
                        //预处理语句id记录
                        if (isset($this->mysqlClient[$fd][$key])) {
                            $clientId = spl_object_hash($this->mysqlClient[$fd][$key]);
                            switch ($bin ->data[4]) {
                                case MysqlPacket::$COM_STMT_PREPARE:
                                    if (isset($this->stmtId[$clientId])) {
                                        $this->stmtId[$clientId]++;
                                    } else {
                                        $this->stmtId[$clientId] = 1;
                                    }
                                    $this->stmtPrepare[$clientId][$this->stmtId[$clientId]] = $this->stmtId[$clientId];
                                    break;
                                case MySQLPacket::$COM_STMT_CLOSE:
                                    $closeStmtId = getPackageLength($data, 5, 4) - 4;
                                    unset($this->stmtPrepare[$clientId][$closeStmtId]);
                                    break;
                            }
                            unset($clientId);
                        }
                    }
                });
            }
        });
    }
public function onReceive(\swoole_server $server, int $fd, int $reactor_id, string $data)
    {
        self::go(function () use ($server, $fd, $reactor_id, $data) {
            if (!isset($this->source[$fd]->auth)) {
                throw new SMProxyException('Must be connected before sending data!');
            }
            if (!isset($this->halfPack[$fd])) {
                $this->halfPack[$fd] = '';
            }
            if ($this->source[$fd]->auth) {
                $headerLength = 4;
            } else {
                $headerLength = 3;
            }
            $packages = packageSplit($data, $this->source[$fd]->auth ?: false, $headerLength, false, $this->halfPack[$fd]);
            foreach ($packages as $package) {
                $data = $package;
                self::go(function () use ($server, $fd, $reactor_id, $data) {
                    $bin = (new MySqlPacketDecoder())->decode($data);
                    if (!$this->source[$fd]->auth) {
                        $this->auth($bin, $server, $fd);
                    } else {
                        $this->query($bin, $data, $fd);
                        if (isset($this->connectReadState[$fd]) && true === $this->connectReadState[$fd]) {
                            $model = 'read';
                        } else {
                            $model = 'write';
                        }
                        $key = $this ->compareModel($model, $server, $fd);
                        if ($data) {
                            if (isset($this->mysqlClient[$fd][$key])) {
                                $this->mysqlClient[$fd][$key]->send($data);
                            } else {
                                $client = MySQLPool::fetch($key, $server, $fd);
                                $result = $client->send($data);
                                if ($result) {
                                    $this->mysqlClient[$fd][$key] = $client;
                                }
                            }
                        }
                        //预处理语句id记录
                        if (isset($this->mysqlClient[$fd][$key])) {
                            $clientId = spl_object_hash($this->mysqlClient[$fd][$key]);
                            switch ($bin ->data[4]) {
                                case MysqlPacket::$COM_STMT_PREPARE:
                                    if (isset($this->stmtId[$clientId])) {
                                        $this->stmtId[$clientId]++;
                                    } else {
                                        $this->stmtId[$clientId] = 1;
                                    }
                                    $this->stmtPrepare[$clientId][$this->stmtId[$clientId]] = $this->stmtId[$clientId];
                                    break;
                                case MySQLPacket::$COM_STMT_CLOSE:
                                    $closeStmtId = getPackageLength($data, 5, 4) - 4;
                                    unset($this->stmtPrepare[$clientId][$closeStmtId]);
                                    break;
                            }
                            unset($clientId);
                        }
                    }
                });
            }
        });
    }
接收消息.

@param $server
@param $fd
@param $reactor_id
@param $data