1 /* 2 * Copyright (c) 2021 Huawei Device Co., Ltd. 3 * Licensed under the Apache License, Version 2.0 (the "License"); 4 * you may not use this file except in compliance with the License. 5 * You may obtain a copy of the License at 6 * 7 * http://www.apache.org/licenses/LICENSE-2.0 8 * 9 * Unless required by applicable law or agreed to in writing, software 10 * distributed under the License is distributed on an "AS IS" BASIS, 11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12 * See the License for the specific language governing permissions and 13 * limitations under the License. 14 */ 15 16 #ifndef PROTOCOLPROTO_H 17 #define PROTOCOLPROTO_H 18 19 #include <cstdint> 20 #include <memory> 21 #include <shared_mutex> 22 #include "message.h" 23 #include "frame_header.h" 24 #include "parse_result.h" 25 #include "serial_buffer.h" 26 #include "message_transform.h" 27 #include "communicator_type_define.h" 28 #include "iprocess_communicator.h" 29 30 namespace DistributedDB { 31 struct PhyHeaderInfo { 32 uint64_t sourceId; 33 uint32_t frameId; 34 FrameType frameType; 35 }; 36 37 struct FrameFragmentInfo { 38 uint8_t *oringinalBytesAddr; 39 uint32_t extendHeadSize; 40 uint32_t splitLength; 41 uint16_t fragCount; 42 }; 43 44 struct FragmentPacket { 45 uint8_t *ptrPacket; 46 uint32_t leftLength; 47 }; 48 49 class ProtocolProto { 50 public: 51 // For application layer frame 52 static uint32_t GetAppLayerFrameHeaderLength(); 53 static uint32_t GetLengthBeforeSerializedData(); 54 55 // For communication layer frame 56 static uint32_t GetCommLayerFrameHeaderLength(); 57 58 // For handling application layer message. Return a heap object. 59 static SerialBuffer *ToSerialBuffer(const Message *inMsg, int &outErrorNo, 60 std::shared_ptr<ExtendHeaderHandle> &extendHandle, bool onlyMsgHeader = false); 61 static Message *ToMessage(const SerialBuffer *inBuff, int &outErrorNo, bool onlyMsgHeader = false); 62 63 // For handling communication layer frame. Return a heap object. 64 static SerialBuffer *BuildEmptyFrameForVersionNegotiate(int &outErrorNo); 65 static SerialBuffer *BuildFeedbackMessageFrame(const Message *inMsg, const LabelType &inLabel, int &outErrorNo); 66 static SerialBuffer *BuildLabelExchange(uint64_t inDistinctValue, uint64_t inSequenceId, 67 const std::set<LabelType> &inLabels, int &outErrorNo); 68 static SerialBuffer *BuildLabelExchangeAck(uint64_t inDistinctValue, uint64_t inSequenceId, int &outErrorNo); 69 70 // Return E_OK if no error happened. outPieces.size equal zero means not split, in this case, use ori buff. 71 static int SplitFrameIntoPacketsIfNeed(const SerialBuffer *inBuff, uint32_t inMtuSize, 72 std::vector<std::pair<std::vector<uint8_t>, uint32_t>> &outPieces); 73 static int AnalyzeSplitStructure(const ParseResult &inResult, uint32_t &outFragLen, uint32_t &outLastFragLen); 74 75 // inFrame is the destination, pktBytes and pktLength are the source, fragOffset and fragLength give the boundary 76 static int CombinePacketIntoFrame(SerialBuffer *inFrame, const uint8_t *pktBytes, uint32_t pktLength, 77 uint32_t fragOffset, uint32_t fragLength); 78 79 // Must not be called in multi-thread 80 // Return E_ALREADY_REGISTER if msgId is already registered 81 // Return E_INVALID_ARGS if member of inFunc not all valid 82 static int RegTransformFunction(uint32_t msgId, const TransformFunc &inFunc); 83 84 static void UnRegTransformFunction(uint32_t msgId); 85 86 // For application layer frame. In send case. Focus on frame. 87 static int SetDivergeHeader(SerialBuffer *inBuff, const LabelType &inCommLabel); 88 89 // For both application and communication layer frame. In send case. Focus on frame. 90 static int SetPhyHeader(SerialBuffer *inBuff, const PhyHeaderInfo &inInfo); 91 92 // In receive case, return error if parse fail. 93 static int CheckAndParsePacket(const std::string &srcTarget, const uint8_t *bytes, uint32_t length, 94 ParseResult &outResult); 95 96 // The CommPhyHeader had already been parsed into outResult 97 static int CheckAndParseFrame(const SerialBuffer *inBuff, ParseResult &outResult); 98 99 // Dfx method for helping debugging 100 static void DisplayPacketInformation(const uint8_t *bytes, uint32_t length); 101 102 ProtocolProto() = delete; 103 ~ProtocolProto() = delete; 104 private: 105 static int CalculateXorSum(const uint8_t *bytes, uint32_t length, uint64_t &outSum); 106 107 // For handling application layer message 108 static int CalculateDataSerializeLength(const Message *inMsg, uint32_t &outLength); 109 static int SerializeMessage(SerialBuffer *inBuff, const Message *inMsg); 110 static int DeSerializeMessage(const SerialBuffer *inBuff, Message *inMsg, bool onlyMsgHeader); 111 static bool IsSupportMessageVersion(uint16_t version); 112 static bool IsFeedbackErrorMessage(uint32_t errorNo); 113 114 static int ParseCommPhyHeader(const std::string &srcTarget, const uint8_t *bytes, uint32_t length, 115 ParseResult &inResult); 116 static int ParseCommPhyHeaderCheckMagicAndVersion(const uint8_t *bytes, uint32_t length); 117 static int ParseCommPhyHeaderCheckField(const std::string &srcTarget, const CommPhyHeader &phyHeader, 118 const uint8_t *bytes, uint32_t length); 119 static int ParseCommPhyOptHeader(const uint8_t *bytes, uint32_t length, ParseResult &inResult); 120 static int ParseCommDivergeHeader(const uint8_t *bytes, uint32_t length, ParseResult &inResult); 121 static int ParseCommLayerPayload(const uint8_t *bytes, uint32_t length, ParseResult &inResult); 122 static int ParseLabelExchange(const uint8_t *bytes, uint32_t length, ParseResult &inResult); 123 static int ParseLabelExchangeAck(const uint8_t *bytes, uint32_t length, ParseResult &inResult); 124 125 static int FrameFragmentation(const uint8_t *splitStartBytes, const FrameFragmentInfo &fragmentInfo, 126 const CommPhyHeader &framePhyHeader, std::vector<std::pair<std::vector<uint8_t>, uint32_t>> &outPieces); 127 static int FillFragmentPacket(const CommPhyHeader &phyHeader, const CommPhyOptHeader &phyOptHeader, 128 const uint8_t *fragBytes, uint32_t fragLen, FragmentPacket &outPacket); 129 static int FillFragmentPacketExtendHead(uint8_t *headBytesAddr, uint32_t headLen, FragmentPacket &outPacket); 130 static int GetExtendHeadDataSize(std::shared_ptr<ExtendHeaderHandle> &extendHandle, uint32_t &headSize); 131 static int FillExtendHeadDataIfNeed(std::shared_ptr<ExtendHeaderHandle> &extendHandle, SerialBuffer *buffer, 132 uint32_t headSize); 133 134 static int GetTransformFunc(uint32_t messageId, TransformFunc &function); 135 136 static std::shared_mutex msgIdMutex_; 137 static std::map<uint32_t, TransformFunc> msgIdMapFunc_; 138 }; 139 } // namespace DistributedDB 140 141 #endif // PROTOCOLPROTO_H 142