src/MysqlProxy.php
public function onClientReceive(\Swoole\Coroutine\Client $cli, string $data)
    {
        if ($this->connected) {
            $packages = [$data];
        } else {
            $packages = packageSplit($data, true, 3, true);
        }
        foreach ($packages as $package) {
            $data = $package;
            self::go(function () use ($cli, $data) {
                $fd = $this->serverFd;
                $binaryPacket = new BinaryPacket();
                $binaryPacket->data = getBytes($data);
                $binaryPacket->packetLength = $binaryPacket->calcPacketSize();
                if (isset($binaryPacket->data[4])) {
                    $send = true;
                    if ($binaryPacket->data[4] == ErrorPacket::$FIELD_COUNT) {
                        $errorPacket = new ErrorPacket();
                        $errorPacket->read($binaryPacket);
                        $errorPacket->errno = ErrorCode::ER_SYNTAX_ERROR;
                        $data = getString($errorPacket->write());
                    } elseif (!$this->connected) {
                        if ($binaryPacket->data[4] == OkPacket::$FIELD_COUNT) {
                            if (!array_diff_assoc($binaryPacket->data, OkPacket::$AUTH_OK) ||
                                !array_diff_assoc($binaryPacket->data, OkPacket::$FAST_AUTH_OK) ||
                                !array_diff_assoc($binaryPacket->data, OkPacket::$SWITCH_AUTH_OK) ||
                                !array_diff_assoc($binaryPacket->data, OkPacket::$FULL_AUTH_OK)) {
                                $send = false;
                                $this->connected = true;
                                $this->chan->push($this);
                            }
                            # 快速认证
                        } elseif ($binaryPacket->data[4] == 0x01) {
                            # 请求公钥
                            if ($binaryPacket->packetLength == 6) {
                                if ($binaryPacket->data[$binaryPacket->packetLength - 1] == 4) {
                                    $data = getString(array_merge(getMysqlPackSize(1), [3, 2]));
                                    $this->send($data);
                                }
                            } else {
                                $this->serverPublicKey = substr($data, 5, strlen($data) - 2);
                                $encryptData = SecurityUtil::sha2RsaEncrypt($this->account['password'], $this->salt, $this->serverPublicKey);
                                $data = getString(array_merge(getMysqlPackSize(strlen($encryptData)), [5])) . $encryptData;
                                $this->send($data);
                            }
                            $send = false;
                        } elseif ($binaryPacket->data[4] == 0xfe) {
                            $mm = new MySQLMessage($binaryPacket->data);
                            $mm->move(5);
                            $pluginName = $mm->readStringWithNull();
                            $this->salt = $mm->readBytesWithNull();
                            $password = $this->processAuth($pluginName ?: 'mysql_native_password');
                            $this->send(getString(array_merge(getMysqlPackSize(count($password)), [3], $password)));
                            $send = false;
                        } elseif (!$this->auth) {
                            $handshakePacket = (new HandshakePacket())->read($binaryPacket);
                            $this->salt = array_merge($handshakePacket->seed, $handshakePacket->restOfScrambleBuff);
                            $password = $this->processAuth($handshakePacket->pluginName);
                            $clientFlag = Capabilities::CLIENT_CAPABILITIES;
                            $authPacket = new AuthPacket();
                            $authPacket->pluginName = $handshakePacket->pluginName;
                            $authPacket->packetId = 1;
                            if (isset($this->database) && $this->database) {
                                $authPacket->database = $this->database;
                            } else {
                                $authPacket->database = 0;
                            }
                            if ($authPacket->database) {
                                $clientFlag |= Capabilities::CLIENT_CONNECT_WITH_DB;
                            }
                            if (version_compare($handshakePacket->serverVersion, '5.0', '>=')) {
                                $clientFlag |= Capabilities::CLIENT_MULTI_RESULTS;
                            }
                            $authPacket->clientFlags = $clientFlag;
                            $authPacket->serverCapabilities = $handshakePacket->serverCapabilities;
                            $authPacket->maxPacketSize =
                                CONFIG['server']['swoole_client_setting']['package_max_length'] ?? 16777215;
                            $authPacket->charsetIndex = CharsetUtil::getIndex($this->charset ?? 'utf8mb4');
                            $authPacket->user = $this->account['user'];
                            $authPacket->password = $password;
                            $this->auth = true;
                            $this->send(getString($authPacket->write()));
                            $send = false;
                        }
                    }
                    if ($send && $this->server->exist($fd)) {
                        $this->server->send($fd, $data);
                    }
                }
            });
        }
    }
