| <?php |
| namespace com\fenqile\fsof\common\protocol\fsof; |
| |
| use com\fenqile\fsof\consumer\Type; |
| use Icecave\Flax\Serialization\Encoder; |
| use Icecave\Flax\DubboParser as Decoder; |
| use com\fenqile\fsof\consumer\ConsumerException; |
| |
| /** |
| * |
| * Dubbo网络协议 |
| * +------------------------------------------------------------------------------------------+ |
| * | 包头（二进制数据 16bit） | 包体 | |
| * +-----------------------------------------------------------------------------------------+ |
| * | 版本号 | 命令&serialize | 空白 | 包序号| 长度 | 数据 | |
| * +-----------------------------------------------------------------------------------------+ |
| * | magic(2) | cmd&serialize(1)|(1) |sn(8) | len(4) | data(N)| |
| * +-----------------------------------------------------------------------------------------+ |
| * |
| * magic：协议包起始标识, 0xdabb |
| * -------------------------------------------------------------------------------------------- |
| * cmd：命令类型：FLAG_REQUEST为0x80, FLAG_TWOWAY为0x40, FLAG_EVENT为0x20 |
| * serialize：序列化方案编号：与cmd共用一个字节,采用json,对应dubbo中编号为6 |
| * -------------------------------------------------------------------------------------------- |
| * sn：请求序号，consumer会为每个请求编制一个进程内唯一序号 |
| * ，provider处理完请求后在返回的数据包中会携带该sn号，供consumer判断当前的数据是对应哪个请求 |
| * --------------------------------------------------------------------------------------------* |
| * len：数据报文长度 |
| * -------------------------------------------------------------------------------------------- |
| * data：数据报文，目前采用json进行序列化 |
| * -------------------------------------------------------------------------------------------- |
| */ |
| class DubboParser |
| { |
| //dubbo协议基本信息 |
| const PACKAGE_HEDA_LEN = 16; |
| const MAX_RECV_LEN = 1048576;//1024*1024; |
| const RESPONSE_TCP_SEGMENT_LEN = 1048576;//1*1024*1024; |
| |
| //fsof协议ver字段，ver字段既指示协议版本信息，也作为magic使用 |
| const DUBBO_PROTOCOL_MAGIC = 0xdabb; |
| |
| //serialize 方案编号 |
| const DUBBO_PROTOCOL_SERIALIZE_FAST_JSON = 6; //fastjson serialization code |
| |
| const DUBBO_PROTOCOL_SERIALIZE_HESSIAN2 = 2; //hessian2 serialization code |
| |
| const DUBBO_PROTOCOL_NAME_MAP_CODE = [ |
| 'hessian2' => self::DUBBO_PROTOCOL_SERIALIZE_HESSIAN2, |
| 'fastjson' => self::DUBBO_PROTOCOL_SERIALIZE_FAST_JSON |
| ]; |
| |
| |
| //fsof协议包头cmd字段含义 |
| const FLAG_REQUEST = 0x80; //request |
| const FLAG_TWOWAY = 0x40; //two_way |
| const FLAG_HEARTBEAT_EVENT = 0x20; //heart_event |
| const SERIALIZATION_MASK = 0x1f; //serialization_mask |
| |
| const UPPER_MASK = 0xffffffff00000000; |
| const LOWER_MASK = 0x00000000ffffffff; |
| |
| const RESPONSE_WITH_EXCEPTION = 0; |
| const RESPONSE_VALUE = 1; |
| const RESPONSE_NULL_VALUE = 2; |
| |
| private $logger; |
| |
| public function __construct() |
| { |
| $this->logger = \Logger::getLogger(__CLASS__); |
| } |
| |
| public function packRequest(DubboRequest $request) |
| { |
| if (self::DUBBO_PROTOCOL_SERIALIZE_HESSIAN2 == (self::DUBBO_PROTOCOL_NAME_MAP_CODE[$request->getSerialization()]??null)) { |
| $reqData = $this->buildBodyForHessian2($request); |
| $serialize_type = self::DUBBO_PROTOCOL_SERIALIZE_HESSIAN2; |
| } else { |
| $reqData = $this->buildBodyForFastJson($request); |
| $serialize_type = self::DUBBO_PROTOCOL_SERIALIZE_FAST_JSON; |
| } |
| $upper = ($request->getSn() & self::UPPER_MASK) >> 32; |
| $lower = $request->getSn() & self::LOWER_MASK; |
| $flag = (self::FLAG_REQUEST | $serialize_type); |
| if ($request->isTwoWay()) $flag |= self::FLAG_TWOWAY; |
| if ($request->isHeartbeatEvent()) $flag |= self::FLAG_HEARTBEAT_EVENT; |
| $out = pack("n1C1a1N1N1N1", |
| self::DUBBO_PROTOCOL_MAGIC, |
| $flag, |
| "", |
| $upper, |
| $lower, |
| strlen($reqData)); |
| return $out . $reqData; |
| } |
| |
| public function buildBodyForFastJson(DubboRequest $request) |
| { |
| $reqData = json_encode($request->getDubboVersion()) . PHP_EOL . |
| json_encode($request->getService()) . PHP_EOL; |
| if ($request->getVersion()) { |
| $reqData .= json_encode($request->getVersion()) . PHP_EOL; |
| } else { |
| $reqData .= '""' . PHP_EOL; |
| } |
| $reqData .= json_encode($request->getMethod()) . PHP_EOL; |
| $reqData .= json_encode($this->typeRefs($request)) . PHP_EOL; |
| foreach ($request->getParams() as $value) { |
| $reqData .= json_encode($value) . PHP_EOL; |
| } |
| $attach = array(); |
| $attach['path'] = $request->getService(); |
| $attach['interface'] = $request->getService(); |
| if ($request->getGroup()) { |
| $attach['group'] = $request->getGroup(); |
| } |
| if ($request->getVersion()) { |
| $attach['version'] = $request->getVersion(); |
| } |
| $attach['timeout'] = $request->getTimeout(); |
| $request->setAttach($attach); |
| $reqData .= json_encode($request->getAttach()); |
| return $reqData; |
| |
| } |
| |
| public function buildBodyForHessian2(DubboRequest $request) |
| { |
| $encode = new Encoder(); |
| $reqData = ''; |
| $reqData .= $encode->encode($request->getDubboVersion()); |
| $reqData .= $encode->encode($request->getService()); |
| if ($request->getVersion()) { |
| $reqData .= $encode->encode($request->getVersion()); |
| } else { |
| $reqData .= $encode->encode(''); |
| } |
| $reqData .= $encode->encode($request->getMethod()); |
| $reqData .= $encode->encode($this->typeRefs($request)); |
| foreach ($request->getParams() as $value) { |
| $reqData .= $encode->encode($value); |
| } |
| $attach = ['path' => $request->getService(), 'interface' => $request->getService(), 'timeout' => $request->getTimeout()]; |
| if ($request->getGroup()) { |
| $attach['group'] = $request->getGroup(); |
| } |
| if ($request->getVersion()) { |
| $attach['version'] = $request->getVersion(); |
| } |
| $reqData .= $encode->encode($attach); |
| return $reqData; |
| } |
| |
| private function typeRefs(DubboRequest $request) |
| { |
| $typeRefs = ''; |
| foreach ($request->getTypes() as $type) { |
| $typeRefs .= $type; |
| } |
| return $typeRefs; |
| } |
| |
| |
| public function parseResponseHeader(DubboResponse $response) |
| { |
| $res_header = substr($response->getFullData(), 0, self::PACKAGE_HEDA_LEN); |
| $format = 'n1magic/C1flag/C1status/N1upper/N1lower/N1len'; |
| $_arr = unpack($format, $res_header); |
| $response->setStatus($_arr['status']); |
| $response->setSn($_arr["upper"] << 32 | $_arr["lower"]); |
| $flag = $_arr["flag"]; |
| if (($flag & self::FLAG_HEARTBEAT_EVENT) != 0) { |
| $response->setHeartbeatEvent(true); |
| } |
| $response->setSerialization($flag & self::SERIALIZATION_MASK); |
| $response->setLen($_arr["len"]); |
| return $response; |
| } |
| |
| public function parseResponseBody(DubboResponse $response) |
| { |
| if (DubboResponse::OK == $response->getStatus()) { |
| if (self::DUBBO_PROTOCOL_SERIALIZE_FAST_JSON == $response->getSerialization()) { |
| $this->parseResponseBodyForFastJson($response); |
| } else if (self::DUBBO_PROTOCOL_SERIALIZE_HESSIAN2 == $response->getSerialization()) { |
| $this->parseResponseBodyForHessian2($response); |
| } else { |
| throw new ConsumerException(sprintf('返回的序列化类型:(%s), 不支持解析!', $response->getSerialization())); |
| } |
| } else { |
| throw new ConsumerException($response->getFullData()); |
| } |
| return $response; |
| } |
| |
| private function parseResponseBodyForFastJson(DubboResponse $response) |
| { |
| $_data = substr($response->getFullData(), self::PACKAGE_HEDA_LEN); |
| $response->setResponseBody($_data); |
| list($status, $content) = explode(PHP_EOL, $_data); |
| if ($response->isHeartbeatEvent()) { |
| $response->setResult(json_decode($status, true)); |
| } else { |
| switch ($status) { |
| case self::RESPONSE_NULL_VALUE: |
| break; |
| case self::RESPONSE_VALUE: |
| $response->setResult(json_decode($content, true)); |
| break; |
| case self::RESPONSE_WITH_EXCEPTION: |
| $exception = json_decode($content, true); |
| if (is_array($exception) && array_key_exists('message', $exception)) { |
| throw new ConsumerException($exception['message']); |
| } else if (is_string($exception)) { |
| throw new ConsumerException($exception); |
| } else { |
| throw new ConsumerException("provider occur error"); |
| } |
| break; |
| default: |
| return false; |
| } |
| } |
| return $response; |
| } |
| |
| private function parseResponseBodyForHessian2(DubboResponse $response) |
| { |
| if (!$response->isHeartbeatEvent()) { |
| $_data = $response->getFullData(); |
| $decode = new Decoder($_data); |
| $content = $decode->getData($_data); |
| $response->setResult($content); |
| } |
| return $response; |
| } |
| |
| |
| public function parseRequestHeader(DubboRequest &$request) |
| { |
| $_data = substr($request->getFullData(), 0, self::PACKAGE_HEDA_LEN); |
| $format = 'n1magic/C1flag/C1blank/N1upper/N1lower/N1len'; |
| $_arr = unpack($format, $_data); |
| $flag = $_arr['flag']; |
| $request->setTwoWay(($flag & self::FLAG_TWOWAY) != 0); |
| if (($flag & self::FLAG_HEARTBEAT_EVENT) != 0) { |
| $request->setHeartbeatEvent(true); |
| } |
| $request->setSerialization($flag & self::DUBBO_PROTOCOL_SERIALIZE_FAST_JSON); |
| $request->setSn($_arr['upper'] << 32 | $_arr['lower']); |
| $request->setDataLen($_arr['len']); |
| $request->setRequestLen($request->getDataLen() + self::PACKAGE_HEDA_LEN); |
| return $request; |
| } |
| |
| public function parseRequestBody(DubboRequest &$request) |
| { |
| if ($request->getSerialization() != self::DUBBO_PROTOCOL_SERIALIZE_FAST_JSON) { |
| $this->logger->error("unknown serialization type :" . $request->getSerialization()); |
| return false; |
| } |
| $cliData = substr($request->getFullData(), self::PACKAGE_HEDA_LEN); |
| if ($cliData) { |
| if ($request->isHeartbeatEvent()) { |
| //心跳请求,不需要数据回送 |
| } else { |
| $dataArr = explode(PHP_EOL, $cliData); |
| $request->setDubboVersion(json_decode($dataArr[0], true)); |
| $request->setService(json_decode($dataArr[1], true)); |
| $request->setVersion(json_decode($dataArr[2], true)); |
| $methodName = json_decode($dataArr[3], true); |
| if ($methodName == "\$invoke") { |
| //泛化调用 |
| $request->setMethod(json_decode($dataArr[5], true)); |
| $request->setParams(json_decode($dataArr[7], true)); |
| $attach = json_decode($dataArr[8], true); |
| } else { |
| //非泛化调用 |
| $request->setMethod($methodName); |
| $paramTypes = json_decode($dataArr[4], true); |
| if ($paramTypes == "") { |
| //调用没有参数的方法 |
| $request->setTypes(NULL); |
| $request->setParams(NULL); |
| $attach = json_decode($dataArr[5], true); |
| } else { |
| $typeArr = explode(";", $paramTypes); |
| $typeArrLen = count($typeArr); |
| $request->setParamNum($typeArrLen - 1); |
| $params = array(); |
| for ($i = 0; $i < $typeArrLen - 1; $i++) { |
| $params[$i] = json_decode($dataArr[5 + $i], true); |
| } |
| $request->setParams($params); |
| $attach = json_decode($dataArr[5 + ($typeArrLen - 1)], true); |
| } |
| } |
| $request->setAttach($attach); |
| if (array_key_exists('group', $attach)) { |
| $request->setGroup($attach['group']); |
| } |
| return $request; |
| } |
| } |
| return false; |
| } |
| |
| |
| public function packResponse(DubboResponse &$response) |
| { |
| if ($response->getStatus() != DubboResponse::OK) { |
| $resData = json_encode($response->getErrorMsg()); |
| } else { |
| if ($response->getErrorMsg() != NULL && $response->getErrorMsg() != "") { |
| $resData = json_encode(self::RESPONSE_WITH_EXCEPTION) . PHP_EOL . json_encode($response->getErrorMsg()); |
| } else if ($response->getResult() == NULL) { |
| $resData = json_encode(self::RESPONSE_NULL_VALUE); |
| } else { |
| $resData = json_encode(self::RESPONSE_VALUE) . PHP_EOL . json_encode($response->getResult()); |
| } |
| } |
| $resData = $resData . PHP_EOL; |
| $upper = ($response->getSn() & self::UPPER_MASK) >> 32; |
| $lower = $response->getSn() & self::LOWER_MASK; |
| $flag = self::DUBBO_PROTOCOL_SERIALIZE_FAST_JSON; |
| if ($response->isHeartbeatEvent()) { |
| $flag |= self::FLAG_HEARTBEAT_EVENT; |
| } |
| $out = pack("n1C1C1N1N1N1", |
| self::DUBBO_PROTOCOL_MAGIC, |
| $flag, |
| $response->getStatus(), |
| $upper, |
| $lower, |
| strlen($resData)); |
| |
| return $out . $resData; |
| } |
| |
| public function isNormalResponse(DubboResponse $response) |
| { |
| return !($response->isHeartbeatEvent()); |
| } |
| |
| public function isNormalRequest(DubboRequest $request) |
| { |
| return !($request->isHeartbeatEvent()); |
| } |
| |
| public function isOneWayRequest(DubboRequest $request) |
| { |
| return !($request->isTwoWay()); |
| } |
| |
| public function isHearBeatRequest(DubboRequest $request) |
| { |
| return $request->isHeartbeatEvent(); |
| } |
| |
| public function isHearBeatResponse(DubboResponse $response) |
| { |
| return $response->isHeartbeatEvent(); |
| } |
| |
| public static function getReqInQueueTime(DubboRequest $request) |
| { |
| $ret = 0; |
| if (!empty($request->reqInfo)) { |
| $ret = isset($request->reqInfo['inqueue_time']) ? $request->reqInfo['inqueue_time'] : 0; |
| } |
| return $ret; |
| } |
| } |
