src/message/MessageDecoder.cpp - rocketmq-client-cpp - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 #include "MessageDecoder.h"

 #include <algorithm>  // std::move
 #include <exception>  // std::exception
 #include <sstream>    // std::stringstream

 #ifndef WIN32
 #include <arpa/inet.h>   // htons
 #include <netinet/in.h>  // sockaddr_in, sockaddr_in6
 #else
 #include "Winsock2.h"
 #endif

 #include "ByteOrder.h"
 #include "Logging.h"
 #include "MessageAccessor.hpp"
 #include "MessageExtImpl.h"
 #include "MessageSysFlag.h"
 #include "SocketUtil.h"
 #include "UtilAll.h"

 static const char NAME_VALUE_SEPARATOR = 1;
 static const char PROPERTY_SEPARATOR = 2;

 namespace rocketmq {

 std::string MessageDecoder::createMessageId(const struct sockaddr* sa, int64_t offset) {
   int msgIdLength = IpaddrSize(sa) + /* port field size */ 4 + sizeof(offset);
   std::unique_ptr<ByteBuffer> byteBuffer(ByteBuffer::allocate(msgIdLength));
   if (sa->sa_family == AF_INET) {
     struct sockaddr_in* sin = (struct sockaddr_in*)sa;
     byteBuffer->put(ByteArray(reinterpret_cast<char*>(&sin->sin_addr), kIPv4AddrSize));
     byteBuffer->putInt(ntohs(sin->sin_port));
   } else {
     struct sockaddr_in6* sin6 = (struct sockaddr_in6*)sa;
     byteBuffer->put(ByteArray(reinterpret_cast<char*>(&sin6->sin6_addr), kIPv6AddrSize));
     byteBuffer->putInt(ntohs(sin6->sin6_port));
   }
   byteBuffer->putLong(offset);
   byteBuffer->flip();
   return UtilAll::bytes2string(byteBuffer->array(), msgIdLength);
 }

 MessageId MessageDecoder::decodeMessageId(const std::string& msgId) {
   size_t ip_length = msgId.length() == 32 ? kIPv4AddrSize * 2 : kIPv6AddrSize * 2;

   ByteArray byteArray(ip_length / 2);
   std::string ip = msgId.substr(0, ip_length);
   UtilAll::string2bytes(byteArray.array(), ip);

   std::string port = msgId.substr(ip_length, 8);
   uint32_t portInt = std::stoul(port, nullptr, 16);

   auto* sin = IPPortToSockaddr(byteArray, portInt);

   std::string offset = msgId.substr(ip_length + 8);
   uint64_t offsetInt = std::stoull(offset, nullptr, 16);

   return MessageId(sin, offsetInt);
 }

 MessageExtPtr MessageDecoder::clientDecode(ByteBuffer& byteBuffer, bool readBody) {
   return decode(byteBuffer, readBody, true, true);
 }

 MessageExtPtr MessageDecoder::decode(ByteBuffer& byteBuffer) {
   return decode(byteBuffer, true);
 }

 MessageExtPtr MessageDecoder::decode(ByteBuffer& byteBuffer, bool readBody) {
   return decode(byteBuffer, readBody, true, false);
 }

 MessageExtPtr MessageDecoder::decode(ByteBuffer& byteBuffer, bool readBody, bool deCompressBody, bool isClient) {
   try {
     auto msgExt = isClient ? std::make_shared<MessageClientExtImpl>() : std::make_shared<MessageExtImpl>();

     // 1 TOTALSIZE
     int32_t storeSize = byteBuffer.getInt();
     msgExt->set_store_size(storeSize);

     // 2 MAGICCODE sizeof(int)
     byteBuffer.getInt();

     // 3 BODYCRC
     int32_t bodyCRC = byteBuffer.getInt();
     msgExt->set_body_crc(bodyCRC);

     // 4 QUEUEID
     int32_t queueId = byteBuffer.getInt();
     msgExt->set_queue_id(queueId);

     // 5 FLAG
     int32_t flag = byteBuffer.getInt();
     msgExt->set_flag(flag);

     // 6 QUEUEOFFSET
     int64_t queueOffset = byteBuffer.getLong();
     msgExt->set_queue_offset(queueOffset);

     // 7 PHYSICALOFFSET
     int64_t physicOffset = byteBuffer.getLong();
     msgExt->set_commit_log_offset(physicOffset);

     // 8 SYSFLAG
     int32_t sysFlag = byteBuffer.getInt();
     msgExt->set_sys_flag(sysFlag);

     // 9 BORNTIMESTAMP
     int64_t bornTimeStamp = byteBuffer.getLong();
     msgExt->set_born_timestamp(bornTimeStamp);

     // 10 BORNHOST
     int bornHostLength = (sysFlag & MessageSysFlag::BORNHOST_V6_FLAG) == 0 ? kIPv4AddrSize : kIPv6AddrSize;
     ByteArray bornHost(bornHostLength);
     byteBuffer.get(bornHost, 0, bornHostLength);
     int32_t bornPort = byteBuffer.getInt();
     msgExt->set_born_host(IPPortToSockaddr(bornHost, bornPort));

     // 11 STORETIMESTAMP
     int64_t storeTimestamp = byteBuffer.getLong();
     msgExt->set_store_timestamp(storeTimestamp);

     // 12 STOREHOST
     int storeHostLength = (sysFlag & MessageSysFlag::STOREHOST_V6_FLAG) == 0 ? kIPv4AddrSize : kIPv6AddrSize;
     ByteArray storeHost(storeHostLength);
     byteBuffer.get(storeHost, 0, storeHostLength);
     int32_t storePort = byteBuffer.getInt();
     msgExt->set_store_host(IPPortToSockaddr(storeHost, storePort));

     // 13 RECONSUMETIMES
     int32_t reconsumeTimes = byteBuffer.getInt();
     msgExt->set_reconsume_times(reconsumeTimes);

     // 14 Prepared Transaction Offset
     int64_t preparedTransactionOffset = byteBuffer.getLong();
     msgExt->set_prepared_transaction_offset(preparedTransactionOffset);

     // 15 BODY
     int uncompress_failed = false;
     int32_t bodyLen = byteBuffer.getInt();
     if (bodyLen > 0) {
       if (readBody) {
         ByteArray body(byteBuffer.array() + byteBuffer.arrayOffset() + byteBuffer.position(), bodyLen);
         byteBuffer.position(byteBuffer.position() + bodyLen);

         // decompress body
         if (deCompressBody && (sysFlag & MessageSysFlag::COMPRESSED_FLAG) == MessageSysFlag::COMPRESSED_FLAG) {
           std::string origin_body;
           if (UtilAll::inflate(body, origin_body)) {
             msgExt->set_body(std::move(origin_body));
           } else {
             uncompress_failed = true;
           }
         } else {
           msgExt->set_body(std::string(body.array(), body.size()));
         }
       } else {
         // skip body
         byteBuffer.position(byteBuffer.position() + bodyLen);
       }
     }

     // 16 TOPIC
     int8_t topicLen = byteBuffer.get();
     ByteArray topic(topicLen);
     byteBuffer.get(topic);
     msgExt->set_topic(topic.array(), topic.size());

     // 17 properties
     int16_t propertiesLen = byteBuffer.getShort();
     if (propertiesLen > 0) {
       ByteArray properties(propertiesLen);
       byteBuffer.get(properties);
       std::string propertiesString(properties.array(), properties.size());
       std::map<std::string, std::string> propertiesMap = string2messageProperties(propertiesString);
       MessageAccessor::setProperties(*msgExt, std::move(propertiesMap));
     }

     // 18 msg ID
     std::string msgId = createMessageId(msgExt->store_host(), (int64_t)msgExt->commit_log_offset());
     msgExt->MessageExtImpl::set_msg_id(msgId);

     if (uncompress_failed) {
       LOG_WARN_NEW("can not uncompress message, id:{}", msgExt->msg_id());
     }

     return msgExt;
   } catch (const std::exception& e) {
     byteBuffer.position(byteBuffer.limit());
     return nullptr;
   }
 }

 std::vector<MessageExtPtr> MessageDecoder::decodes(ByteBuffer& byteBuffer) {
   return decodes(byteBuffer, true);
 }

 std::vector<MessageExtPtr> MessageDecoder::decodes(ByteBuffer& byteBuffer, bool readBody) {
   std::vector<MessageExtPtr> msgExts;
   while (byteBuffer.hasRemaining()) {
     auto msgExt = clientDecode(byteBuffer, readBody);
     if (nullptr == msgExt) {
       break;
     }
     msgExts.emplace_back(std::move(msgExt));
   }
   return msgExts;
 }

 std::string MessageDecoder::messageProperties2String(const std::map<std::string, std::string>& properties) {
   std::string os;

   for (const auto& it : properties) {
     const auto& name = it.first;
     const auto& value = it.second;

     // skip compressed flag
     if (MQMessageConst::PROPERTY_ALREADY_COMPRESSED_FLAG == name) {
       continue;
     }

     os.append(name);
     os += NAME_VALUE_SEPARATOR;
     os.append(value);
     os += PROPERTY_SEPARATOR;
   }

   return os;
 }

 std::map<std::string, std::string> MessageDecoder::string2messageProperties(const std::string& properties) {
   std::vector<std::string> out;
   UtilAll::Split(out, properties, PROPERTY_SEPARATOR);

   std::map<std::string, std::string> map;
   for (size_t i = 0; i < out.size(); i++) {
     std::vector<std::string> outValue;
     UtilAll::Split(outValue, out[i], NAME_VALUE_SEPARATOR);

     if (outValue.size() == 2) {
       map[outValue[0]] = outValue[1];
     }
   }
   return map;
 }

 std::string MessageDecoder::encodeMessage(Message& message) {
   const auto& body = message.body();
   uint32_t bodyLen = body.size();
   std::string properties = MessageDecoder::messageProperties2String(message.properties());
   uint16_t propertiesLength = (int16_t)properties.size();
   uint32_t storeSize = 4                        // 1 TOTALSIZE
                        + 4                      // 2 MAGICCODE
                        + 4                      // 3 BODYCRC
                        + 4                      // 4 FLAG
                        + 4 + bodyLen            // 5 BODY
                        + 2 + propertiesLength;  // 6 PROPERTIES

   // TOTALSIZE|MAGICCODE|BODYCRC|FLAG|BodyLen|Body|propertiesLength|properties
   std::string encodeMsg;

   // 1 TOTALSIZE
   uint32_t storeSize_net = ByteOrderUtil::NorminalBigEndian(storeSize);
   encodeMsg.append((char*)&storeSize_net, sizeof(uint32_t));

   // 2 MAGICCODE
   uint32_t magicCode = 0;
   uint32_t magicCode_net = ByteOrderUtil::NorminalBigEndian(magicCode);
   encodeMsg.append((char*)&magicCode_net, sizeof(uint32_t));

   // 3 BODYCRC
   uint32_t bodyCrc = 0;
   uint32_t bodyCrc_net = ByteOrderUtil::NorminalBigEndian(bodyCrc);
   encodeMsg.append((char*)&bodyCrc_net, sizeof(uint32_t));

   // 4 FLAG
   uint32_t flag = message.flag();
   uint32_t flag_net = ByteOrderUtil::NorminalBigEndian(flag);
   encodeMsg.append((char*)&flag_net, sizeof(uint32_t));

   // 5 BODY
   uint32_t bodyLen_net = ByteOrderUtil::NorminalBigEndian(bodyLen);
   encodeMsg.append((char*)&bodyLen_net, sizeof(uint32_t));
   encodeMsg.append(body.data(), body.size());

   // 6 properties
   uint16_t propertiesLength_net = ByteOrderUtil::NorminalBigEndian(propertiesLength);
   encodeMsg.append((char*)&propertiesLength_net, sizeof(uint16_t));
   encodeMsg.append(std::move(properties));

   return encodeMsg;
 }

 std::string MessageDecoder::encodeMessages(std::vector<MQMessage>& msgs) {
   std::string encodedBody;
   for (auto msg : msgs) {
     encodedBody.append(encodeMessage(msg));
   }
   return encodedBody;
 }

 }  // namespace rocketmq
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	#include "MessageDecoder.h"

	#include <algorithm> // std::move
	#include <exception> // std::exception
	#include <sstream> // std::stringstream

	#ifndef WIN32
	#include <arpa/inet.h> // htons
	#include <netinet/in.h> // sockaddr_in, sockaddr_in6
	#else
	#include "Winsock2.h"
	#endif

	#include "ByteOrder.h"
	#include "Logging.h"
	#include "MessageAccessor.hpp"
	#include "MessageExtImpl.h"
	#include "MessageSysFlag.h"
	#include "SocketUtil.h"
	#include "UtilAll.h"

	static const char NAME_VALUE_SEPARATOR = 1;
	static const char PROPERTY_SEPARATOR = 2;

	namespace rocketmq {

	std::string MessageDecoder::createMessageId(const struct sockaddr* sa, int64_t offset) {
	int msgIdLength = IpaddrSize(sa) + /* port field size */ 4 + sizeof(offset);
	std::unique_ptr<ByteBuffer> byteBuffer(ByteBuffer::allocate(msgIdLength));
	if (sa->sa_family == AF_INET) {
	struct sockaddr_in* sin = (struct sockaddr_in*)sa;
	byteBuffer->put(ByteArray(reinterpret_cast<char*>(&sin->sin_addr), kIPv4AddrSize));
	byteBuffer->putInt(ntohs(sin->sin_port));
	} else {
	struct sockaddr_in6* sin6 = (struct sockaddr_in6*)sa;
	byteBuffer->put(ByteArray(reinterpret_cast<char*>(&sin6->sin6_addr), kIPv6AddrSize));
	byteBuffer->putInt(ntohs(sin6->sin6_port));
	}
	byteBuffer->putLong(offset);
	byteBuffer->flip();
	return UtilAll::bytes2string(byteBuffer->array(), msgIdLength);
	}

	MessageId MessageDecoder::decodeMessageId(const std::string& msgId) {
	size_t ip_length = msgId.length() == 32 ? kIPv4AddrSize * 2 : kIPv6AddrSize * 2;

	ByteArray byteArray(ip_length / 2);
	std::string ip = msgId.substr(0, ip_length);
	UtilAll::string2bytes(byteArray.array(), ip);

	std::string port = msgId.substr(ip_length, 8);
	uint32_t portInt = std::stoul(port, nullptr, 16);

	auto* sin = IPPortToSockaddr(byteArray, portInt);

	std::string offset = msgId.substr(ip_length + 8);
	uint64_t offsetInt = std::stoull(offset, nullptr, 16);

	return MessageId(sin, offsetInt);
	}

	MessageExtPtr MessageDecoder::clientDecode(ByteBuffer& byteBuffer, bool readBody) {
	return decode(byteBuffer, readBody, true, true);
	}

	MessageExtPtr MessageDecoder::decode(ByteBuffer& byteBuffer) {
	return decode(byteBuffer, true);
	}

	MessageExtPtr MessageDecoder::decode(ByteBuffer& byteBuffer, bool readBody) {
	return decode(byteBuffer, readBody, true, false);
	}

	MessageExtPtr MessageDecoder::decode(ByteBuffer& byteBuffer, bool readBody, bool deCompressBody, bool isClient) {
	try {
	auto msgExt = isClient ? std::make_shared<MessageClientExtImpl>() : std::make_shared<MessageExtImpl>();

	// 1 TOTALSIZE
	int32_t storeSize = byteBuffer.getInt();
	msgExt->set_store_size(storeSize);

	// 2 MAGICCODE sizeof(int)
	byteBuffer.getInt();

	// 3 BODYCRC
	int32_t bodyCRC = byteBuffer.getInt();
	msgExt->set_body_crc(bodyCRC);

	// 4 QUEUEID
	int32_t queueId = byteBuffer.getInt();
	msgExt->set_queue_id(queueId);

	// 5 FLAG
	int32_t flag = byteBuffer.getInt();
	msgExt->set_flag(flag);

	// 6 QUEUEOFFSET
	int64_t queueOffset = byteBuffer.getLong();
	msgExt->set_queue_offset(queueOffset);

	// 7 PHYSICALOFFSET
	int64_t physicOffset = byteBuffer.getLong();
	msgExt->set_commit_log_offset(physicOffset);

	// 8 SYSFLAG
	int32_t sysFlag = byteBuffer.getInt();
	msgExt->set_sys_flag(sysFlag);

	// 9 BORNTIMESTAMP
	int64_t bornTimeStamp = byteBuffer.getLong();
	msgExt->set_born_timestamp(bornTimeStamp);

	// 10 BORNHOST
	int bornHostLength = (sysFlag & MessageSysFlag::BORNHOST_V6_FLAG) == 0 ? kIPv4AddrSize : kIPv6AddrSize;
	ByteArray bornHost(bornHostLength);
	byteBuffer.get(bornHost, 0, bornHostLength);
	int32_t bornPort = byteBuffer.getInt();
	msgExt->set_born_host(IPPortToSockaddr(bornHost, bornPort));

	// 11 STORETIMESTAMP
	int64_t storeTimestamp = byteBuffer.getLong();
	msgExt->set_store_timestamp(storeTimestamp);

	// 12 STOREHOST
	int storeHostLength = (sysFlag & MessageSysFlag::STOREHOST_V6_FLAG) == 0 ? kIPv4AddrSize : kIPv6AddrSize;
	ByteArray storeHost(storeHostLength);
	byteBuffer.get(storeHost, 0, storeHostLength);
	int32_t storePort = byteBuffer.getInt();
	msgExt->set_store_host(IPPortToSockaddr(storeHost, storePort));

	// 13 RECONSUMETIMES
	int32_t reconsumeTimes = byteBuffer.getInt();
	msgExt->set_reconsume_times(reconsumeTimes);

	// 14 Prepared Transaction Offset
	int64_t preparedTransactionOffset = byteBuffer.getLong();
	msgExt->set_prepared_transaction_offset(preparedTransactionOffset);

	// 15 BODY
	int uncompress_failed = false;
	int32_t bodyLen = byteBuffer.getInt();
	if (bodyLen > 0) {
	if (readBody) {
	ByteArray body(byteBuffer.array() + byteBuffer.arrayOffset() + byteBuffer.position(), bodyLen);
	byteBuffer.position(byteBuffer.position() + bodyLen);

	// decompress body
	if (deCompressBody && (sysFlag & MessageSysFlag::COMPRESSED_FLAG) == MessageSysFlag::COMPRESSED_FLAG) {
	std::string origin_body;
	if (UtilAll::inflate(body, origin_body)) {
	msgExt->set_body(std::move(origin_body));
	} else {
	uncompress_failed = true;
	}
	} else {
	msgExt->set_body(std::string(body.array(), body.size()));
	}
	} else {
	// skip body
	byteBuffer.position(byteBuffer.position() + bodyLen);
	}
	}

	// 16 TOPIC
	int8_t topicLen = byteBuffer.get();
	ByteArray topic(topicLen);
	byteBuffer.get(topic);
	msgExt->set_topic(topic.array(), topic.size());

	// 17 properties
	int16_t propertiesLen = byteBuffer.getShort();
	if (propertiesLen > 0) {
	ByteArray properties(propertiesLen);
	byteBuffer.get(properties);
	std::string propertiesString(properties.array(), properties.size());
	std::map<std::string, std::string> propertiesMap = string2messageProperties(propertiesString);
	MessageAccessor::setProperties(*msgExt, std::move(propertiesMap));
	}

	// 18 msg ID
	std::string msgId = createMessageId(msgExt->store_host(), (int64_t)msgExt->commit_log_offset());
	msgExt->MessageExtImpl::set_msg_id(msgId);

	if (uncompress_failed) {
	LOG_WARN_NEW("can not uncompress message, id:{}", msgExt->msg_id());
	}

	return msgExt;
	} catch (const std::exception& e) {
	byteBuffer.position(byteBuffer.limit());
	return nullptr;
	}
	}

	std::vector<MessageExtPtr> MessageDecoder::decodes(ByteBuffer& byteBuffer) {
	return decodes(byteBuffer, true);
	}

	std::vector<MessageExtPtr> MessageDecoder::decodes(ByteBuffer& byteBuffer, bool readBody) {
	std::vector<MessageExtPtr> msgExts;
	while (byteBuffer.hasRemaining()) {
	auto msgExt = clientDecode(byteBuffer, readBody);
	if (nullptr == msgExt) {
	break;
	}
	msgExts.emplace_back(std::move(msgExt));
	}
	return msgExts;
	}

	std::string MessageDecoder::messageProperties2String(const std::map<std::string, std::string>& properties) {
	std::string os;

	for (const auto& it : properties) {
	const auto& name = it.first;
	const auto& value = it.second;

	// skip compressed flag
	if (MQMessageConst::PROPERTY_ALREADY_COMPRESSED_FLAG == name) {
	continue;
	}

	os.append(name);
	os += NAME_VALUE_SEPARATOR;
	os.append(value);
	os += PROPERTY_SEPARATOR;
	}

	return os;
	}

	std::map<std::string, std::string> MessageDecoder::string2messageProperties(const std::string& properties) {
	std::vector<std::string> out;
	UtilAll::Split(out, properties, PROPERTY_SEPARATOR);

	std::map<std::string, std::string> map;
	for (size_t i = 0; i < out.size(); i++) {
	std::vector<std::string> outValue;
	UtilAll::Split(outValue, out[i], NAME_VALUE_SEPARATOR);

	if (outValue.size() == 2) {
	map[outValue[0]] = outValue[1];
	}
	}
	return map;
	}

	std::string MessageDecoder::encodeMessage(Message& message) {
	const auto& body = message.body();
	uint32_t bodyLen = body.size();
	std::string properties = MessageDecoder::messageProperties2String(message.properties());
	uint16_t propertiesLength = (int16_t)properties.size();
	uint32_t storeSize = 4 // 1 TOTALSIZE
	+ 4 // 2 MAGICCODE
	+ 4 // 3 BODYCRC
	+ 4 // 4 FLAG
	+ 4 + bodyLen // 5 BODY
	+ 2 + propertiesLength; // 6 PROPERTIES

	// TOTALSIZE\|MAGICCODE\|BODYCRC\|FLAG\|BodyLen\|Body\|propertiesLength\|properties
	std::string encodeMsg;

	// 1 TOTALSIZE
	uint32_t storeSize_net = ByteOrderUtil::NorminalBigEndian(storeSize);
	encodeMsg.append((char*)&storeSize_net, sizeof(uint32_t));

	// 2 MAGICCODE
	uint32_t magicCode = 0;
	uint32_t magicCode_net = ByteOrderUtil::NorminalBigEndian(magicCode);
	encodeMsg.append((char*)&magicCode_net, sizeof(uint32_t));

	// 3 BODYCRC
	uint32_t bodyCrc = 0;
	uint32_t bodyCrc_net = ByteOrderUtil::NorminalBigEndian(bodyCrc);
	encodeMsg.append((char*)&bodyCrc_net, sizeof(uint32_t));

	// 4 FLAG
	uint32_t flag = message.flag();
	uint32_t flag_net = ByteOrderUtil::NorminalBigEndian(flag);
	encodeMsg.append((char*)&flag_net, sizeof(uint32_t));

	// 5 BODY
	uint32_t bodyLen_net = ByteOrderUtil::NorminalBigEndian(bodyLen);
	encodeMsg.append((char*)&bodyLen_net, sizeof(uint32_t));
	encodeMsg.append(body.data(), body.size());

	// 6 properties
	uint16_t propertiesLength_net = ByteOrderUtil::NorminalBigEndian(propertiesLength);
	encodeMsg.append((char*)&propertiesLength_net, sizeof(uint16_t));
	encodeMsg.append(std::move(properties));

	return encodeMsg;
	}

	std::string MessageDecoder::encodeMessages(std::vector<MQMessage>& msgs) {
	std::string encodedBody;
	for (auto msg : msgs) {
	encodedBody.append(encodeMessage(msg));
	}
	return encodedBody;
	}

	} // namespace rocketmq