public function onClientReceive(\Swoole\Coroutine\Client $cli, string $data)
    {
        if ($this->connected) {
            $packages = [$data];
        } else {
            $packages = packageSplit($data, true, 3, true);
        }
        foreach ($packages as $package) {
            $data = $package;
            self::go(function () use ($cli, $data) {
                $fd = $this->serverFd;
                $binaryPacket = new BinaryPacket();
                $binaryPacket->data = getBytes($data);
                $binaryPacket->packetLength = $binaryPacket->calcPacketSize();
                if (isset($binaryPacket->data[4])) {
                    $send = true;
                    if ($binaryPacket->data[4] == ErrorPacket::$FIELD_COUNT) {
                        $errorPacket = new ErrorPacket();
                        $errorPacket->read($binaryPacket);
                        $errorPacket->errno = ErrorCode::ER_SYNTAX_ERROR;
                        $data = getString($errorPacket->write());
                    } elseif (!$this->connected) {
                        if ($binaryPacket->data[4] == OkPacket::$FIELD_COUNT) {
                            if (!array_diff_assoc($binaryPacket->data, OkPacket::$AUTH_OK) ||
                                !array_diff_assoc($binaryPacket->data, OkPacket::$FAST_AUTH_OK) ||
                                !array_diff_assoc($binaryPacket->data, OkPacket::$SWITCH_AUTH_OK) ||
                                !array_diff_assoc($binaryPacket->data, OkPacket::$FULL_AUTH_OK)) {
                                $send = false;
                                $this->connected = true;
                                $this->chan->push($this);
                            }
                            # 快速认证
                        } elseif ($binaryPacket->data[4] == 0x01) {
                            # 请求公钥
                            if ($binaryPacket->packetLength == 6) {
                                if ($binaryPacket->data[$binaryPacket->packetLength - 1] == 4) {
                                    $data = getString(array_merge(getMysqlPackSize(1), [3, 2]));
                                    $this->send($data);
                                }
                            } else {
                                $this->serverPublicKey = substr($data, 5, strlen($data) - 2);
                                $encryptData = SecurityUtil::sha2RsaEncrypt($this->account['password'], $this->salt, $this->serverPublicKey);
                                $data = getString(array_merge(getMysqlPackSize(strlen($encryptData)), [5])) . $encryptData;
                                $this->send($data);
                            }
                            $send = false;
                        } elseif ($binaryPacket->data[4] == 0xfe) {
                            $mm = new MySQLMessage($binaryPacket->data);
                            $mm->move(5);
                            $pluginName = $mm->readStringWithNull();
                            $this->salt = $mm->readBytesWithNull();
                            $password = $this->processAuth($pluginName ?: 'mysql_native_password');
                            $this->send(getString(array_merge(getMysqlPackSize(count($password)), [3], $password)));
                            $send = false;
                        } elseif (!$this->auth) {
                            $handshakePacket = (new HandshakePacket())->read($binaryPacket);
                            $this->salt = array_merge($handshakePacket->seed, $handshakePacket->restOfScrambleBuff);
                            $password = $this->processAuth($handshakePacket->pluginName);
                            $clientFlag = Capabilities::CLIENT_CAPABILITIES;
                            $authPacket = new AuthPacket();
                            $authPacket->pluginName = $handshakePacket->pluginName;
                            $authPacket->packetId = 1;
                            if (isset($this->database) && $this->database) {
                                $authPacket->database = $this->database;
                            } else {
                                $authPacket->database = 0;
                            }
                            if ($authPacket->database) {
                                $clientFlag |= Capabilities::CLIENT_CONNECT_WITH_DB;
                            }
                            if (version_compare($handshakePacket->serverVersion, '5.0', '>=')) {
                                $clientFlag |= Capabilities::CLIENT_MULTI_RESULTS;
                            }
                            $authPacket->clientFlags = $clientFlag;
                            $authPacket->serverCapabilities = $handshakePacket->serverCapabilities;
                            $authPacket->maxPacketSize =
                                CONFIG['server']['swoole_client_setting']['package_max_length'] ?? 16777215;
                            $authPacket->charsetIndex = CharsetUtil::getIndex($this->charset ?? 'utf8mb4');
                            $authPacket->user = $this->account['user'];
                            $authPacket->password = $password;
                            $this->auth = true;
                            $this->send(getString($authPacket->write()));
                            $send = false;
                        }
                    }
                    if ($send && $this->server->exist($fd)) {
                        $this->server->send($fd, $data);
                    }
                }
            });
        }
    }
mysql 客户端消息转发.

@param $cli
@param $data