You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by if...@apache.org on 2020/12/29 03:36:21 UTC
[rocketmq-client-cpp] 04/29: refactor(MQMessage): MQMessage[Ext] as
reference
This is an automated email from the ASF dual-hosted git repository.
ifplusor pushed a commit to branch re_dev
in repository https://gitbox.apache.org/repos/asf/rocketmq-client-cpp.git
commit 4d79bcf0f13b6962cfd0f1d86d649f372e32e158
Author: James Yin <yw...@hotmail.com>
AuthorDate: Sat Jul 18 21:47:05 2020 +0800
refactor(MQMessage): MQMessage[Ext] as reference
---
example/PullConsumer.cpp | 10 +-
example/common.h | 4 +-
include/DefaultMQProducer.h | 42 ++--
include/DefaultMQPullConsumer.h | 4 +-
include/DefaultMQPushConsumer.h | 4 +-
include/MQAdmin.h | 2 +-
include/MQConsumer.h | 4 +-
include/MQMessage.h | 121 ++++++------
include/MQMessageConst.h | 2 +-
include/MQMessageExt.h | 138 ++++++-------
include/MQMessageListener.h | 13 +-
include/MQProducer.h | 46 ++---
include/Message.h | 94 +++++++++
include/MessageExt.h | 88 +++++++++
include/MessageUtil.h | 13 +-
include/QueryResult.h | 6 +-
include/RequestCallback.h | 4 +-
include/RocketMQClient.h | 10 +-
include/TopicFilterType.h | 7 +-
include/TransactionMQProducer.h | 2 +-
src/ClientRemotingProcessor.cpp | 6 +-
src/MQAdminImpl.cpp | 4 +-
src/MQAdminImpl.h | 2 +-
src/MQClientAPIImpl.cpp | 24 +--
src/MQClientAPIImpl.h | 14 +-
src/MQClientImpl.cpp | 2 +-
src/MQClientImpl.h | 2 +-
src/common/ByteOrder.h | 21 +-
src/common/MessageSysFlag.cpp | 3 +
src/common/MessageSysFlag.h | 3 +
src/common/SendCallbackWrap.cpp | 4 +-
src/common/SendCallbackWrap.h | 8 +-
src/common/UtilAll.cpp | 62 +++---
src/common/UtilAll.h | 3 +-
src/common/Validators.cpp | 2 +-
src/common/Validators.h | 4 +-
src/consumer/ConsumeMessageConcurrentlyService.cpp | 19 +-
src/consumer/ConsumeMessageOrderlyService.cpp | 14 +-
src/consumer/ConsumeMsgService.h | 10 +-
src/consumer/DefaultMQPullConsumer.cpp | 4 +-
src/consumer/DefaultMQPushConsumer.cpp | 4 +-
src/consumer/DefaultMQPushConsumerImpl.cpp | 38 ++--
src/consumer/DefaultMQPushConsumerImpl.h | 6 +-
src/consumer/ProcessQueue.cpp | 8 +-
src/consumer/ProcessQueue.h | 14 +-
src/consumer/PullAPIWrapper.cpp | 18 +-
src/consumer/PullResult.cpp | 47 +++--
{include => src/consumer}/PullResult.h | 39 ++--
src/extern/CBatchMessage.cpp | 9 +-
src/extern/CProducer.cpp | 20 +-
src/extern/CPullConsumer.cpp | 20 +-
src/extern/CPushConsumer.cpp | 10 +-
src/message/MQMessage.cpp | 174 +++++------------
src/message/MQMessageConst.cpp | 60 ++++++
src/message/MQMessageExt.cpp | 133 ++++---------
src/message/MQMessageId.h | 55 ------
src/message/MessageAccessor.h | 18 +-
src/message/MessageBatch.cpp | 37 ++--
src/message/MessageBatch.h | 20 +-
src/message/MessageClientIDSetter.h | 4 +-
src/message/{MQDecoder.cpp => MessageDecoder.cpp} | 137 +++++++------
src/message/{MQDecoder.h => MessageDecoder.h} | 36 ++--
src/message/MessageExtImpl.cpp | 214 +++++++++++++++++++++
src/message/MessageExtImpl.h | 117 +++++++++++
src/message/MessageId.h | 58 ++++++
src/message/MessageImpl.cpp | 187 ++++++++++++++++++
src/message/MessageImpl.h | 91 +++++++++
src/message/MessageUtil.cpp | 52 +++--
src/producer/DefaultMQProducer.cpp | 42 ++--
src/producer/DefaultMQProducerImpl.cpp | 177 +++++++++--------
src/producer/DefaultMQProducerImpl.h | 60 +++---
src/producer/MQProducerInner.h | 2 +-
src/producer/RequestResponseFuture.cpp | 6 +-
src/producer/RequestResponseFuture.h | 10 +-
src/producer/TransactionMQProducer.cpp | 2 +-
src/transport/SocketUtil.cpp | 14 +-
test/src/message/MessageBatchTest.cpp | 24 +--
.../{MQDecoderTest.cpp => MessageDecoderTest.cpp} | 26 +--
.../{MQMessageIdTest.cpp => MessageIdTest.cpp} | 6 +-
79 files changed, 1776 insertions(+), 1044 deletions(-)
diff --git a/example/PullConsumer.cpp b/example/PullConsumer.cpp
index 7bd7ffd..e7472b3 100644
--- a/example/PullConsumer.cpp
+++ b/example/PullConsumer.cpp
@@ -77,18 +77,18 @@ int main(int argc, char* argv[]) {
do {
try {
PullResult result = consumer.pull(mq, "*", getMessageQueueOffset(mq), 32);
- g_msgCount += result.msgFoundList.size();
- std::cout << result.msgFoundList.size() << std::endl;
+ g_msgCount += result.msg_found_list_.size();
+ std::cout << result.msg_found_list_.size() << std::endl;
// if pull request timeout or received NULL response, pullStatus will be
// setted to BROKER_TIMEOUT,
// And nextBeginOffset/minOffset/MaxOffset will be setted to 0
- if (result.pullStatus != BROKER_TIMEOUT) {
- putMessageQueueOffset(mq, result.nextBeginOffset);
+ if (result.pull_status_ != BROKER_TIMEOUT) {
+ putMessageQueueOffset(mq, result.next_begin_offset_);
PrintPullResult(&result);
} else {
std::cout << "broker timeout occur" << std::endl;
}
- switch (result.pullStatus) {
+ switch (result.pull_status_) {
case FOUND:
case NO_MATCHED_MSG:
case OFFSET_ILLEGAL:
diff --git a/example/common.h b/example/common.h
index 84683bc..90bff8e 100644
--- a/example/common.h
+++ b/example/common.h
@@ -114,9 +114,9 @@ static void PrintResult(rocketmq::SendResult* result) {
void PrintPullResult(rocketmq::PullResult* result) {
std::cout << result->toString() << std::endl;
- if (result->pullStatus == rocketmq::FOUND) {
+ if (result->pull_status_ == rocketmq::FOUND) {
std::cout << result->toString() << std::endl;
- for (auto& msg : result->msgFoundList) {
+ for (auto& msg : result->msg_found_list_) {
std::cout << "=======================================================" << std::endl
<< msg->toString() << std::endl;
}
diff --git a/include/DefaultMQProducer.h b/include/DefaultMQProducer.h
index 91e0e2b..2219dfe 100644
--- a/include/DefaultMQProducer.h
+++ b/include/DefaultMQProducer.h
@@ -38,43 +38,43 @@ class ROCKETMQCLIENT_API DefaultMQProducer : public MQProducer, public DefaultMQ
void shutdown() override;
// Sync: caller will be responsible for the lifecycle of messages.
- SendResult send(MQMessagePtr msg) override;
- SendResult send(MQMessagePtr msg, long timeout) override;
- SendResult send(MQMessagePtr msg, const MQMessageQueue& mq) override;
- SendResult send(MQMessagePtr msg, const MQMessageQueue& mq, long timeout) override;
+ SendResult send(MQMessage& msg) override;
+ SendResult send(MQMessage& msg, long timeout) override;
+ SendResult send(MQMessage& msg, const MQMessageQueue& mq) override;
+ SendResult send(MQMessage& msg, const MQMessageQueue& mq, long timeout) override;
// Async: don't delete msg object, until callback occur.
- void send(MQMessagePtr msg, SendCallback* sendCallback) noexcept override;
- void send(MQMessagePtr msg, SendCallback* sendCallback, long timeout) noexcept override;
- void send(MQMessagePtr msg, const MQMessageQueue& mq, SendCallback* sendCallback) noexcept override;
- void send(MQMessagePtr msg, const MQMessageQueue& mq, SendCallback* sendCallback, long timeout) noexcept override;
+ void send(MQMessage& msg, SendCallback* sendCallback) noexcept override;
+ void send(MQMessage& msg, SendCallback* sendCallback, long timeout) noexcept override;
+ void send(MQMessage& msg, const MQMessageQueue& mq, SendCallback* sendCallback) noexcept override;
+ void send(MQMessage& msg, const MQMessageQueue& mq, SendCallback* sendCallback, long timeout) noexcept override;
// Oneyway: same as sync send, but don't care its result.
- void sendOneway(MQMessagePtr msg) override;
- void sendOneway(MQMessagePtr msg, const MQMessageQueue& mq) override;
+ void sendOneway(MQMessage& msg) override;
+ void sendOneway(MQMessage& msg, const MQMessageQueue& mq) override;
// Select
- SendResult send(MQMessagePtr msg, MessageQueueSelector* selector, void* arg) override;
- SendResult send(MQMessagePtr msg, MessageQueueSelector* selector, void* arg, long timeout) override;
- void send(MQMessagePtr msg, MessageQueueSelector* selector, void* arg, SendCallback* sendCallback) noexcept override;
- void send(MQMessagePtr msg,
+ SendResult send(MQMessage& msg, MessageQueueSelector* selector, void* arg) override;
+ SendResult send(MQMessage& msg, MessageQueueSelector* selector, void* arg, long timeout) override;
+ void send(MQMessage& msg, MessageQueueSelector* selector, void* arg, SendCallback* sendCallback) noexcept override;
+ void send(MQMessage& msg,
MessageQueueSelector* selector,
void* arg,
SendCallback* sendCallback,
long timeout) noexcept override;
- void sendOneway(MQMessagePtr msg, MessageQueueSelector* selector, void* arg) override;
+ void sendOneway(MQMessage& msg, MessageQueueSelector* selector, void* arg) override;
// Transaction
- TransactionSendResult sendMessageInTransaction(MQMessagePtr msg, void* arg) override;
+ TransactionSendResult sendMessageInTransaction(MQMessage& msg, void* arg) override;
// Batch: power by sync send, caller will be responsible for the lifecycle of messages.
- SendResult send(std::vector<MQMessagePtr>& msgs) override;
- SendResult send(std::vector<MQMessagePtr>& msgs, long timeout) override;
- SendResult send(std::vector<MQMessagePtr>& msgs, const MQMessageQueue& mq) override;
- SendResult send(std::vector<MQMessagePtr>& msgs, const MQMessageQueue& mq, long timeout) override;
+ SendResult send(std::vector<MQMessage>& msgs) override;
+ SendResult send(std::vector<MQMessage>& msgs, long timeout) override;
+ SendResult send(std::vector<MQMessage>& msgs, const MQMessageQueue& mq) override;
+ SendResult send(std::vector<MQMessage>& msgs, const MQMessageQueue& mq, long timeout) override;
// RPC
- MQMessagePtr request(MQMessagePtr msg, long timeout) override;
+ MQMessage request(MQMessage& msg, long timeout) override;
public: // DefaultMQProducerConfig
bool isSendLatencyFaultEnable() const override;
diff --git a/include/DefaultMQPullConsumer.h b/include/DefaultMQPullConsumer.h
index 8461969..8b4c719 100755
--- a/include/DefaultMQPullConsumer.h
+++ b/include/DefaultMQPullConsumer.h
@@ -37,8 +37,8 @@ class ROCKETMQCLIENT_API DefaultMQPullConsumer : public MQPullConsumer, public D
void start() override;
void shutdown() override;
- bool sendMessageBack(MQMessageExt& msg, int delayLevel) override;
- bool sendMessageBack(MQMessageExt& msg, int delayLevel, const std::string& brokerName) override;
+ bool sendMessageBack(MessageExtPtr msg, int delayLevel) override;
+ bool sendMessageBack(MessageExtPtr msg, int delayLevel, const std::string& brokerName) override;
void fetchSubscribeMessageQueues(const std::string& topic, std::vector<MQMessageQueue>& mqs) override;
public: // MQPullConsumer
diff --git a/include/DefaultMQPushConsumer.h b/include/DefaultMQPushConsumer.h
index e90eb95..83e4732 100755
--- a/include/DefaultMQPushConsumer.h
+++ b/include/DefaultMQPushConsumer.h
@@ -33,8 +33,8 @@ class ROCKETMQCLIENT_API DefaultMQPushConsumer : public MQPushConsumer, public D
void start() override;
void shutdown() override;
- bool sendMessageBack(MQMessageExt& msg, int delayLevel) override;
- bool sendMessageBack(MQMessageExt& msg, int delayLevel, const std::string& brokerName) override;
+ bool sendMessageBack(MessageExtPtr msg, int delayLevel) override;
+ bool sendMessageBack(MessageExtPtr msg, int delayLevel, const std::string& brokerName) override;
void fetchSubscribeMessageQueues(const std::string& topic, std::vector<MQMessageQueue>& mqs) override;
public: // MQPushConsumer
diff --git a/include/MQAdmin.h b/include/MQAdmin.h
index 50179ce..eb681fb 100644
--- a/include/MQAdmin.h
+++ b/include/MQAdmin.h
@@ -79,7 +79,7 @@ class ROCKETMQCLIENT_API MQAdmin {
* @param offsetMsgId message id
* @return message
*/
- virtual MQMessageExtPtr viewMessage(const std::string& offsetMsgId) = 0;
+ virtual MQMessageExt viewMessage(const std::string& offsetMsgId) = 0;
/**
* Query messages
diff --git a/include/MQConsumer.h b/include/MQConsumer.h
index ebbda3f..54ff576 100755
--- a/include/MQConsumer.h
+++ b/include/MQConsumer.h
@@ -35,8 +35,8 @@ class ROCKETMQCLIENT_API MQConsumer {
virtual void start() = 0;
virtual void shutdown() = 0;
- virtual bool sendMessageBack(MQMessageExt& msg, int delayLevel) = 0;
- virtual bool sendMessageBack(MQMessageExt& msg, int delayLevel, const std::string& brokerName) = 0;
+ virtual bool sendMessageBack(MessageExtPtr msg, int delayLevel) = 0;
+ virtual bool sendMessageBack(MessageExtPtr msg, int delayLevel, const std::string& brokerName) = 0;
virtual void fetchSubscribeMessageQueues(const std::string& topic, std::vector<MQMessageQueue>& mqs) = 0;
};
diff --git a/include/MQMessage.h b/include/MQMessage.h
index 19f581d..3016ec5 100644
--- a/include/MQMessage.h
+++ b/include/MQMessage.h
@@ -14,25 +14,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef __MQ_MESSAGE_H__
-#define __MQ_MESSAGE_H__
+#ifndef ROCKETMQ_MQMESSAGE_H_
+#define ROCKETMQ_MQMESSAGE_H_
-#include <map>
-#include <memory>
-#include <sstream>
-#include <string>
-#include <vector>
+#include <cstddef> // std::nullptr_t
+#include "Message.h"
#include "MQMessageConst.h"
namespace rocketmq {
-class MQMessage;
-typedef MQMessage* MQMessagePtr;
-typedef std::shared_ptr<MQMessage> MQMessagePtr2;
-typedef std::unique_ptr<MQMessage> MQMessagePtr3;
-
-class ROCKETMQCLIENT_API MQMessage {
+/**
+ * MQMessage - Reference of Message
+ */
+class ROCKETMQCLIENT_API MQMessage : virtual public Message // interface
+{
public:
MQMessage();
MQMessage(const std::string& topic, const std::string& body);
@@ -41,68 +37,83 @@ class ROCKETMQCLIENT_API MQMessage {
MQMessage(const std::string& topic,
const std::string& tags,
const std::string& keys,
- const int flag,
+ int32_t flag,
const std::string& body,
bool waitStoreMsgOK);
- MQMessage(const MQMessage& other);
- MQMessage& operator=(const MQMessage& other);
+ public:
+ // reference constructor
+ MQMessage(MessagePtr impl) : message_impl_(impl) {}
+
+ // copy constructor
+ MQMessage(const MQMessage& other) : message_impl_(other.message_impl_) {}
+
+ // assign operator
+ MQMessage& operator=(const MQMessage& other) {
+ if (this != &other) {
+ message_impl_ = other.message_impl_;
+ }
+ return *this;
+ }
+
+ bool operator==(std::nullptr_t) noexcept { return nullptr == message_impl_; }
+
+ friend bool operator==(std::nullptr_t, const MQMessage& message) noexcept { return nullptr == message.message_impl_; }
+ // convert to boolean
+ operator bool() noexcept { return nullptr != message_impl_; }
+
+ public:
virtual ~MQMessage();
- const std::string& getProperty(const std::string& name) const;
- void putProperty(const std::string& name, const std::string& value);
- void clearProperty(const std::string& name);
+ public: // Message
+ const std::string& getProperty(const std::string& name) const override;
+ void putProperty(const std::string& name, const std::string& value) override;
+ void clearProperty(const std::string& name) override;
- const std::string& getTopic() const;
- void setTopic(const std::string& topic);
- void setTopic(const char* body, int len);
+ const std::string& getTopic() const override;
+ void setTopic(const std::string& topic) override;
+ void setTopic(const char* body, int len) override;
- const std::string& getTags() const;
- void setTags(const std::string& tags);
+ const std::string& getTags() const override;
+ void setTags(const std::string& tags) override;
- const std::string& getKeys() const;
- void setKeys(const std::string& keys);
- void setKeys(const std::vector<std::string>& keys);
+ const std::string& getKeys() const override;
+ void setKeys(const std::string& keys) override;
+ void setKeys(const std::vector<std::string>& keys) override;
- int getDelayTimeLevel() const;
- void setDelayTimeLevel(int level);
+ int getDelayTimeLevel() const override;
+ void setDelayTimeLevel(int level) override;
- bool isWaitStoreMsgOK() const;
- void setWaitStoreMsgOK(bool waitStoreMsgOK);
+ bool isWaitStoreMsgOK() const override;
+ void setWaitStoreMsgOK(bool waitStoreMsgOK) override;
- int getFlag() const;
- void setFlag(int flag);
+ int32_t getFlag() const override;
+ void setFlag(int32_t flag) override;
- const std::string& getBody() const;
- void setBody(const char* body, int len);
- void setBody(const std::string& body);
- void setBody(std::string&& body);
+ const std::string& getBody() const override;
+ void setBody(const char* body, int len) override;
+ void setBody(const std::string& body) override;
+ void setBody(std::string&& body) override;
- const std::string& getTransactionId() const;
- void setTransactionId(const std::string& transactionId);
+ const std::string& getTransactionId() const override;
+ void setTransactionId(const std::string& transactionId) override;
- const std::map<std::string, std::string>& getProperties() const;
- void setProperties(const std::map<std::string, std::string>& properties);
- void setProperties(std::map<std::string, std::string>&& properties);
+ const std::map<std::string, std::string>& getProperties() const override;
+ void setProperties(const std::map<std::string, std::string>& properties) override;
+ void setProperties(std::map<std::string, std::string>&& properties) override;
- virtual bool isBatch() { return false; }
+ bool isBatch() const override;
- virtual std::string toString() const {
- std::stringstream ss;
- ss << "Message [topic=" << m_topic << ", flag=" << m_flag << ", tag=" << getTags() << ", transactionId='"
- << m_transactionId + "']";
- return ss.str();
- }
+ std::string toString() const override;
+
+ public:
+ MessagePtr getMessageImpl();
protected:
- std::string m_topic;
- int m_flag;
- std::map<std::string, std::string> m_properties;
- std::string m_body;
- std::string m_transactionId;
+ MessagePtr message_impl_;
};
} // namespace rocketmq
-#endif // __MQ_MESSAGE_H__
+#endif // ROCKETMQ_MQMESSAGE_H_
diff --git a/include/MQMessageConst.h b/include/MQMessageConst.h
index 0657219..071d94d 100644
--- a/include/MQMessageConst.h
+++ b/include/MQMessageConst.h
@@ -17,7 +17,7 @@
#ifndef __MQ_MESSAGE_CONST_H__
#define __MQ_MESSAGE_CONST_H__
-#include <string>
+#include <string> // std::string
#include "RocketMQClient.h"
diff --git a/include/MQMessageExt.h b/include/MQMessageExt.h
index 42e5078..0001f70 100644
--- a/include/MQMessageExt.h
+++ b/include/MQMessageExt.h
@@ -14,30 +14,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef __MQ_MESSAGE_EXT_H__
-#define __MQ_MESSAGE_EXT_H__
-
-#ifdef WIN32
-// clang-format off
-#include <Winsock2.h>
-#include <Windows.h>
-// clang-format on
-#else
-#include <sys/socket.h>
-#endif
+#ifndef ROCKETMQ_MQMESSAGEEXT_H_
+#define ROCKETMQ_MQMESSAGEEXT_H_
+#include "MessageExt.h"
#include "MQMessage.h"
-#include "TopicFilterType.h"
namespace rocketmq {
-class MQMessageExt;
-
-typedef MQMessageExt* MQMessageExtPtr;
-typedef std::shared_ptr<MQMessageExt> MQMessageExtPtr2;
-
-// message extend class, which was generated on broker
-class ROCKETMQCLIENT_API MQMessageExt : public MQMessage {
+/**
+ * MQMessageExt - Reference of MessageExt
+ */
+class ROCKETMQCLIENT_API MQMessageExt : public MQMessage, // base
+ virtual public MessageExt // interface
+{
public:
MQMessageExt();
MQMessageExt(int queueId,
@@ -45,84 +35,64 @@ class ROCKETMQCLIENT_API MQMessageExt : public MQMessage {
const struct sockaddr* bornHost,
int64_t storeTimestamp,
const struct sockaddr* storeHost,
- std::string msgId);
-
- virtual ~MQMessageExt();
-
- static TopicFilterType parseTopicFilterType(int32_t sysFlag);
-
- int32_t getStoreSize() const;
- void setStoreSize(int32_t storeSize);
-
- int32_t getBodyCRC() const;
- void setBodyCRC(int32_t bodyCRC);
+ const std::string& msgId);
- int32_t getQueueId() const;
- void setQueueId(int32_t queueId);
+ public:
+ // reference constructor
+ MQMessageExt(MessageExtPtr impl) : MQMessage(impl) {}
+
+ // copy constructor
+ MQMessageExt(const MQMessageExt& other) : MQMessage(other) {}
+
+ // assign operator
+ MQMessageExt& operator=(const MQMessageExt& other) {
+ if (this != &other) {
+ message_impl_ = other.message_impl_;
+ }
+ return *this;
+ }
- int64_t getQueueOffset() const;
- void setQueueOffset(int64_t queueOffset);
+ public:
+ virtual ~MQMessageExt();
- int64_t getCommitLogOffset() const;
- void setCommitLogOffset(int64_t physicOffset);
+ public: // MessageExt
+ int32_t getStoreSize() const override;
+ void setStoreSize(int32_t storeSize) override;
- int32_t getSysFlag() const;
- void setSysFlag(int32_t sysFlag);
+ int32_t getBodyCRC() const override;
+ void setBodyCRC(int32_t bodyCRC) override;
- int64_t getBornTimestamp() const;
- void setBornTimestamp(int64_t bornTimestamp);
+ int32_t getQueueId() const override;
+ void setQueueId(int32_t queueId) override;
- const struct sockaddr* getBornHost() const;
- std::string getBornHostString() const;
- void setBornHost(const struct sockaddr* bornHost);
+ int64_t getQueueOffset() const override;
+ void setQueueOffset(int64_t queueOffset) override;
- int64_t getStoreTimestamp() const;
- void setStoreTimestamp(int64_t storeTimestamp);
+ int64_t getCommitLogOffset() const override;
+ void setCommitLogOffset(int64_t physicOffset) override;
- const struct sockaddr* getStoreHost() const;
- std::string getStoreHostString() const;
- void setStoreHost(const struct sockaddr* storeHost);
+ int32_t getSysFlag() const override;
+ void setSysFlag(int32_t sysFlag) override;
- int32_t getReconsumeTimes() const;
- void setReconsumeTimes(int32_t reconsumeTimes);
+ int64_t getBornTimestamp() const override;
+ void setBornTimestamp(int64_t bornTimestamp) override;
- int64_t getPreparedTransactionOffset() const;
- void setPreparedTransactionOffset(int64_t preparedTransactionOffset);
+ std::string getBornHostString() const override;
+ const struct sockaddr* getBornHost() const override;
+ void setBornHost(const struct sockaddr* bornHost) override;
- virtual const std::string& getMsgId() const;
- virtual void setMsgId(const std::string& msgId);
+ int64_t getStoreTimestamp() const override;
+ void setStoreTimestamp(int64_t storeTimestamp) override;
- std::string toString() const override {
- std::stringstream ss;
- ss << "MessageExt [queueId=" << m_queueId << ", storeSize=" << m_storeSize << ", queueOffset=" << m_queueOffset
- << ", sysFlag=" << m_sysFlag << ", bornTimestamp=" << m_bornTimestamp << ", bornHost=" << getBornHostString()
- << ", storeTimestamp=" << m_storeTimestamp << ", storeHost=" << getStoreHostString() << ", msgId=" << getMsgId()
- << ", commitLogOffset=" << m_commitLogOffset << ", bodyCRC=" << m_bodyCRC
- << ", reconsumeTimes=" << m_reconsumeTimes << ", preparedTransactionOffset=" << m_preparedTransactionOffset
- << ", toString()=" << MQMessage::toString() << "]";
- return ss.str();
- }
+ std::string getStoreHostString() const override;
+ const struct sockaddr* getStoreHost() const override;
+ void setStoreHost(const struct sockaddr* storeHost) override;
- private:
- int32_t m_storeSize;
- int32_t m_bodyCRC;
- int32_t m_queueId;
- int64_t m_queueOffset;
- int64_t m_commitLogOffset;
- int32_t m_sysFlag;
- int64_t m_bornTimestamp;
- struct sockaddr* m_bornHost;
- int64_t m_storeTimestamp;
- struct sockaddr* m_storeHost;
- int32_t m_reconsumeTimes;
- int64_t m_preparedTransactionOffset;
- std::string m_msgId;
-};
+ int32_t getReconsumeTimes() const override;
+ void setReconsumeTimes(int32_t reconsumeTimes) override;
-class ROCKETMQCLIENT_API MQMessageClientExt : public MQMessageExt {
- public:
- const std::string& getOffsetMsgId() const;
- void setOffsetMsgId(const std::string& offsetMsgId);
+ int64_t getPreparedTransactionOffset() const override;
+ void setPreparedTransactionOffset(int64_t preparedTransactionOffset) override;
const std::string& getMsgId() const override;
void setMsgId(const std::string& msgId) override;
@@ -130,4 +100,4 @@ class ROCKETMQCLIENT_API MQMessageClientExt : public MQMessageExt {
} // namespace rocketmq
-#endif // __MQ_MESSAGE_EXT_H__
+#endif // ROCKETMQ_MQMESSAGEEXT_H_
diff --git a/include/MQMessageListener.h b/include/MQMessageListener.h
index 61bb243..b127632 100644
--- a/include/MQMessageListener.h
+++ b/include/MQMessageListener.h
@@ -38,18 +38,7 @@ class ROCKETMQCLIENT_API MQMessageListener {
virtual MessageListenerType getMessageListenerType() { return messageListenerDefaultly; }
- // Recommended
- virtual ConsumeStatus consumeMessage(const std::vector<MQMessageExtPtr2>& msgs) {
- std::vector<MQMessageExtPtr> msgs2;
- msgs2.reserve(msgs.size());
- for (auto& msg : msgs) {
- msgs2.push_back(msg.get());
- }
- return consumeMessage(msgs2);
- }
-
- // SDK will be responsible for the lifecycle of messages.
- virtual ConsumeStatus consumeMessage(const std::vector<MQMessageExtPtr>& msgs) { return RECONSUME_LATER; };
+ virtual ConsumeStatus consumeMessage(const std::vector<MQMessageExt>& msgs) { return RECONSUME_LATER; };
};
class ROCKETMQCLIENT_API MessageListenerConcurrently : virtual public MQMessageListener {
diff --git a/include/MQProducer.h b/include/MQProducer.h
index 81a1c31..dc32fa3 100644
--- a/include/MQProducer.h
+++ b/include/MQProducer.h
@@ -36,47 +36,43 @@ class ROCKETMQCLIENT_API MQProducer {
virtual void shutdown() = 0;
// Sync
- virtual SendResult send(MQMessagePtr msg) = 0;
- virtual SendResult send(MQMessagePtr msg, long timeout) = 0;
- virtual SendResult send(MQMessagePtr msg, const MQMessageQueue& mq) = 0;
- virtual SendResult send(MQMessagePtr msg, const MQMessageQueue& mq, long timeout) = 0;
+ virtual SendResult send(MQMessage& msg) = 0;
+ virtual SendResult send(MQMessage& msg, long timeout) = 0;
+ virtual SendResult send(MQMessage& msg, const MQMessageQueue& mq) = 0;
+ virtual SendResult send(MQMessage& msg, const MQMessageQueue& mq, long timeout) = 0;
// Async
- // TODO: replace MQMessagePtr by MQMessagePtr2(shared_ptr)
- virtual void send(MQMessagePtr msg, SendCallback* sendCallback) noexcept = 0;
- virtual void send(MQMessagePtr msg, SendCallback* sendCallback, long timeout) noexcept = 0;
- virtual void send(MQMessagePtr msg, const MQMessageQueue& mq, SendCallback* sendCallback) noexcept = 0;
- virtual void send(MQMessagePtr msg, const MQMessageQueue& mq, SendCallback* sendCallback, long timeout) noexcept = 0;
+ virtual void send(MQMessage& msg, SendCallback* sendCallback) noexcept = 0;
+ virtual void send(MQMessage& msg, SendCallback* sendCallback, long timeout) noexcept = 0;
+ virtual void send(MQMessage& msg, const MQMessageQueue& mq, SendCallback* sendCallback) noexcept = 0;
+ virtual void send(MQMessage& msg, const MQMessageQueue& mq, SendCallback* sendCallback, long timeout) noexcept = 0;
// Oneyway
- virtual void sendOneway(MQMessagePtr msg) = 0;
- virtual void sendOneway(MQMessagePtr msg, const MQMessageQueue& mq) = 0;
+ virtual void sendOneway(MQMessage& msg) = 0;
+ virtual void sendOneway(MQMessage& msg, const MQMessageQueue& mq) = 0;
// Select
- virtual SendResult send(MQMessagePtr msg, MessageQueueSelector* selector, void* arg) = 0;
- virtual SendResult send(MQMessagePtr msg, MessageQueueSelector* selector, void* arg, long timeout) = 0;
- virtual void send(MQMessagePtr msg,
- MessageQueueSelector* selector,
- void* arg,
- SendCallback* sendCallback) noexcept = 0;
- virtual void send(MQMessagePtr msg,
+ virtual SendResult send(MQMessage& msg, MessageQueueSelector* selector, void* arg) = 0;
+ virtual SendResult send(MQMessage& msg, MessageQueueSelector* selector, void* arg, long timeout) = 0;
+ virtual void send(MQMessage& msg, MessageQueueSelector* selector, void* arg, SendCallback* sendCallback) noexcept = 0;
+ virtual void send(MQMessage& msg,
MessageQueueSelector* selector,
void* arg,
SendCallback* sendCallback,
long timeout) noexcept = 0;
- virtual void sendOneway(MQMessagePtr msg, MessageQueueSelector* selector, void* arg) = 0;
+ virtual void sendOneway(MQMessage& msg, MessageQueueSelector* selector, void* arg) = 0;
// Transaction
- virtual TransactionSendResult sendMessageInTransaction(MQMessagePtr msg, void* arg) = 0;
+ virtual TransactionSendResult sendMessageInTransaction(MQMessage& msg, void* arg) = 0;
// Batch
- virtual SendResult send(std::vector<MQMessagePtr>& msgs) = 0;
- virtual SendResult send(std::vector<MQMessagePtr>& msgs, long timeout) = 0;
- virtual SendResult send(std::vector<MQMessagePtr>& msgs, const MQMessageQueue& mq) = 0;
- virtual SendResult send(std::vector<MQMessagePtr>& msgs, const MQMessageQueue& mq, long timeout) = 0;
+ virtual SendResult send(std::vector<MQMessage>& msgs) = 0;
+ virtual SendResult send(std::vector<MQMessage>& msgs, long timeout) = 0;
+ virtual SendResult send(std::vector<MQMessage>& msgs, const MQMessageQueue& mq) = 0;
+ virtual SendResult send(std::vector<MQMessage>& msgs, const MQMessageQueue& mq, long timeout) = 0;
// RPC
- virtual MQMessagePtr request(MQMessagePtr msg, long timeout) = 0;
+ virtual MQMessage request(MQMessage& msg, long timeout) = 0;
};
} // namespace rocketmq
diff --git a/include/Message.h b/include/Message.h
new file mode 100644
index 0000000..0320c0e
--- /dev/null
+++ b/include/Message.h
@@ -0,0 +1,94 @@
+/*
+ * 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.
+ */
+#ifndef ROCKETMQ_MESSAGE_H_
+#define ROCKETMQ_MESSAGE_H_
+
+#include <map> // std::map
+#include <string> // std::string
+#include <vector> // std::vector
+#include <memory> // std::shared_ptr
+
+#include "RocketMQClient.h"
+
+namespace rocketmq {
+
+class Message;
+typedef std::shared_ptr<Message> MessagePtr;
+
+/**
+ * Message Interface
+ */
+class ROCKETMQCLIENT_API Message {
+ public:
+ virtual ~Message() = default;
+
+ public:
+ // property
+ virtual const std::string& getProperty(const std::string& name) const = 0;
+ virtual void putProperty(const std::string& name, const std::string& value) = 0;
+ virtual void clearProperty(const std::string& name) = 0;
+
+ // topic
+ virtual const std::string& getTopic() const = 0;
+ virtual void setTopic(const std::string& topic) = 0;
+ virtual void setTopic(const char* body, int len) = 0;
+
+ // tags
+ virtual const std::string& getTags() const = 0;
+ virtual void setTags(const std::string& tags) = 0;
+
+ // keys
+ virtual const std::string& getKeys() const = 0;
+ virtual void setKeys(const std::string& keys) = 0;
+ virtual void setKeys(const std::vector<std::string>& keys) = 0;
+
+ // delay time level
+ virtual int getDelayTimeLevel() const = 0;
+ virtual void setDelayTimeLevel(int level) = 0;
+
+ // wait store message ok
+ virtual bool isWaitStoreMsgOK() const = 0;
+ virtual void setWaitStoreMsgOK(bool waitStoreMsgOK) = 0;
+
+ // flag
+ virtual int32_t getFlag() const = 0;
+ virtual void setFlag(int32_t flag) = 0;
+
+ // body
+ virtual const std::string& getBody() const = 0;
+ virtual void setBody(const char* body, int len) = 0;
+ virtual void setBody(const std::string& body) = 0;
+ virtual void setBody(std::string&& body) = 0;
+
+ // transaction id
+ virtual const std::string& getTransactionId() const = 0;
+ virtual void setTransactionId(const std::string& transactionId) = 0;
+
+ // properties
+ virtual const std::map<std::string, std::string>& getProperties() const = 0;
+ virtual void setProperties(const std::map<std::string, std::string>& properties) = 0;
+ virtual void setProperties(std::map<std::string, std::string>&& properties) = 0;
+
+ // batch flag
+ virtual bool isBatch() const { return false; }
+
+ virtual std::string toString() const = 0;
+};
+
+} // namespace rocketmq
+
+#endif // ROCKETMQ_MESSAGE_H_
diff --git a/include/MessageExt.h b/include/MessageExt.h
new file mode 100644
index 0000000..172aa92
--- /dev/null
+++ b/include/MessageExt.h
@@ -0,0 +1,88 @@
+/*
+ * 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.
+ */
+#ifndef ROCKETMQ_MESSAGEEXT_H_
+#define ROCKETMQ_MESSAGEEXT_H_
+
+#ifdef WIN32
+// clang-format off
+#include <Winsock2.h>
+#include <Windows.h>
+// clang-format on
+#else
+#include <sys/socket.h>
+#endif
+
+#include "Message.h"
+
+namespace rocketmq {
+
+class MessageExt;
+typedef std::shared_ptr<MessageExt> MessageExtPtr;
+
+/**
+ * MessageExt - Message extend class, which was generated on broker
+ */
+class ROCKETMQCLIENT_API MessageExt : virtual public Message // base interface
+{
+ public:
+ virtual ~MessageExt() = default;
+
+ virtual int32_t getStoreSize() const = 0;
+ virtual void setStoreSize(int32_t storeSize) = 0;
+
+ virtual int32_t getBodyCRC() const = 0;
+ virtual void setBodyCRC(int32_t bodyCRC) = 0;
+
+ virtual int32_t getQueueId() const = 0;
+ virtual void setQueueId(int32_t queueId) = 0;
+
+ virtual int64_t getQueueOffset() const = 0;
+ virtual void setQueueOffset(int64_t queueOffset) = 0;
+
+ virtual int64_t getCommitLogOffset() const = 0;
+ virtual void setCommitLogOffset(int64_t physicOffset) = 0;
+
+ virtual int32_t getSysFlag() const = 0;
+ virtual void setSysFlag(int32_t sysFlag) = 0;
+
+ virtual int64_t getBornTimestamp() const = 0;
+ virtual void setBornTimestamp(int64_t bornTimestamp) = 0;
+
+ virtual std::string getBornHostString() const = 0;
+ virtual const struct sockaddr* getBornHost() const = 0;
+ virtual void setBornHost(const struct sockaddr* bornHost) = 0;
+
+ virtual int64_t getStoreTimestamp() const = 0;
+ virtual void setStoreTimestamp(int64_t storeTimestamp) = 0;
+
+ virtual std::string getStoreHostString() const = 0;
+ virtual const struct sockaddr* getStoreHost() const = 0;
+ virtual void setStoreHost(const struct sockaddr* storeHost) = 0;
+
+ virtual int32_t getReconsumeTimes() const = 0;
+ virtual void setReconsumeTimes(int32_t reconsumeTimes) = 0;
+
+ virtual int64_t getPreparedTransactionOffset() const = 0;
+ virtual void setPreparedTransactionOffset(int64_t preparedTransactionOffset) = 0;
+
+ virtual const std::string& getMsgId() const = 0;
+ virtual void setMsgId(const std::string& msgId) = 0;
+};
+
+} // namespace rocketmq
+
+#endif // ROCKETMQ_MESSAGEEXT_H_
diff --git a/include/MessageUtil.h b/include/MessageUtil.h
index ad6808a..d44fd8f 100644
--- a/include/MessageUtil.h
+++ b/include/MessageUtil.h
@@ -14,10 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef __MESSAGE_UTIL_H__
-#define __MESSAGE_UTIL_H__
-
-#include <string>
+#ifndef ROCKETMQ_MESSAGEUTIL_H_
+#define ROCKETMQ_MESSAGEUTIL_H_
#include "MQClientException.h"
#include "MQMessage.h"
@@ -26,12 +24,11 @@ namespace rocketmq {
class ROCKETMQCLIENT_API MessageUtil {
public:
- static MQMessagePtr createReplyMessage(const MQMessagePtr requestMessage,
- const std::string& body) throw(MQClientException);
+ static MQMessage createReplyMessage(const Message& requestMessage, const std::string& body) throw(MQClientException);
- static const std::string& getReplyToClient(const MQMessagePtr msg);
+ static const std::string& getReplyToClient(const Message& msg);
};
} // namespace rocketmq
-#endif // __MESSAGE_UTIL_H__
+#endif // ROCKETMQ_MESSAGEUTIL_H_
diff --git a/include/QueryResult.h b/include/QueryResult.h
index 958a499..1939fd3 100644
--- a/include/QueryResult.h
+++ b/include/QueryResult.h
@@ -23,18 +23,18 @@ namespace rocketmq {
class ROCKETMQCLIENT_API QueryResult {
public:
- QueryResult(uint64_t indexLastUpdateTimestamp, const std::vector<MQMessageExtPtr>& messageList) {
+ QueryResult(uint64_t indexLastUpdateTimestamp, const std::vector<MQMessageExt>& messageList) {
m_indexLastUpdateTimestamp = indexLastUpdateTimestamp;
m_messageList = messageList;
}
uint64_t getIndexLastUpdateTimestamp() { return m_indexLastUpdateTimestamp; }
- std::vector<MQMessageExtPtr>& getMessageList() { return m_messageList; }
+ std::vector<MQMessageExt>& getMessageList() { return m_messageList; }
private:
uint64_t m_indexLastUpdateTimestamp;
- std::vector<MQMessageExtPtr> m_messageList;
+ std::vector<MQMessageExt> m_messageList;
};
} // namespace rocketmq
diff --git a/include/RequestCallback.h b/include/RequestCallback.h
index 8b31098..d79e955 100644
--- a/include/RequestCallback.h
+++ b/include/RequestCallback.h
@@ -28,10 +28,10 @@ class ROCKETMQCLIENT_API RequestCallback {
public:
virtual ~RequestCallback() = default;
- virtual void onSuccess(MQMessagePtr3 message) { onSuccess(message.get()); }
+ virtual void onSuccess(MessagePtr message) { onSuccess(MQMessage(message)); }
// SDK will be responsible for the lifecycle of message.
- virtual void onSuccess(MQMessagePtr message) = 0;
+ virtual void onSuccess(MQMessage message) = 0;
virtual void onException(MQException& e) noexcept = 0;
diff --git a/include/RocketMQClient.h b/include/RocketMQClient.h
index fc5bb6a..d624793 100644
--- a/include/RocketMQClient.h
+++ b/include/RocketMQClient.h
@@ -14,10 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef __ROCKETMQ_CLIENT_H__
-#define __ROCKETMQ_CLIENT_H__
+#ifndef ROCKETMQ_ROCKETMQCLIENT_H_
+#define ROCKETMQ_ROCKETMQCLIENT_H_
+#ifdef __cplusplus
#include <cstdint>
+#else
+#include <stdint.h>
+#endif
#ifdef WIN32
#ifdef ROCKETMQCLIENT_EXPORTS
@@ -49,4 +53,4 @@
#define FILE_SEPARATOR "/"
#endif
-#endif // __ROCKETMQ_CLIENT_H__
+#endif // ROCKETMQ_ROCKETMQCLIENT_H_
diff --git a/include/TopicFilterType.h b/include/TopicFilterType.h
index 9f57a42..6043347 100644
--- a/include/TopicFilterType.h
+++ b/include/TopicFilterType.h
@@ -14,8 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef __TOPIC_FILTER_TYPE_H__
-#define __TOPIC_FILTER_TYPE_H__
+#ifndef ROCKETMQ_TOPICFILTERTYPE_H_
+#define ROCKETMQ_TOPICFILTERTYPE_H_
namespace rocketmq {
@@ -24,6 +24,7 @@ enum TopicFilterType {
* each msg could only have one tag
*/
SINGLE_TAG,
+
/**
* not support now
*/
@@ -32,4 +33,4 @@ enum TopicFilterType {
} // namespace rocketmq
-#endif // __TOPIC_FILTER_TYPE_H__
+#endif // ROCKETMQ_TOPICFILTERTYPE_H_
diff --git a/include/TransactionMQProducer.h b/include/TransactionMQProducer.h
index 0a1380f..4efe26d 100644
--- a/include/TransactionMQProducer.h
+++ b/include/TransactionMQProducer.h
@@ -37,7 +37,7 @@ class ROCKETMQCLIENT_API TransactionMQProducer : public DefaultMQProducer, virtu
void shutdown() override;
// Transaction: don't delete msg object, until callback occur.
- TransactionSendResult sendMessageInTransaction(MQMessagePtr msg, void* arg) override;
+ TransactionSendResult sendMessageInTransaction(MQMessage& msg, void* arg) override;
};
} // namespace rocketmq
diff --git a/src/ClientRemotingProcessor.cpp b/src/ClientRemotingProcessor.cpp
index 4f8d8a4..97286a1 100644
--- a/src/ClientRemotingProcessor.cpp
+++ b/src/ClientRemotingProcessor.cpp
@@ -17,7 +17,7 @@
#include "ClientRemotingProcessor.h"
#include "ConsumerRunningInfo.h"
-#include "MQDecoder.h"
+#include "MessageDecoder.h"
#include "MQProtos.h"
#include "MessageAccessor.h"
#include "MessageSysFlag.h"
@@ -65,7 +65,7 @@ RemotingCommand* ClientRemotingProcessor::checkTransactionState(const std::strin
auto requestBody = request->getBody();
if (requestBody != nullptr && requestBody->getSize() > 0) {
- MQMessageExtPtr2 messageExt = MQDecoder::decode(*requestBody);
+ MessageExtPtr messageExt = MessageDecoder::decode(*requestBody);
if (messageExt != nullptr) {
const auto& transactionId = messageExt->getProperty(MQMessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
if (!transactionId.empty()) {
@@ -173,7 +173,7 @@ RemotingCommand* ClientRemotingProcessor::receiveReplyMessage(RemotingCommand* r
}
msg->setFlag(requestHeader->getFlag());
- MessageAccessor::setProperties(*msg, MQDecoder::string2messageProperties(requestHeader->getProperties()));
+ MessageAccessor::setProperties(*msg, MessageDecoder::string2messageProperties(requestHeader->getProperties()));
MessageAccessor::putProperty(*msg, MQMessageConst::PROPERTY_REPLY_MESSAGE_ARRIVE_TIME,
UtilAll::to_string(receiveTime));
msg->setBornTimestamp(requestHeader->getBornTimestamp());
diff --git a/src/MQAdminImpl.cpp b/src/MQAdminImpl.cpp
index be67fce..5f334a2 100644
--- a/src/MQAdminImpl.cpp
+++ b/src/MQAdminImpl.cpp
@@ -115,9 +115,9 @@ int64_t MQAdminImpl::earliestMsgStoreTime(const MQMessageQueue& mq) {
THROW_MQEXCEPTION(MQClientException, "The broker is not exist", -1);
}
-MQMessageExtPtr MQAdminImpl::viewMessage(const std::string& msgId) {
+MQMessageExt MQAdminImpl::viewMessage(const std::string& msgId) {
try {
- return nullptr;
+ return MQMessageExt(nullptr);
} catch (MQException& e) {
THROW_MQEXCEPTION(MQClientException, "message id illegal", -1);
}
diff --git a/src/MQAdminImpl.h b/src/MQAdminImpl.h
index ee3d18d..27949c7 100644
--- a/src/MQAdminImpl.h
+++ b/src/MQAdminImpl.h
@@ -40,7 +40,7 @@ class MQAdminImpl {
int64_t minOffset(const MQMessageQueue& mq);
int64_t earliestMsgStoreTime(const MQMessageQueue& mq);
- MQMessageExtPtr viewMessage(const std::string& msgId);
+ MQMessageExt viewMessage(const std::string& msgId);
QueryResult queryMessage(const std::string& topic, const std::string& key, int maxNum, int64_t begin, int64_t end);
private:
diff --git a/src/MQClientAPIImpl.cpp b/src/MQClientAPIImpl.cpp
index 23fea6a..1dcc163 100644
--- a/src/MQClientAPIImpl.cpp
+++ b/src/MQClientAPIImpl.cpp
@@ -87,7 +87,7 @@ void MQClientAPIImpl::createTopic(const std::string& addr, const std::string& de
SendResult* MQClientAPIImpl::sendMessage(const std::string& addr,
const std::string& brokerName,
- const MQMessagePtr msg,
+ const MessagePtr msg,
std::unique_ptr<SendMessageRequestHeader> requestHeader,
int timeoutMillis,
CommunicationMode communicationMode,
@@ -98,7 +98,7 @@ SendResult* MQClientAPIImpl::sendMessage(const std::string& addr,
SendResult* MQClientAPIImpl::sendMessage(const std::string& addr,
const std::string& brokerName,
- const MQMessagePtr msg,
+ const MessagePtr msg,
std::unique_ptr<SendMessageRequestHeader> requestHeader,
int timeoutMillis,
CommunicationMode communicationMode,
@@ -149,7 +149,7 @@ SendResult* MQClientAPIImpl::sendMessage(const std::string& addr,
void MQClientAPIImpl::sendMessageAsync(const std::string& addr,
const std::string& brokerName,
- const MQMessagePtr msg,
+ const MessagePtr msg,
RemotingCommand&& request,
SendCallback* sendCallback,
TopicPublishInfoPtr topicPublishInfo,
@@ -177,7 +177,7 @@ void MQClientAPIImpl::sendMessageAsyncImpl(SendCallbackWrap* cbw, int64_t timeou
SendResult* MQClientAPIImpl::sendMessageSync(const std::string& addr,
const std::string& brokerName,
- const MQMessagePtr msg,
+ const MessagePtr msg,
RemotingCommand& request,
int timeoutMillis) {
// block until response
@@ -187,7 +187,7 @@ SendResult* MQClientAPIImpl::sendMessageSync(const std::string& addr,
}
SendResult* MQClientAPIImpl::processSendResponse(const std::string& brokerName,
- const MQMessagePtr msg,
+ const MessagePtr msg,
RemotingCommand* response) {
SendStatus sendStatus = SEND_OK;
switch (response->getCode()) {
@@ -217,11 +217,11 @@ SendResult* MQClientAPIImpl::processSendResponse(const std::string& brokerName,
// MessageBatch
if (msg->isBatch()) {
- const auto& messages = static_cast<const MessageBatch*>(msg)->getMessages();
+ const auto& messages = std::dynamic_pointer_cast<MessageBatch>(msg)->getMessages();
uniqMsgId.clear();
uniqMsgId.reserve(33 * messages.size() + 1);
for (const auto& message : messages) {
- uniqMsgId.append(MessageClientIDSetter::getUniqID(*message));
+ uniqMsgId.append(MessageClientIDSetter::getUniqID(message));
uniqMsgId.append(",");
}
if (!uniqMsgId.empty()) {
@@ -307,7 +307,7 @@ PullResult* MQClientAPIImpl::processPullResponse(RemotingCommand* response) {
responseHeader->maxOffset, (int)responseHeader->suggestWhichBrokerId, response->getBody());
}
-MQMessageExtPtr MQClientAPIImpl::viewMessage(const std::string& addr, int64_t phyoffset, int timeoutMillis) {
+MQMessageExt MQClientAPIImpl::viewMessage(const std::string& addr, int64_t phyoffset, int timeoutMillis) {
auto* requestHeader = new ViewMessageRequestHeader();
requestHeader->offset = phyoffset;
@@ -552,17 +552,17 @@ void MQClientAPIImpl::endTransactionOneway(const std::string& addr,
}
void MQClientAPIImpl::consumerSendMessageBack(const std::string& addr,
- MQMessageExt& msg,
+ MessageExtPtr msg,
const std::string& consumerGroup,
int delayLevel,
int timeoutMillis,
int maxConsumeRetryTimes) {
auto* requestHeader = new ConsumerSendMsgBackRequestHeader();
requestHeader->group = consumerGroup;
- requestHeader->originTopic = msg.getTopic();
- requestHeader->offset = msg.getCommitLogOffset();
+ requestHeader->originTopic = msg->getTopic();
+ requestHeader->offset = msg->getCommitLogOffset();
requestHeader->delayLevel = delayLevel;
- requestHeader->originMsgId = msg.getMsgId();
+ requestHeader->originMsgId = msg->getMsgId();
requestHeader->maxReconsumeTimes = maxConsumeRetryTimes;
RemotingCommand request(CONSUMER_SEND_MSG_BACK, requestHeader);
diff --git a/src/MQClientAPIImpl.h b/src/MQClientAPIImpl.h
index b2f94fa..dcf8da9 100644
--- a/src/MQClientAPIImpl.h
+++ b/src/MQClientAPIImpl.h
@@ -60,14 +60,14 @@ class MQClientAPIImpl {
SendResult* sendMessage(const std::string& addr,
const std::string& brokerName,
- const MQMessagePtr msg,
+ const MessagePtr msg,
std::unique_ptr<SendMessageRequestHeader> requestHeader,
int timeoutMillis,
CommunicationMode communicationMode,
DefaultMQProducerImplPtr producer);
SendResult* sendMessage(const std::string& addr,
const std::string& brokerName,
- const MQMessagePtr msg,
+ const MessagePtr msg,
std::unique_ptr<SendMessageRequestHeader> requestHeader,
int timeoutMillis,
CommunicationMode communicationMode,
@@ -76,7 +76,7 @@ class MQClientAPIImpl {
MQClientInstancePtr instance,
int retryTimesWhenSendFailed,
DefaultMQProducerImplPtr producer);
- SendResult* processSendResponse(const std::string& brokerName, const MQMessagePtr msg, RemotingCommand* pResponse);
+ SendResult* processSendResponse(const std::string& brokerName, const MessagePtr msg, RemotingCommand* pResponse);
PullResult* pullMessage(const std::string& addr,
PullMessageRequestHeader* requestHeader,
@@ -85,7 +85,7 @@ class MQClientAPIImpl {
PullCallback* pullCallback);
PullResult* processPullResponse(RemotingCommand* pResponse);
- MQMessageExtPtr viewMessage(const std::string& addr, int64_t phyoffset, int timeoutMillis);
+ MQMessageExt viewMessage(const std::string& addr, int64_t phyoffset, int timeoutMillis);
int64_t searchOffset(const std::string& addr,
const std::string& topic,
@@ -125,7 +125,7 @@ class MQClientAPIImpl {
const std::string& remark);
void consumerSendMessageBack(const std::string& addr,
- MQMessageExt& msg,
+ MessageExtPtr msg,
const std::string& consumerGroup,
int delayLevel,
int timeoutMillis,
@@ -166,13 +166,13 @@ class MQClientAPIImpl {
SendResult* sendMessageSync(const std::string& addr,
const std::string& brokerName,
- const MQMessagePtr msg,
+ const MessagePtr msg,
RemotingCommand& request,
int timeoutMillis);
void sendMessageAsync(const std::string& addr,
const std::string& brokerName,
- const MQMessagePtr msg,
+ const MessagePtr msg,
RemotingCommand&& request,
SendCallback* sendCallback,
TopicPublishInfoPtr topicPublishInfo,
diff --git a/src/MQClientImpl.cpp b/src/MQClientImpl.cpp
index 58376b5..83b110c 100644
--- a/src/MQClientImpl.cpp
+++ b/src/MQClientImpl.cpp
@@ -71,7 +71,7 @@ int64_t MQClientImpl::earliestMsgStoreTime(const MQMessageQueue& mq) {
return m_clientInstance->getMQAdminImpl()->earliestMsgStoreTime(mq);
}
-MQMessageExtPtr MQClientImpl::viewMessage(const std::string& msgId) {
+MQMessageExt MQClientImpl::viewMessage(const std::string& msgId) {
return m_clientInstance->getMQAdminImpl()->viewMessage(msgId);
}
diff --git a/src/MQClientImpl.h b/src/MQClientImpl.h
index 3b689bd..0807818 100644
--- a/src/MQClientImpl.h
+++ b/src/MQClientImpl.h
@@ -35,7 +35,7 @@ class MQClientImpl : public MQAdmin {
int64_t maxOffset(const MQMessageQueue& mq) override;
int64_t minOffset(const MQMessageQueue& mq) override;
int64_t earliestMsgStoreTime(const MQMessageQueue& mq) override;
- MQMessageExtPtr viewMessage(const std::string& offsetMsgId) override;
+ MQMessageExt viewMessage(const std::string& offsetMsgId) override;
QueryResult queryMessage(const std::string& topic,
const std::string& key,
int maxNum,
diff --git a/src/common/ByteOrder.h b/src/common/ByteOrder.h
index 1d4d7a2..cdc52a9 100644
--- a/src/common/ByteOrder.h
+++ b/src/common/ByteOrder.h
@@ -14,8 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef __BYTE_ORDER_H__
-#define __BYTE_ORDER_H__
+#ifndef ROCKETMQ_COMMON_BYTEORDER_H_
+#define ROCKETMQ_COMMON_BYTEORDER_H_
#include <RocketMQClient.h>
@@ -24,9 +24,10 @@ namespace rocketmq {
/**
* Contains static methods for converting the byte order between different endiannesses.
*/
-class ROCKETMQCLIENT_API ByteOrder {
+class ByteOrder {
public:
//==============================================================================
+
/** Swaps the upper and lower bytes of a 16-bit integer. */
static uint16_t swap(uint16_t value);
@@ -37,6 +38,7 @@ class ROCKETMQCLIENT_API ByteOrder {
static uint64_t swap(uint64_t value);
//==============================================================================
+
/** Swaps the byte order of a 16-bit int if the CPU is big-endian */
static uint16_t swapIfBigEndian(uint16_t value);
@@ -56,6 +58,7 @@ class ROCKETMQCLIENT_API ByteOrder {
static uint64_t swapIfLittleEndian(uint64_t value);
//==============================================================================
+
/** Turns 4 bytes into a little-endian integer. */
static uint32_t littleEndianInt(const void* bytes);
@@ -75,6 +78,7 @@ class ROCKETMQCLIENT_API ByteOrder {
static uint16_t bigEndianShort(const void* bytes);
//==============================================================================
+
/** Converts 3 little-endian bytes into a signed 24-bit value (which is
* sign-extended to 32 bits). */
static int littleEndian24Bit(const void* bytes);
@@ -90,6 +94,7 @@ class ROCKETMQCLIENT_API ByteOrder {
static void bigEndian24BitToChars(int value, void* destBytes);
//==============================================================================
+
/** Returns true if the current CPU is big-endian. */
static bool isBigEndian();
};
@@ -109,6 +114,7 @@ inline uint64_t ByteOrder::swap(uint64_t value) {
}
#if __BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__ //__BYTE_ORDER__ is defined by GCC
+
inline uint16_t ByteOrder::swapIfBigEndian(const uint16_t v) {
return v;
}
@@ -160,7 +166,9 @@ inline uint16_t ByteOrder::bigEndianShort(const void* const bytes) {
inline bool ByteOrder::isBigEndian() {
return false;
}
-#else
+
+#else // __BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__
+
inline uint16_t ByteOrder::swapIfBigEndian(const uint16_t v) {
return swap(v);
}
@@ -212,7 +220,8 @@ inline uint16_t ByteOrder::bigEndianShort(const void* const bytes) {
inline bool ByteOrder::isBigEndian() {
return true;
}
-#endif
+
+#endif // __BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__
inline int ByteOrder::littleEndian24Bit(const void* const bytes) {
return (((int)static_cast<const int8_t*>(bytes)[2]) << 16) | (((int)static_cast<const uint8_t*>(bytes)[1]) << 8) |
@@ -238,4 +247,4 @@ inline void ByteOrder::bigEndian24BitToChars(const int value, void* const destBy
} // namespace rocketmq
-#endif // __BYTE_ORDER_H__
+#endif // ROCKETMQ_COMMON_BYTEORDER_H_
diff --git a/src/common/MessageSysFlag.cpp b/src/common/MessageSysFlag.cpp
index 6c980b0..a61d36c 100644
--- a/src/common/MessageSysFlag.cpp
+++ b/src/common/MessageSysFlag.cpp
@@ -26,6 +26,9 @@ const int MessageSysFlag::TransactionPreparedType = 0x1 << 2;
const int MessageSysFlag::TransactionCommitType = 0x2 << 2;
const int MessageSysFlag::TransactionRollbackType = 0x3 << 2;
+const int MessageSysFlag::BronhostV6Flag = 0x1 << 4;
+const int MessageSysFlag::StorehostV6Flag = 0x1 << 5;
+
int MessageSysFlag::getTransactionValue(int flag) {
return flag & TransactionRollbackType;
}
diff --git a/src/common/MessageSysFlag.h b/src/common/MessageSysFlag.h
index a5e8852..8c61579 100644
--- a/src/common/MessageSysFlag.h
+++ b/src/common/MessageSysFlag.h
@@ -34,6 +34,9 @@ class MessageSysFlag {
static const int TransactionPreparedType;
static const int TransactionCommitType;
static const int TransactionRollbackType;
+
+ static const int BronhostV6Flag;
+ static const int StorehostV6Flag;
};
} // namespace rocketmq
diff --git a/src/common/SendCallbackWrap.cpp b/src/common/SendCallbackWrap.cpp
index 4df8883..698fe56 100755
--- a/src/common/SendCallbackWrap.cpp
+++ b/src/common/SendCallbackWrap.cpp
@@ -22,7 +22,7 @@
#include "Logging.h"
#include "MQClientAPIImpl.h"
#include "MQClientInstance.h"
-#include "MQDecoder.h"
+#include "MessageDecoder.h"
#include "MQMessageQueue.h"
#include "MQProtos.h"
#include "PullAPIWrapper.h"
@@ -34,7 +34,7 @@ namespace rocketmq {
SendCallbackWrap::SendCallbackWrap(const std::string& addr,
const std::string& brokerName,
- const MQMessagePtr msg,
+ const MessagePtr msg,
RemotingCommand&& request,
SendCallback* sendCallback,
TopicPublishInfoPtr topicPublishInfo,
diff --git a/src/common/SendCallbackWrap.h b/src/common/SendCallbackWrap.h
index a969f75..2ef8ffa 100755
--- a/src/common/SendCallbackWrap.h
+++ b/src/common/SendCallbackWrap.h
@@ -22,7 +22,7 @@
#include "DefaultMQProducerImpl.h"
#include "InvokeCallback.h"
#include "MQClientInstance.h"
-#include "MQMessage.h"
+#include "Message.h"
#include "RemotingCommand.h"
#include "ResponseFuture.h"
#include "SendCallback.h"
@@ -34,7 +34,7 @@ class SendCallbackWrap : public InvokeCallback {
public:
SendCallbackWrap(const std::string& addr,
const std::string& brokerName,
- const MQMessagePtr msg,
+ const MessagePtr msg,
RemotingCommand&& request,
SendCallback* sendCallback,
TopicPublishInfoPtr topicPublishInfo,
@@ -47,7 +47,7 @@ class SendCallbackWrap : public InvokeCallback {
void onExceptionImpl(ResponseFuture* responseFuture, long timeoutMillis, MQException& e, bool needRetry);
const std::string& getAddr() { return m_addr; }
- const MQMessagePtr getMessage() { return m_msg; }
+ const MessagePtr getMessage() { return m_msg; }
RemotingCommand& getRemotingCommand() { return m_request; }
void setRetrySendTimes(int retrySendTimes) { m_times = retrySendTimes; }
@@ -57,7 +57,7 @@ class SendCallbackWrap : public InvokeCallback {
private:
std::string m_addr;
std::string m_brokerName;
- const MQMessagePtr m_msg;
+ const MessagePtr m_msg;
RemotingCommand m_request;
SendCallback* m_sendCallback;
TopicPublishInfoPtr m_topicPublishInfo;
diff --git a/src/common/UtilAll.cpp b/src/common/UtilAll.cpp
index 4333ecd..7f2dbb2 100644
--- a/src/common/UtilAll.cpp
+++ b/src/common/UtilAll.cpp
@@ -68,44 +68,54 @@ int32_t UtilAll::HashCode(const std::string& str) {
return h;
}
-static const int hex2int[256] = {
- -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
- -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9,
- -1, -1, -1, -1, -1, -1, -1, 10, 11, 12, 13, 14, 15, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
- -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, 10, 11, 12, 13, 14, 15, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
- -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
- -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
- -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
- -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
- -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1};
-
-uint64_t UtilAll::hexstr2ull(const char* str) {
- uint64_t num = 0;
- unsigned char* ch = (unsigned char*)str;
- while (*ch != '\0') {
- num = (num << 4) + hex2int[*ch];
- ch++;
- }
- return num;
-}
-
-static const char sHexAlphabet[] = "0123456789ABCDEF";
+static const char kHexAlphabet[] = "0123456789ABCDEF";
std::string UtilAll::bytes2string(const char* bytes, size_t len) {
if (bytes == nullptr || len <= 0) {
- return std::string();
+ return null;
}
std::string buffer;
buffer.reserve(len * 2 + 1);
for (std::size_t i = 0; i < len; i++) {
- unsigned char v = (unsigned char)bytes[i];
- buffer.append(1, sHexAlphabet[v >> 4]);
- buffer.append(1, sHexAlphabet[v & 0x0FU]);
+ uint8_t v = (uint8_t)bytes[i];
+ buffer.append(1, kHexAlphabet[v >> 4]);
+ buffer.append(1, kHexAlphabet[v & 0x0FU]);
}
return buffer;
}
+// clang-format off
+static const uint8_t kHexIndex[256] = {
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 0, 0, 0, 0, 0,
+ 0,10,11,12,13,14,15, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0,10,11,12,13,14,15, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0
+};
+// clang-format on
+
+void UtilAll::string2bytes(char* dest, const std::string& src) {
+ if (dest == nullptr || src.empty()) {
+ return;
+ }
+ for (size_t i = 0; i < src.size() / 2; i++) {
+ size_t pos = i * 2;
+ dest[i] = kHexIndex[(uint8_t)src[pos]] << 4 | kHexIndex[(uint8_t)src[pos + 1]];
+ }
+}
+
bool UtilAll::isRetryTopic(const std::string& topic) {
return topic.find(RETRY_GROUP_TOPIC_PREFIX) == 0;
}
diff --git a/src/common/UtilAll.h b/src/common/UtilAll.h
index 2e2c34f..a919db7 100644
--- a/src/common/UtilAll.h
+++ b/src/common/UtilAll.h
@@ -93,9 +93,8 @@ class UtilAll {
static int32_t HashCode(const std::string& str);
- static uint64_t hexstr2ull(const char* str);
-
static std::string bytes2string(const char* bytes, size_t len);
+ static void string2bytes(char* dest, const std::string& src);
static bool isRetryTopic(const std::string& topic);
static std::string getRetryTopic(const std::string& consumerGroup);
diff --git a/src/common/Validators.cpp b/src/common/Validators.cpp
index c293aa4..4cd843e 100644
--- a/src/common/Validators.cpp
+++ b/src/common/Validators.cpp
@@ -95,7 +95,7 @@ void Validators::checkGroup(const std::string& group) {
}
}
-void Validators::checkMessage(const MQMessage& msg, int maxMessageSize) {
+void Validators::checkMessage(const Message& msg, int maxMessageSize) {
checkTopic(msg.getTopic());
const auto& body = msg.getBody();
diff --git a/src/common/Validators.h b/src/common/Validators.h
index e5a6093..3ffe527 100644
--- a/src/common/Validators.h
+++ b/src/common/Validators.h
@@ -20,7 +20,7 @@
#include <string>
#include "MQClientException.h"
-#include "MQMessage.h"
+#include "Message.h"
#include "UtilAll.h"
namespace rocketmq {
@@ -31,7 +31,7 @@ class Validators {
static std::string getGroupWithRegularExpression(const std::string& origin, const std::string& patternStr);
static void checkTopic(const std::string& topic);
static void checkGroup(const std::string& group);
- static void checkMessage(const MQMessage& msg, int maxMessageSize);
+ static void checkMessage(const Message& msg, int maxMessageSize);
public:
static const std::string validPatternStr;
diff --git a/src/consumer/ConsumeMessageConcurrentlyService.cpp b/src/consumer/ConsumeMessageConcurrentlyService.cpp
index 26063e5..42cde12 100755
--- a/src/consumer/ConsumeMessageConcurrentlyService.cpp
+++ b/src/consumer/ConsumeMessageConcurrentlyService.cpp
@@ -43,7 +43,7 @@ void ConsumeMessageConcurrentlyService::shutdown() {
m_consumeExecutor.shutdown();
}
-void ConsumeMessageConcurrentlyService::submitConsumeRequest(std::vector<MQMessageExtPtr2>& msgs,
+void ConsumeMessageConcurrentlyService::submitConsumeRequest(std::vector<MessageExtPtr>& msgs,
ProcessQueuePtr processQueue,
const MQMessageQueue& messageQueue,
const bool dispathToConsume) {
@@ -51,7 +51,7 @@ void ConsumeMessageConcurrentlyService::submitConsumeRequest(std::vector<MQMessa
std::bind(&ConsumeMessageConcurrentlyService::ConsumeRequest, this, msgs, processQueue, messageQueue));
}
-void ConsumeMessageConcurrentlyService::submitConsumeRequestLater(std::vector<MQMessageExtPtr2>& msgs,
+void ConsumeMessageConcurrentlyService::submitConsumeRequestLater(std::vector<MessageExtPtr>& msgs,
ProcessQueuePtr processQueue,
const MQMessageQueue& messageQueue) {
m_scheduledExecutorService.schedule(
@@ -59,7 +59,7 @@ void ConsumeMessageConcurrentlyService::submitConsumeRequestLater(std::vector<MQ
5000L, time_unit::milliseconds);
}
-void ConsumeMessageConcurrentlyService::ConsumeRequest(std::vector<MQMessageExtPtr2>& msgs,
+void ConsumeMessageConcurrentlyService::ConsumeRequest(std::vector<MessageExtPtr>& msgs,
ProcessQueuePtr processQueue,
const MQMessageQueue& messageQueue) {
if (processQueue->isDropped()) {
@@ -83,11 +83,16 @@ void ConsumeMessageConcurrentlyService::ConsumeRequest(std::vector<MQMessageExtP
processQueue->setLastConsumeTimestamp(consumeTimestamp);
if (!msgs.empty()) {
auto timestamp = UtilAll::to_string(consumeTimestamp);
- for (auto& msg : msgs) {
+ for (const auto& msg : msgs) {
MessageAccessor::setConsumeStartTimeStamp(*msg, timestamp);
}
}
- status = m_messageListener->consumeMessage(msgs);
+ std::vector<MQMessageExt> message_list;
+ message_list.reserve(msgs.size());
+ for (const auto& msg : msgs) {
+ message_list.emplace_back(msg);
+ }
+ status = m_messageListener->consumeMessage(message_list);
} catch (const std::exception& e) {
LOG_WARN_NEW("encounter unexpected exception when consume messages.\n{}", e.what());
}
@@ -122,13 +127,13 @@ void ConsumeMessageConcurrentlyService::ConsumeRequest(std::vector<MQMessageExtP
break;
case CLUSTERING: {
// send back msg to broker
- std::vector<MQMessageExtPtr2> msgBackFailed;
+ std::vector<MessageExtPtr> msgBackFailed;
int idx = ackIndex + 1;
for (auto iter = msgs.begin() + idx; iter != msgs.end(); idx++) {
LOG_WARN_NEW("consume fail, MQ is:{}, its msgId is:{}, index is:{}, reconsume times is:{}",
messageQueue.toString(), (*iter)->getMsgId(), idx, (*iter)->getReconsumeTimes());
auto& msg = (*iter);
- bool result = m_consumer->sendMessageBack(*msg, 0, messageQueue.getBrokerName());
+ bool result = m_consumer->sendMessageBack(msg, 0, messageQueue.getBrokerName());
if (!result) {
msg->setReconsumeTimes(msg->getReconsumeTimes() + 1);
msgBackFailed.push_back(msg);
diff --git a/src/consumer/ConsumeMessageOrderlyService.cpp b/src/consumer/ConsumeMessageOrderlyService.cpp
index 9411666..ec6c3ff 100755
--- a/src/consumer/ConsumeMessageOrderlyService.cpp
+++ b/src/consumer/ConsumeMessageOrderlyService.cpp
@@ -67,7 +67,7 @@ bool ConsumeMessageOrderlyService::lockOneMQ(const MQMessageQueue& mq) {
return m_consumer->getRebalanceImpl()->lock(mq);
}
-void ConsumeMessageOrderlyService::submitConsumeRequest(std::vector<MQMessageExtPtr2>& msgs,
+void ConsumeMessageOrderlyService::submitConsumeRequest(std::vector<MessageExtPtr>& msgs,
ProcessQueuePtr processQueue,
const MQMessageQueue& messageQueue,
const bool dispathToConsume) {
@@ -87,7 +87,7 @@ void ConsumeMessageOrderlyService::submitConsumeRequestLater(ProcessQueuePtr pro
timeMillis = std::max(10L, std::min(timeMillis, 30000L));
- static std::vector<MQMessageExtPtr2> dummy;
+ static std::vector<MessageExtPtr> dummy;
m_scheduledExecutorService.schedule(std::bind(&ConsumeMessageOrderlyService::submitConsumeRequest, this,
std::ref(dummy), processQueue, messageQueue, true),
timeMillis, time_unit::milliseconds);
@@ -145,7 +145,7 @@ void ConsumeMessageOrderlyService::ConsumeRequest(ProcessQueuePtr processQueue,
const int consumeBatchSize = m_consumer->getDefaultMQPushConsumerConfig()->getConsumeMessageBatchMaxSize();
- std::vector<MQMessageExtPtr2> msgs;
+ std::vector<MessageExtPtr> msgs;
processQueue->takeMessages(msgs, consumeBatchSize);
m_consumer->resetRetryTopic(msgs, m_consumer->getDefaultMQPushConsumerConfig()->getGroupName());
if (!msgs.empty()) {
@@ -157,8 +157,12 @@ void ConsumeMessageOrderlyService::ConsumeRequest(ProcessQueuePtr processQueue,
messageQueue.toString());
break;
}
-
- status = m_messageListener->consumeMessage(msgs);
+ std::vector<MQMessageExt> message_list;
+ message_list.reserve(msgs.size());
+ for (const auto& msg : msgs) {
+ message_list.emplace_back(msg);
+ }
+ status = m_messageListener->consumeMessage(message_list);
} catch (const std::exception& e) {
LOG_WARN_NEW("encounter unexpected exception when consume messages.\n{}", e.what());
}
diff --git a/src/consumer/ConsumeMsgService.h b/src/consumer/ConsumeMsgService.h
index c854d76..b8e1d8a 100755
--- a/src/consumer/ConsumeMsgService.h
+++ b/src/consumer/ConsumeMsgService.h
@@ -33,7 +33,7 @@ class ConsumeMsgService {
virtual void start() {}
virtual void shutdown() {}
- virtual void submitConsumeRequest(std::vector<MQMessageExtPtr2>& msgs,
+ virtual void submitConsumeRequest(std::vector<MessageExtPtr>& msgs,
ProcessQueuePtr processQueue,
const MQMessageQueue& messageQueue,
const bool dispathToConsume) = 0;
@@ -47,17 +47,17 @@ class ConsumeMessageConcurrentlyService : public ConsumeMsgService {
void start() override;
void shutdown() override;
- void submitConsumeRequest(std::vector<MQMessageExtPtr2>& msgs,
+ void submitConsumeRequest(std::vector<MessageExtPtr>& msgs,
ProcessQueuePtr processQueue,
const MQMessageQueue& messageQueue,
const bool dispathToConsume) override;
- void ConsumeRequest(std::vector<MQMessageExtPtr2>& msgs,
+ void ConsumeRequest(std::vector<MessageExtPtr>& msgs,
ProcessQueuePtr processQueue,
const MQMessageQueue& messageQueue);
private:
- void submitConsumeRequestLater(std::vector<MQMessageExtPtr2>& msgs,
+ void submitConsumeRequestLater(std::vector<MessageExtPtr>& msgs,
ProcessQueuePtr processQueue,
const MQMessageQueue& messageQueue);
@@ -78,7 +78,7 @@ class ConsumeMessageOrderlyService : public ConsumeMsgService {
void shutdown() override;
void stopThreadPool();
- void submitConsumeRequest(std::vector<MQMessageExtPtr2>& msgs,
+ void submitConsumeRequest(std::vector<MessageExtPtr>& msgs,
ProcessQueuePtr processQueue,
const MQMessageQueue& messageQueue,
const bool dispathToConsume) override;
diff --git a/src/consumer/DefaultMQPullConsumer.cpp b/src/consumer/DefaultMQPullConsumer.cpp
index 4512a38..66d324e 100644
--- a/src/consumer/DefaultMQPullConsumer.cpp
+++ b/src/consumer/DefaultMQPullConsumer.cpp
@@ -45,11 +45,11 @@ void DefaultMQPullConsumer::shutdown() {
m_pullConsumerDelegate->shutdown();
}
-bool DefaultMQPullConsumer::sendMessageBack(MQMessageExt& msg, int delayLevel) {
+bool DefaultMQPullConsumer::sendMessageBack(MessageExtPtr msg, int delayLevel) {
return m_pullConsumerDelegate->sendMessageBack(msg, delayLevel);
}
-bool DefaultMQPullConsumer::sendMessageBack(MQMessageExt& msg, int delayLevel, const std::string& brokerName) {
+bool DefaultMQPullConsumer::sendMessageBack(MessageExtPtr msg, int delayLevel, const std::string& brokerName) {
return m_pullConsumerDelegate->sendMessageBack(msg, delayLevel, brokerName);
}
diff --git a/src/consumer/DefaultMQPushConsumer.cpp b/src/consumer/DefaultMQPushConsumer.cpp
index 0d717fd..4ef9b36 100644
--- a/src/consumer/DefaultMQPushConsumer.cpp
+++ b/src/consumer/DefaultMQPushConsumer.cpp
@@ -48,11 +48,11 @@ void DefaultMQPushConsumer::shutdown() {
m_pushConsumerDelegate->shutdown();
}
-bool DefaultMQPushConsumer::sendMessageBack(MQMessageExt& msg, int delayLevel) {
+bool DefaultMQPushConsumer::sendMessageBack(MessageExtPtr msg, int delayLevel) {
return m_pushConsumerDelegate->sendMessageBack(msg, delayLevel);
}
-bool DefaultMQPushConsumer::sendMessageBack(MQMessageExt& msg, int delayLevel, const std::string& brokerName) {
+bool DefaultMQPushConsumer::sendMessageBack(MessageExtPtr msg, int delayLevel, const std::string& brokerName) {
return m_pushConsumerDelegate->sendMessageBack(msg, delayLevel, brokerName);
}
diff --git a/src/consumer/DefaultMQPushConsumerImpl.cpp b/src/consumer/DefaultMQPushConsumerImpl.cpp
index 40630f3..800b4c7 100644
--- a/src/consumer/DefaultMQPushConsumerImpl.cpp
+++ b/src/consumer/DefaultMQPushConsumerImpl.cpp
@@ -63,39 +63,39 @@ class AsyncPullCallback : public AutoDeletePullCallback {
PullResult result = defaultMQPushConsumer->getPullAPIWrapper()->processPullResult(m_pullRequest->getMessageQueue(),
pullResult, m_subscriptionData);
- switch (result.pullStatus) {
+ switch (result.pull_status()) {
case FOUND: {
int64_t prevRequestOffset = m_pullRequest->getNextOffset();
- m_pullRequest->setNextOffset(result.nextBeginOffset);
+ m_pullRequest->setNextOffset(result.next_begin_offset());
int64_t firstMsgOffset = (std::numeric_limits<int64_t>::max)();
- if (result.msgFoundList.empty()) {
+ if (result.msg_found_list().empty()) {
defaultMQPushConsumer->executePullRequestImmediately(m_pullRequest);
} else {
- firstMsgOffset = (*result.msgFoundList.begin())->getQueueOffset();
+ firstMsgOffset = result.msg_found_list()[0]->getQueueOffset();
- m_pullRequest->getProcessQueue()->putMessage(result.msgFoundList);
+ m_pullRequest->getProcessQueue()->putMessage(result.msg_found_list());
defaultMQPushConsumer->getConsumerMsgService()->submitConsumeRequest(
- result.msgFoundList, m_pullRequest->getProcessQueue(), m_pullRequest->getMessageQueue(), true);
+ result.msg_found_list(), m_pullRequest->getProcessQueue(), m_pullRequest->getMessageQueue(), true);
defaultMQPushConsumer->executePullRequestImmediately(m_pullRequest);
}
- if (result.nextBeginOffset < prevRequestOffset || firstMsgOffset < prevRequestOffset) {
+ if (result.next_begin_offset() < prevRequestOffset || firstMsgOffset < prevRequestOffset) {
LOG_WARN_NEW(
"[BUG] pull message result maybe data wrong, nextBeginOffset:{} firstMsgOffset:{} prevRequestOffset:{}",
- result.nextBeginOffset, firstMsgOffset, prevRequestOffset);
+ result.next_begin_offset(), firstMsgOffset, prevRequestOffset);
}
} break;
case NO_NEW_MSG:
case NO_MATCHED_MSG:
- m_pullRequest->setNextOffset(result.nextBeginOffset);
+ m_pullRequest->setNextOffset(result.next_begin_offset());
defaultMQPushConsumer->correctTagsOffset(m_pullRequest);
defaultMQPushConsumer->executePullRequestImmediately(m_pullRequest);
break;
case NO_LATEST_MSG:
- m_pullRequest->setNextOffset(result.nextBeginOffset);
+ m_pullRequest->setNextOffset(result.next_begin_offset());
defaultMQPushConsumer->correctTagsOffset(m_pullRequest);
defaultMQPushConsumer->executePullRequestLater(
m_pullRequest,
@@ -104,7 +104,7 @@ class AsyncPullCallback : public AutoDeletePullCallback {
case OFFSET_ILLEGAL: {
LOG_WARN_NEW("the pull request offset illegal, {} {}", m_pullRequest->toString(), result.toString());
- m_pullRequest->setNextOffset(result.nextBeginOffset);
+ m_pullRequest->setNextOffset(result.next_begin_offset());
m_pullRequest->getProcessQueue()->setDropped(true);
// update and persist offset, then removeProcessQueue
@@ -177,13 +177,13 @@ int DefaultMQPushConsumerImpl::getMaxReconsumeTimes() {
}
}
-bool DefaultMQPushConsumerImpl::sendMessageBack(MQMessageExt& msg, int delayLevel) {
+bool DefaultMQPushConsumerImpl::sendMessageBack(MessageExtPtr msg, int delayLevel) {
return sendMessageBack(msg, delayLevel, null);
}
-bool DefaultMQPushConsumerImpl::sendMessageBack(MQMessageExt& msg, int delayLevel, const std::string& brokerName) {
+bool DefaultMQPushConsumerImpl::sendMessageBack(MessageExtPtr msg, int delayLevel, const std::string& brokerName) {
try {
- std::string brokerAddr = brokerName.empty() ? socketAddress2String(msg.getStoreHost())
+ std::string brokerAddr = brokerName.empty() ? socketAddress2String(msg->getStoreHost())
: m_clientInstance->findBrokerAddressInPublish(brokerName);
m_clientInstance->getMQClientAPIImpl()->consumerSendMessageBack(
@@ -191,7 +191,7 @@ bool DefaultMQPushConsumerImpl::sendMessageBack(MQMessageExt& msg, int delayLeve
return true;
} catch (const std::exception& e) {
LOG_ERROR_NEW("sendMessageBack exception, group: {}, msg: {}. {}", m_pushConsumerConfig->getGroupName(),
- msg.toString(), e.what());
+ msg->toString(), e.what());
}
return false;
}
@@ -293,9 +293,8 @@ void DefaultMQPushConsumerImpl::start() {
if (!registerOK) {
m_serviceState = CREATE_JUST;
m_consumeService->shutdown();
- THROW_MQEXCEPTION(MQClientException,
- "The cousumer group[" + m_pushConsumerConfig->getGroupName() +
- "] has been created before, specify another name please.",
+ THROW_MQEXCEPTION(MQClientException, "The cousumer group[" + m_pushConsumerConfig->getGroupName() +
+ "] has been created before, specify another name please.",
-1);
}
@@ -575,7 +574,8 @@ void DefaultMQPushConsumerImpl::pullMessage(PullRequestPtr pullRequest) {
}
}
-void DefaultMQPushConsumerImpl::resetRetryTopic(std::vector<MQMessageExtPtr2>& msgs, const std::string& consumerGroup) {
+void DefaultMQPushConsumerImpl::resetRetryTopic(const std::vector<MessageExtPtr>& msgs,
+ const std::string& consumerGroup) {
std::string groupTopic = UtilAll::getRetryTopic(consumerGroup);
for (auto& msg : msgs) {
std::string retryTopic = msg->getProperty(MQMessageConst::PROPERTY_RETRY_TOPIC);
diff --git a/src/consumer/DefaultMQPushConsumerImpl.h b/src/consumer/DefaultMQPushConsumerImpl.h
index 67f03ef..95563a2 100755
--- a/src/consumer/DefaultMQPushConsumerImpl.h
+++ b/src/consumer/DefaultMQPushConsumerImpl.h
@@ -64,8 +64,8 @@ class DefaultMQPushConsumerImpl : public std::enable_shared_from_this<DefaultMQP
void shutdown() override;
public: // MQConsumer
- bool sendMessageBack(MQMessageExt& msg, int delayLevel) override;
- bool sendMessageBack(MQMessageExt& msg, int delayLevel, const std::string& brokerName) override;
+ bool sendMessageBack(MessageExtPtr msg, int delayLevel) override;
+ bool sendMessageBack(MessageExtPtr msg, int delayLevel, const std::string& brokerName) override;
void fetchSubscribeMessageQueues(const std::string& topic, std::vector<MQMessageQueue>& mqs) override;
public: // MQPushConsumer
@@ -105,7 +105,7 @@ class DefaultMQPushConsumerImpl : public std::enable_shared_from_this<DefaultMQP
void pullMessage(PullRequestPtr pullrequest);
- void resetRetryTopic(std::vector<MQMessageExtPtr2>& msgs, const std::string& consumerGroup);
+ void resetRetryTopic(const std::vector<MessageExtPtr>& msgs, const std::string& consumerGroup);
public:
bool isConsumeOrderly() { return m_consumeOrderly; }
diff --git a/src/consumer/ProcessQueue.cpp b/src/consumer/ProcessQueue.cpp
index 5355448..285caed 100644
--- a/src/consumer/ProcessQueue.cpp
+++ b/src/consumer/ProcessQueue.cpp
@@ -47,7 +47,7 @@ bool ProcessQueue::isPullExpired() const {
return (UtilAll::currentTimeMillis() - m_lastPullTimestamp) > PullMaxIdleTime;
}
-void ProcessQueue::putMessage(std::vector<MQMessageExtPtr2>& msgs) {
+void ProcessQueue::putMessage(const std::vector<MessageExtPtr>& msgs) {
std::lock_guard<std::mutex> lock(m_lockTreeMap);
for (const auto& msg : msgs) {
@@ -61,7 +61,7 @@ void ProcessQueue::putMessage(std::vector<MQMessageExtPtr2>& msgs) {
LOG_DEBUG_NEW("ProcessQueue: putMessage m_queueOffsetMax:{}", m_queueOffsetMax);
}
-int64_t ProcessQueue::removeMessage(std::vector<MQMessageExtPtr2>& msgs) {
+int64_t ProcessQueue::removeMessage(const std::vector<MessageExtPtr>& msgs) {
int64_t result = -1;
const auto now = UtilAll::currentTimeMillis();
@@ -131,7 +131,7 @@ int64_t ProcessQueue::commit() {
}
}
-void ProcessQueue::makeMessageToCosumeAgain(std::vector<MQMessageExtPtr2>& msgs) {
+void ProcessQueue::makeMessageToCosumeAgain(std::vector<MessageExtPtr>& msgs) {
std::lock_guard<std::mutex> lock(m_lockTreeMap);
for (const auto& msg : msgs) {
m_msgTreeMap[msg->getQueueOffset()] = msg;
@@ -139,7 +139,7 @@ void ProcessQueue::makeMessageToCosumeAgain(std::vector<MQMessageExtPtr2>& msgs)
}
}
-void ProcessQueue::takeMessages(std::vector<MQMessageExtPtr2>& out_msgs, int batchSize) {
+void ProcessQueue::takeMessages(std::vector<MessageExtPtr>& out_msgs, int batchSize) {
std::lock_guard<std::mutex> lock(m_lockTreeMap);
for (int i = 0; i != batchSize; i++) {
const auto& it = m_msgTreeMap.begin();
diff --git a/src/consumer/ProcessQueue.h b/src/consumer/ProcessQueue.h
index 84672c5..e2780ed 100644
--- a/src/consumer/ProcessQueue.h
+++ b/src/consumer/ProcessQueue.h
@@ -23,7 +23,7 @@
#include <mutex>
#include <vector>
-#include "MQMessageExt.h"
+#include "MessageExt.h"
namespace rocketmq {
@@ -44,8 +44,8 @@ class ROCKETMQCLIENT_API ProcessQueue {
bool isLockExpired() const;
bool isPullExpired() const;
- void putMessage(std::vector<MQMessageExtPtr2>& msgs);
- int64_t removeMessage(std::vector<MQMessageExtPtr2>& msgs);
+ void putMessage(const std::vector<MessageExtPtr>& msgs);
+ int64_t removeMessage(const std::vector<MessageExtPtr>& msgs);
int getCacheMsgCount();
int64_t getCacheMinOffset();
@@ -58,8 +58,8 @@ class ROCKETMQCLIENT_API ProcessQueue {
void setLocked(bool locked);
int64_t commit();
- void makeMessageToCosumeAgain(std::vector<MQMessageExtPtr2>& msgs);
- void takeMessages(std::vector<MQMessageExtPtr2>& out_msgs, int batchSize);
+ void makeMessageToCosumeAgain(std::vector<MessageExtPtr>& msgs);
+ void takeMessages(std::vector<MessageExtPtr>& out_msgs, int batchSize);
void clearAllMsgs();
@@ -84,9 +84,9 @@ class ROCKETMQCLIENT_API ProcessQueue {
private:
std::mutex m_lockTreeMap;
- std::map<int64_t, MQMessageExtPtr2> m_msgTreeMap;
+ std::map<int64_t, MessageExtPtr> m_msgTreeMap;
std::timed_mutex m_lockConsume;
- std::map<int64_t, MQMessageExtPtr2> m_consumingMsgOrderlyTreeMap;
+ std::map<int64_t, MessageExtPtr> m_consumingMsgOrderlyTreeMap;
std::atomic<long> m_tryUnlockTimes;
volatile int64_t m_queueOffsetMax;
std::atomic<bool> m_dropped;
diff --git a/src/consumer/PullAPIWrapper.cpp b/src/consumer/PullAPIWrapper.cpp
index 7af5f8e..4b7e936 100644
--- a/src/consumer/PullAPIWrapper.cpp
+++ b/src/consumer/PullAPIWrapper.cpp
@@ -18,7 +18,7 @@
#include "MQClientAPIImpl.h"
#include "MQClientInstance.h"
-#include "MQDecoder.h"
+#include "MessageDecoder.h"
#include "MessageAccessor.h"
#include "PullResultExt.h"
#include "PullSysFlag.h"
@@ -58,10 +58,10 @@ PullResult PullAPIWrapper::processPullResult(const MQMessageQueue& mq,
// update node
updatePullFromWhichNode(mq, pullResultExt.suggestWhichBrokerId);
- std::vector<MQMessageExtPtr2> msgListFilterAgain;
- if (FOUND == pullResultExt.pullStatus) {
+ std::vector<MessageExtPtr> msgListFilterAgain;
+ if (FOUND == pullResultExt.pull_status()) {
// decode all msg list
- auto msgList = MQDecoder::decodes(*pullResultExt.msgMemBlock);
+ auto msgList = MessageDecoder::decodes(*pullResultExt.msgMemBlock);
// filter msg list again
if (subscriptionData != nullptr && !subscriptionData->getTagsSet().empty()) {
@@ -81,13 +81,15 @@ PullResult PullAPIWrapper::processPullResult(const MQMessageQueue& mq,
if (UtilAll::stob(tranMsg)) {
msg->setTransactionId(msg->getProperty(MQMessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX));
}
- MessageAccessor::putProperty(*msg, MQMessageConst::PROPERTY_MIN_OFFSET, UtilAll::to_string(pullResult.minOffset));
- MessageAccessor::putProperty(*msg, MQMessageConst::PROPERTY_MAX_OFFSET, UtilAll::to_string(pullResult.maxOffset));
+ MessageAccessor::putProperty(*msg, MQMessageConst::PROPERTY_MIN_OFFSET,
+ UtilAll::to_string(pullResult.min_offset()));
+ MessageAccessor::putProperty(*msg, MQMessageConst::PROPERTY_MAX_OFFSET,
+ UtilAll::to_string(pullResult.max_offset()));
}
}
- return PullResult(pullResultExt.pullStatus, pullResultExt.nextBeginOffset, pullResultExt.minOffset,
- pullResultExt.maxOffset, std::move(msgListFilterAgain));
+ return PullResult(pullResultExt.pull_status(), pullResultExt.next_begin_offset(), pullResultExt.min_offset(),
+ pullResultExt.max_offset(), std::move(msgListFilterAgain));
}
PullResult* PullAPIWrapper::pullKernelImpl(const MQMessageQueue& mq, // 1
diff --git a/src/consumer/PullResult.cpp b/src/consumer/PullResult.cpp
index 8fe82f2..133f2fe 100644
--- a/src/consumer/PullResult.cpp
+++ b/src/consumer/PullResult.cpp
@@ -16,50 +16,55 @@
*/
#include "PullResult.h"
+#include <algorithm> // std::move
+#include <sstream> // std::stringstream
+
#include "UtilAll.h"
namespace rocketmq {
-static const char* EnumStrings[] = {"FOUND", "NO_NEW_MSG", "NO_MATCHED_MSG",
- "NO_LATEST_MSG"
- "OFFSET_ILLEGAL",
- "BROKER_TIMEOUT"};
+static const char* kPullStatusStrings[] = {"FOUND", "NO_NEW_MSG", "NO_MATCHED_MSG",
+ "NO_LATEST_MSG"
+ "OFFSET_ILLEGAL",
+ "BROKER_TIMEOUT"};
-PullResult::PullResult() : pullStatus(NO_MATCHED_MSG), nextBeginOffset(0), minOffset(0), maxOffset(0) {}
+PullResult::PullResult() : pull_status_(NO_MATCHED_MSG), next_begin_offset_(0), min_offset_(0), max_offset_(0) {}
-PullResult::PullResult(PullStatus status) : pullStatus(status), nextBeginOffset(0), minOffset(0), maxOffset(0) {}
+PullResult::PullResult(PullStatus status)
+ : pull_status_(status), next_begin_offset_(0), min_offset_(0), max_offset_(0) {}
PullResult::PullResult(PullStatus pullStatus, int64_t nextBeginOffset, int64_t minOffset, int64_t maxOffset)
- : pullStatus(pullStatus), nextBeginOffset(nextBeginOffset), minOffset(minOffset), maxOffset(maxOffset) {}
+ : pull_status_(pullStatus), next_begin_offset_(nextBeginOffset), min_offset_(minOffset), max_offset_(maxOffset) {}
PullResult::PullResult(PullStatus pullStatus,
int64_t nextBeginOffset,
int64_t minOffset,
int64_t maxOffset,
- const std::vector<MQMessageExtPtr2>& src)
- : pullStatus(pullStatus),
- nextBeginOffset(nextBeginOffset),
- minOffset(minOffset),
- maxOffset(maxOffset),
- msgFoundList(src) {}
+ const std::vector<MessageExtPtr>& msgFoundList)
+ : pull_status_(pullStatus),
+ next_begin_offset_(nextBeginOffset),
+ min_offset_(minOffset),
+ max_offset_(maxOffset),
+ msg_found_list_(msgFoundList) {}
PullResult::PullResult(PullStatus pullStatus,
int64_t nextBeginOffset,
int64_t minOffset,
int64_t maxOffset,
- std::vector<MQMessageExtPtr2>&& src)
- : pullStatus(pullStatus),
- nextBeginOffset(nextBeginOffset),
- minOffset(minOffset),
- maxOffset(maxOffset),
- msgFoundList(std::forward<std::vector<MQMessageExtPtr2>>(src)) {}
+ std::vector<MessageExtPtr>&& msgFoundList)
+ : pull_status_(pullStatus),
+ next_begin_offset_(nextBeginOffset),
+ min_offset_(minOffset),
+ max_offset_(maxOffset),
+ msg_found_list_(std::move(msgFoundList)) {}
PullResult::~PullResult() = default;
std::string PullResult::toString() const {
std::stringstream ss;
- ss << "PullResult [ pullStatus=" << EnumStrings[pullStatus] << ", nextBeginOffset=" << nextBeginOffset
- << ", minOffset=" << minOffset << ", maxOffset=" << maxOffset << ", msgFoundList=" << msgFoundList.size() << " ]";
+ ss << "PullResult [ pullStatus=" << kPullStatusStrings[pull_status_] << ", nextBeginOffset=" << next_begin_offset_
+ << ", minOffset=" << min_offset_ << ", maxOffset=" << max_offset_ << ", msgFoundList=" << msg_found_list_.size()
+ << " ]";
return ss.str();
}
diff --git a/include/PullResult.h b/src/consumer/PullResult.h
similarity index 61%
rename from include/PullResult.h
rename to src/consumer/PullResult.h
index eccf19e..0802fef 100644
--- a/include/PullResult.h
+++ b/src/consumer/PullResult.h
@@ -14,13 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef __PULL_RESULT_H__
-#define __PULL_RESULT_H__
+#ifndef ROCKETMQ_CONSUMER_PULLRESULT_H_
+#define ROCKETMQ_CONSUMER_PULLRESULT_H_
-#include <memory>
-#include <sstream>
-
-#include "MQMessageExt.h"
+#include "MessageExt.h"
namespace rocketmq {
@@ -33,7 +30,7 @@ enum PullStatus {
BROKER_TIMEOUT // indicate pull request timeout or received NULL response
};
-class ROCKETMQCLIENT_API PullResult {
+class PullResult {
public:
PullResult();
PullResult(PullStatus status);
@@ -43,25 +40,35 @@ class ROCKETMQCLIENT_API PullResult {
int64_t nextBeginOffset,
int64_t minOffset,
int64_t maxOffset,
- const std::vector<MQMessageExtPtr2>& src);
+ const std::vector<MessageExtPtr>& msgFoundList);
PullResult(PullStatus pullStatus,
int64_t nextBeginOffset,
int64_t minOffset,
int64_t maxOffset,
- std::vector<MQMessageExtPtr2>&& src);
+ std::vector<MessageExtPtr>&& msgFoundList);
virtual ~PullResult();
std::string toString() const;
- public:
- PullStatus pullStatus;
- int64_t nextBeginOffset;
- int64_t minOffset;
- int64_t maxOffset;
- std::vector<MQMessageExtPtr2> msgFoundList;
+ inline PullStatus pull_status() const { return pull_status_; };
+ inline void set_pull_status(PullStatus pull_status) { pull_status_ = pull_status; }
+
+ inline int64_t next_begin_offset() const { return next_begin_offset_; };
+ inline int64_t min_offset() const { return min_offset_; }
+ inline int64_t max_offset() const { return max_offset_; }
+
+ inline std::vector<MessageExtPtr>& msg_found_list() { return msg_found_list_; }
+ inline const std::vector<MessageExtPtr>& msg_found_list() const { return msg_found_list_; }
+
+ private:
+ PullStatus pull_status_;
+ int64_t next_begin_offset_;
+ int64_t min_offset_;
+ int64_t max_offset_;
+ std::vector<MessageExtPtr> msg_found_list_;
};
} // namespace rocketmq
-#endif // __PULL_RESULT_H__
+#endif // ROCKETMQ_CONSUMER_PULLRESULT_H_
diff --git a/src/extern/CBatchMessage.cpp b/src/extern/CBatchMessage.cpp
index 2ea6ece..ecc452a 100644
--- a/src/extern/CBatchMessage.cpp
+++ b/src/extern/CBatchMessage.cpp
@@ -23,7 +23,7 @@
using namespace rocketmq;
CBatchMessage* CreateBatchMessage() {
- auto* msgs = new std::vector<MQMessage*>();
+ auto* msgs = new std::vector<MQMessage>();
return reinterpret_cast<CBatchMessage*>(msgs);
}
@@ -35,7 +35,7 @@ int AddMessage(CBatchMessage* batchMsg, CMessage* msg) {
return NULL_POINTER;
}
auto* message = reinterpret_cast<MQMessage*>(msg);
- reinterpret_cast<std::vector<MQMessage*>*>(batchMsg)->push_back(message);
+ reinterpret_cast<std::vector<MQMessage>*>(batchMsg)->push_back(*message);
return OK;
}
@@ -43,10 +43,7 @@ int DestroyBatchMessage(CBatchMessage* batchMsg) {
if (batchMsg == NULL) {
return NULL_POINTER;
}
- auto* msgs = reinterpret_cast<std::vector<MQMessage*>*>(batchMsg);
- for (auto* msg : *msgs) {
- delete msg;
- }
+ auto* msgs = reinterpret_cast<std::vector<MQMessage>*>(batchMsg);
delete msgs;
return OK;
}
diff --git a/src/extern/CProducer.cpp b/src/extern/CProducer.cpp
index 60502c2..cf2df53 100644
--- a/src/extern/CProducer.cpp
+++ b/src/extern/CProducer.cpp
@@ -256,7 +256,7 @@ int SendMessageSync(CProducer* producer, CMessage* msg, CSendResult* result) {
try {
auto* defaultMQProducer = reinterpret_cast<DefaultMQProducer*>(producer);
auto* message = reinterpret_cast<MQMessage*>(msg);
- auto sendResult = defaultMQProducer->send(message);
+ auto sendResult = defaultMQProducer->send(*message);
switch (sendResult.getSendStatus()) {
case SEND_OK:
result->sendStatus = E_SEND_OK;
@@ -290,7 +290,7 @@ int SendBatchMessage(CProducer* producer, CBatchMessage* batcMsg, CSendResult* r
}
try {
auto* defaultMQProducer = reinterpret_cast<DefaultMQProducer*>(producer);
- auto* message = reinterpret_cast<std::vector<MQMessage*>*>(batcMsg);
+ auto* message = reinterpret_cast<std::vector<MQMessage>*>(batcMsg);
auto sendResult = defaultMQProducer->send(*message);
switch (sendResult.getSendStatus()) {
case SEND_OK:
@@ -328,7 +328,7 @@ int SendMessageAsync(CProducer* producer,
auto* defaultMQProducer = reinterpret_cast<DefaultMQProducer*>(producer);
auto* message = reinterpret_cast<MQMessage*>(msg);
auto* sendCallback = new CSendCallback(sendSuccessCallback, sendExceptionCallback);
- defaultMQProducer->send(message, sendCallback);
+ defaultMQProducer->send(*message, sendCallback);
return OK;
}
@@ -343,7 +343,7 @@ int SendAsync(CProducer* producer,
auto* defaultMQProducer = reinterpret_cast<DefaultMQProducer*>(producer);
auto* message = reinterpret_cast<MQMessage*>(msg);
auto* sendCallback = new COnSendCallback(sendSuccessCallback, sendExceptionCallback, msg, userData);
- defaultMQProducer->send(message, sendCallback);
+ defaultMQProducer->send(*message, sendCallback);
return OK;
}
@@ -354,7 +354,7 @@ int SendMessageOneway(CProducer* producer, CMessage* msg) {
auto* defaultMQProducer = reinterpret_cast<DefaultMQProducer*>(producer);
auto* message = reinterpret_cast<MQMessage*>(msg);
try {
- defaultMQProducer->sendOneway(message);
+ defaultMQProducer->sendOneway(*message);
} catch (std::exception& e) {
return PRODUCER_SEND_ONEWAY_FAILED;
}
@@ -369,7 +369,7 @@ int SendMessageOnewayOrderly(CProducer* producer, CMessage* msg, QueueSelectorCa
auto* message = reinterpret_cast<MQMessage*>(msg);
try {
SelectMessageQueue selectMessageQueue(selector);
- defaultMQProducer->sendOneway(message, &selectMessageQueue, arg);
+ defaultMQProducer->sendOneway(*message, &selectMessageQueue, arg);
} catch (std::exception& e) {
MQClientErrorContainer::setErr(std::string(e.what()));
return PRODUCER_SEND_ONEWAY_FAILED;
@@ -392,7 +392,7 @@ int SendMessageOrderlyAsync(CProducer* producer,
auto* cSendCallback = new CSendCallback(sendSuccessCallback, sendExceptionCallback);
// Constructing SelectMessageQueue objects through function pointer callback
SelectMessageQueue selectMessageQueue(selectorCallback);
- defaultMQProducer->send(message, &selectMessageQueue, arg, cSendCallback);
+ defaultMQProducer->send(*message, &selectMessageQueue, arg, cSendCallback);
return OK;
}
@@ -410,7 +410,7 @@ int SendMessageOrderly(CProducer* producer,
try {
// Constructing SelectMessageQueue objects through function pointer callback
SelectMessageQueue selectMessageQueue(selectorCallback);
- SendResult sendResult = defaultMQProducer->send(message, &selectMessageQueue, arg);
+ SendResult sendResult = defaultMQProducer->send(*message, &selectMessageQueue, arg);
// Convert SendStatus to CSendStatus
result->sendStatus = CSendStatus((int)sendResult.getSendStatus());
result->offset = sendResult.getQueueOffset();
@@ -433,7 +433,7 @@ int SendMessageOrderlyByShardingKey(CProducer* producer, CMessage* msg, const ch
// Constructing SelectMessageQueue objects through function pointer callback
int retryTimes = 3;
SelectMessageQueueInner selectMessageQueue;
- SendResult sendResult = defaultMQProducer->send(message, &selectMessageQueue, (void*)shardingKey, retryTimes);
+ SendResult sendResult = defaultMQProducer->send(*message, &selectMessageQueue, (void*)shardingKey, retryTimes);
// Convert SendStatus to CSendStatus
result->sendStatus = CSendStatus((int)sendResult.getSendStatus());
result->offset = sendResult.getQueueOffset();
@@ -458,7 +458,7 @@ int SendMessageTransaction(CProducer* producer,
auto* transactionMQProducer = reinterpret_cast<DefaultMQProducer*>(producer);
auto* message = reinterpret_cast<MQMessage*>(msg);
LocalTransactionExecutorInner executorInner(callback, msg, userData);
- auto sendResult = transactionMQProducer->sendMessageInTransaction(message, &executorInner);
+ auto sendResult = transactionMQProducer->sendMessageInTransaction(*message, &executorInner);
result->sendStatus = CSendStatus((int)sendResult.getSendStatus());
result->offset = sendResult.getQueueOffset();
strncpy(result->msgId, sendResult.getMsgId().c_str(), MAX_MESSAGE_ID_LENGTH - 1);
diff --git a/src/extern/CPullConsumer.cpp b/src/extern/CPullConsumer.cpp
index 61210a3..f3fe103 100644
--- a/src/extern/CPullConsumer.cpp
+++ b/src/extern/CPullConsumer.cpp
@@ -187,27 +187,27 @@ CPullResult Pull(CPullConsumer* consumer,
reinterpret_cast<DefaultMQPullConsumer*>(consumer)->pull(messageQueue, subExpression, offset, maxNums);
} catch (std::exception& e) {
MQClientErrorContainer::setErr(std::string(e.what()));
- cppPullResult.pullStatus = BROKER_TIMEOUT;
+ cppPullResult.set_pull_status(BROKER_TIMEOUT);
}
- if (cppPullResult.pullStatus != BROKER_TIMEOUT) {
- pullResult.maxOffset = cppPullResult.maxOffset;
- pullResult.minOffset = cppPullResult.minOffset;
- pullResult.nextBeginOffset = cppPullResult.nextBeginOffset;
+ if (cppPullResult.pull_status() != BROKER_TIMEOUT) {
+ pullResult.maxOffset = cppPullResult.max_offset();
+ pullResult.minOffset = cppPullResult.min_offset();
+ pullResult.nextBeginOffset = cppPullResult.next_begin_offset();
}
- switch (cppPullResult.pullStatus) {
+ switch (cppPullResult.pull_status()) {
case FOUND: {
pullResult.pullStatus = E_FOUND;
- pullResult.size = cppPullResult.msgFoundList.size();
+ pullResult.size = cppPullResult.msg_found_list().size();
PullResult* tmpPullResult = new PullResult(cppPullResult);
pullResult.pData = tmpPullResult;
// Alloc memory to save the pointer to CPP MQMessageExt, which will be release by the CPP SDK core.
// Thus, this memory should be released by users using @ReleasePullResult
pullResult.msgFoundList = (CMessageExt**)malloc(pullResult.size * sizeof(CMessageExt*));
- for (size_t i = 0; i < cppPullResult.msgFoundList.size(); i++) {
- MQMessageExt* msg = tmpPullResult->msgFoundList[i].get();
- pullResult.msgFoundList[i] = (CMessageExt*)(msg);
+ for (size_t i = 0; i < cppPullResult.msg_found_list().size(); i++) {
+ auto msg = tmpPullResult->msg_found_list()[i];
+ pullResult.msgFoundList[i] = reinterpret_cast<CMessageExt*>(&msg);
}
break;
}
diff --git a/src/extern/CPushConsumer.cpp b/src/extern/CPushConsumer.cpp
index 40fbd2b..b93a94f 100644
--- a/src/extern/CPushConsumer.cpp
+++ b/src/extern/CPushConsumer.cpp
@@ -16,7 +16,7 @@
*/
#include "c/CPushConsumer.h"
-#include <map>
+#include <map> // std::map
#include "ClientRPCHook.h"
#include "DefaultMQPushConsumer.h"
@@ -32,13 +32,13 @@ class MessageListenerInner : public MessageListenerConcurrently {
~MessageListenerInner() = default;
- ConsumeStatus consumeMessage(const std::vector<MQMessageExt*>& msgs) override {
+ ConsumeStatus consumeMessage(const std::vector<MQMessageExt>& msgs) override {
// to do user call back
if (m_msgReceivedCallback == nullptr) {
return RECONSUME_LATER;
}
for (auto msg : msgs) {
- auto* message = reinterpret_cast<CMessageExt*>(msg);
+ auto* message = reinterpret_cast<CMessageExt*>(&msg);
if (m_msgReceivedCallback(m_consumer, message) != E_CONSUME_SUCCESS) {
return RECONSUME_LATER;
}
@@ -56,12 +56,12 @@ class MessageListenerOrderlyInner : public MessageListenerOrderly {
MessageListenerOrderlyInner(CPushConsumer* consumer, MessageCallBack callback)
: m_consumer(consumer), m_msgReceivedCallback(callback) {}
- ConsumeStatus consumeMessage(const std::vector<MQMessageExt*>& msgs) {
+ ConsumeStatus consumeMessage(const std::vector<MQMessageExt>& msgs) override {
if (m_msgReceivedCallback == nullptr) {
return RECONSUME_LATER;
}
for (auto msg : msgs) {
- auto* message = reinterpret_cast<CMessageExt*>(msg);
+ auto* message = reinterpret_cast<CMessageExt*>(&msg);
if (m_msgReceivedCallback(m_consumer, message) != E_CONSUME_SUCCESS) {
return RECONSUME_LATER;
}
diff --git a/src/message/MQMessage.cpp b/src/message/MQMessage.cpp
index 3f2a29c..71f8a30 100644
--- a/src/message/MQMessage.cpp
+++ b/src/message/MQMessage.cpp
@@ -16,55 +16,16 @@
*/
#include "MQMessage.h"
-#include "MessageSysFlag.h"
-#include "UtilAll.h"
+#include <algorithm> // std::move
-namespace rocketmq {
+#include "UtilAll.h" // null
+#include "MessageImpl.h"
-const std::string MQMessageConst::PROPERTY_KEYS = "KEYS";
-const std::string MQMessageConst::PROPERTY_TAGS = "TAGS";
-const std::string MQMessageConst::PROPERTY_WAIT_STORE_MSG_OK = "WAIT";
-const std::string MQMessageConst::PROPERTY_DELAY_TIME_LEVEL = "DELAY";
-const std::string MQMessageConst::PROPERTY_RETRY_TOPIC = "RETRY_TOPIC";
-const std::string MQMessageConst::PROPERTY_REAL_TOPIC = "REAL_TOPIC";
-const std::string MQMessageConst::PROPERTY_REAL_QUEUE_ID = "REAL_QID";
-const std::string MQMessageConst::PROPERTY_TRANSACTION_PREPARED = "TRAN_MSG";
-const std::string MQMessageConst::PROPERTY_PRODUCER_GROUP = "PGROUP";
-const std::string MQMessageConst::PROPERTY_MIN_OFFSET = "MIN_OFFSET";
-const std::string MQMessageConst::PROPERTY_MAX_OFFSET = "MAX_OFFSET";
-
-const std::string MQMessageConst::PROPERTY_BUYER_ID = "BUYER_ID";
-const std::string MQMessageConst::PROPERTY_ORIGIN_MESSAGE_ID = "ORIGIN_MESSAGE_ID";
-const std::string MQMessageConst::PROPERTY_TRANSFER_FLAG = "TRANSFER_FLAG";
-const std::string MQMessageConst::PROPERTY_CORRECTION_FLAG = "CORRECTION_FLAG";
-const std::string MQMessageConst::PROPERTY_MQ2_FLAG = "MQ2_FLAG";
-const std::string MQMessageConst::PROPERTY_RECONSUME_TIME = "RECONSUME_TIME";
-const std::string MQMessageConst::PROPERTY_MSG_REGION = "MSG_REGION";
-const std::string MQMessageConst::PROPERTY_TRACE_SWITCH = "TRACE_ON";
-const std::string MQMessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX = "UNIQ_KEY";
-const std::string MQMessageConst::PROPERTY_MAX_RECONSUME_TIMES = "MAX_RECONSUME_TIMES";
-const std::string MQMessageConst::PROPERTY_CONSUME_START_TIMESTAMP = "CONSUME_START_TIME";
-const std::string MQMessageConst::PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET = "TRAN_PREPARED_QUEUE_OFFSET";
-const std::string MQMessageConst::PROPERTY_TRANSACTION_CHECK_TIMES = "TRANSACTION_CHECK_TIMES";
-const std::string MQMessageConst::PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS = "CHECK_IMMUNITY_TIME_IN_SECONDS";
-const std::string MQMessageConst::PROPERTY_INSTANCE_ID = "INSTANCE_ID";
-const std::string MQMessageConst::PROPERTY_CORRELATION_ID = "CORRELATION_ID";
-const std::string MQMessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT = "REPLY_TO_CLIENT";
-const std::string MQMessageConst::PROPERTY_MESSAGE_TTL = "TTL";
-const std::string MQMessageConst::PROPERTY_REPLY_MESSAGE_ARRIVE_TIME = "ARRIVE_TIME";
-const std::string MQMessageConst::PROPERTY_PUSH_REPLY_TIME = "PUSH_REPLY_TIME";
-const std::string MQMessageConst::PROPERTY_CLUSTER = "CLUSTER";
-const std::string MQMessageConst::PROPERTY_MESSAGE_TYPE = "MSG_TYPE";
-
-const std::string MQMessageConst::PROPERTY_ALREADY_COMPRESSED_FLAG = "__ALREADY_COMPRESSED__";
-
-const std::string MQMessageConst::KEY_SEPARATOR = " ";
-
-static const std::string EMPTY_STRING = "";
+namespace rocketmq {
MQMessage::MQMessage() : MQMessage(null, null) {}
-MQMessage::MQMessage(const std::string& topic, const std::string& body) : MQMessage(topic, "", body) {}
+MQMessage::MQMessage(const std::string& topic, const std::string& body) : MQMessage(topic, null, body) {}
MQMessage::MQMessage(const std::string& topic, const std::string& tags, const std::string& body)
: MQMessage(topic, tags, null, body) {}
@@ -78,168 +39,127 @@ MQMessage::MQMessage(const std::string& topic,
MQMessage::MQMessage(const std::string& topic,
const std::string& tags,
const std::string& keys,
- const int flag,
+ int32_t flag,
const std::string& body,
bool waitStoreMsgOK)
- : m_topic(topic), m_flag(flag), m_body(body) {
- if (tags.length() > 0) {
- setTags(tags);
- }
-
- if (keys.length() > 0) {
- setKeys(keys);
- }
-
- setWaitStoreMsgOK(waitStoreMsgOK);
-}
-
-MQMessage::MQMessage(const MQMessage& other) {
- m_topic = other.m_topic;
- m_flag = other.m_flag;
- m_properties = other.m_properties;
- m_body = other.m_body;
-}
-
-MQMessage& MQMessage::operator=(const MQMessage& other) {
- if (this != &other) {
- m_topic = other.m_topic;
- m_flag = other.m_flag;
- m_properties = other.m_properties;
- m_body = other.m_body;
- }
- return *this;
-}
+ : message_impl_(std::make_shared<MessageImpl>(topic, tags, keys, flag, body, waitStoreMsgOK)) {}
MQMessage::~MQMessage() = default;
const std::string& MQMessage::getProperty(const std::string& name) const {
- const auto& it = m_properties.find(name);
- if (it != m_properties.end()) {
- return it->second;
- }
- return EMPTY_STRING;
+ return message_impl_->getProperty(name);
}
void MQMessage::putProperty(const std::string& name, const std::string& value) {
- m_properties[name] = value;
+ message_impl_->putProperty(name, value);
}
void MQMessage::clearProperty(const std::string& name) {
- m_properties.erase(name);
+ message_impl_->clearProperty(name);
}
const std::string& MQMessage::getTopic() const {
- return m_topic;
+ return message_impl_->getTopic();
}
void MQMessage::setTopic(const std::string& topic) {
- m_topic = topic;
+ message_impl_->setTopic(topic);
}
void MQMessage::setTopic(const char* body, int len) {
- m_topic.clear();
- m_topic.append(body, len);
+ message_impl_->setTopic(body, len);
}
const std::string& MQMessage::getTags() const {
- return getProperty(MQMessageConst::PROPERTY_TAGS);
+ return message_impl_->getTags();
}
void MQMessage::setTags(const std::string& tags) {
- putProperty(MQMessageConst::PROPERTY_TAGS, tags);
+ message_impl_->setTags(tags);
}
const std::string& MQMessage::getKeys() const {
- return getProperty(MQMessageConst::PROPERTY_KEYS);
+ return message_impl_->getKeys();
}
void MQMessage::setKeys(const std::string& keys) {
- putProperty(MQMessageConst::PROPERTY_KEYS, keys);
+ message_impl_->setKeys(keys);
}
void MQMessage::setKeys(const std::vector<std::string>& keys) {
- if (keys.empty()) {
- return;
- }
-
- std::vector<std::string>::const_iterator it = keys.begin();
- std::string str;
- str += *it;
- it++;
-
- for (; it != keys.end(); it++) {
- str += MQMessageConst::KEY_SEPARATOR;
- str += *it;
- }
-
- setKeys(str);
+ message_impl_->setKeys(keys);
}
int MQMessage::getDelayTimeLevel() const {
- std::string tmp = getProperty(MQMessageConst::PROPERTY_DELAY_TIME_LEVEL);
- if (!tmp.empty()) {
- return atoi(tmp.c_str());
- }
- return 0;
+ return message_impl_->getDelayTimeLevel();
}
void MQMessage::setDelayTimeLevel(int level) {
- putProperty(MQMessageConst::PROPERTY_DELAY_TIME_LEVEL, UtilAll::to_string(level));
+ message_impl_->setDelayTimeLevel(level);
}
bool MQMessage::isWaitStoreMsgOK() const {
- std::string tmp = getProperty(MQMessageConst::PROPERTY_WAIT_STORE_MSG_OK);
- return tmp.empty() || UtilAll::stob(tmp);
+ return message_impl_->isWaitStoreMsgOK();
}
void MQMessage::setWaitStoreMsgOK(bool waitStoreMsgOK) {
- putProperty(MQMessageConst::PROPERTY_WAIT_STORE_MSG_OK, UtilAll::to_string(waitStoreMsgOK));
+ message_impl_->setWaitStoreMsgOK(waitStoreMsgOK);
}
-int MQMessage::getFlag() const {
- return m_flag;
+int32_t MQMessage::getFlag() const {
+ return message_impl_->getFlag();
}
-void MQMessage::setFlag(int flag) {
- m_flag = flag;
+void MQMessage::setFlag(int32_t flag) {
+ message_impl_->setFlag(flag);
}
const std::string& MQMessage::getBody() const {
- // TODO: return MemoryBlockPtr2
- return m_body;
+ return message_impl_->getBody();
}
void MQMessage::setBody(const char* body, int len) {
- m_body.clear();
- m_body.append(body, len);
+ message_impl_->setBody(body, len);
}
void MQMessage::setBody(const std::string& body) {
- m_body = body;
+ message_impl_->setBody(body);
}
void MQMessage::setBody(std::string&& body) {
- m_body = std::forward<std::string>(body);
+ message_impl_->setBody(std::move(body));
}
const std::string& MQMessage::getTransactionId() const {
- return m_transactionId;
+ return message_impl_->getTransactionId();
}
void MQMessage::setTransactionId(const std::string& transactionId) {
- m_transactionId = transactionId;
+ message_impl_->setTransactionId(transactionId);
}
const std::map<std::string, std::string>& MQMessage::getProperties() const {
- return m_properties;
+ return message_impl_->getProperties();
}
void MQMessage::setProperties(const std::map<std::string, std::string>& properties) {
- m_properties = properties;
+ message_impl_->setProperties(properties);
}
void MQMessage::setProperties(std::map<std::string, std::string>&& properties) {
- m_properties = std::forward<std::map<std::string, std::string>>(properties);
+ message_impl_->setProperties(std::move(properties));
+}
+
+bool MQMessage::isBatch() const {
+ return message_impl_->isBatch();
+}
+
+std::string MQMessage::toString() const {
+ return message_impl_->toString();
+}
+
+MessagePtr MQMessage::getMessageImpl() {
+ return message_impl_;
}
} // namespace rocketmq
diff --git a/src/message/MQMessageConst.cpp b/src/message/MQMessageConst.cpp
new file mode 100644
index 0000000..bf5e90f
--- /dev/null
+++ b/src/message/MQMessageConst.cpp
@@ -0,0 +1,60 @@
+/*
+ * 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 "MQMessageConst.h"
+
+namespace rocketmq {
+
+const std::string MQMessageConst::PROPERTY_KEYS = "KEYS";
+const std::string MQMessageConst::PROPERTY_TAGS = "TAGS";
+const std::string MQMessageConst::PROPERTY_WAIT_STORE_MSG_OK = "WAIT";
+const std::string MQMessageConst::PROPERTY_DELAY_TIME_LEVEL = "DELAY";
+const std::string MQMessageConst::PROPERTY_RETRY_TOPIC = "RETRY_TOPIC";
+const std::string MQMessageConst::PROPERTY_REAL_TOPIC = "REAL_TOPIC";
+const std::string MQMessageConst::PROPERTY_REAL_QUEUE_ID = "REAL_QID";
+const std::string MQMessageConst::PROPERTY_TRANSACTION_PREPARED = "TRAN_MSG";
+const std::string MQMessageConst::PROPERTY_PRODUCER_GROUP = "PGROUP";
+const std::string MQMessageConst::PROPERTY_MIN_OFFSET = "MIN_OFFSET";
+const std::string MQMessageConst::PROPERTY_MAX_OFFSET = "MAX_OFFSET";
+
+const std::string MQMessageConst::PROPERTY_BUYER_ID = "BUYER_ID";
+const std::string MQMessageConst::PROPERTY_ORIGIN_MESSAGE_ID = "ORIGIN_MESSAGE_ID";
+const std::string MQMessageConst::PROPERTY_TRANSFER_FLAG = "TRANSFER_FLAG";
+const std::string MQMessageConst::PROPERTY_CORRECTION_FLAG = "CORRECTION_FLAG";
+const std::string MQMessageConst::PROPERTY_MQ2_FLAG = "MQ2_FLAG";
+const std::string MQMessageConst::PROPERTY_RECONSUME_TIME = "RECONSUME_TIME";
+const std::string MQMessageConst::PROPERTY_MSG_REGION = "MSG_REGION";
+const std::string MQMessageConst::PROPERTY_TRACE_SWITCH = "TRACE_ON";
+const std::string MQMessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX = "UNIQ_KEY";
+const std::string MQMessageConst::PROPERTY_MAX_RECONSUME_TIMES = "MAX_RECONSUME_TIMES";
+const std::string MQMessageConst::PROPERTY_CONSUME_START_TIMESTAMP = "CONSUME_START_TIME";
+const std::string MQMessageConst::PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET = "TRAN_PREPARED_QUEUE_OFFSET";
+const std::string MQMessageConst::PROPERTY_TRANSACTION_CHECK_TIMES = "TRANSACTION_CHECK_TIMES";
+const std::string MQMessageConst::PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS = "CHECK_IMMUNITY_TIME_IN_SECONDS";
+const std::string MQMessageConst::PROPERTY_INSTANCE_ID = "INSTANCE_ID";
+const std::string MQMessageConst::PROPERTY_CORRELATION_ID = "CORRELATION_ID";
+const std::string MQMessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT = "REPLY_TO_CLIENT";
+const std::string MQMessageConst::PROPERTY_MESSAGE_TTL = "TTL";
+const std::string MQMessageConst::PROPERTY_REPLY_MESSAGE_ARRIVE_TIME = "ARRIVE_TIME";
+const std::string MQMessageConst::PROPERTY_PUSH_REPLY_TIME = "PUSH_REPLY_TIME";
+const std::string MQMessageConst::PROPERTY_CLUSTER = "CLUSTER";
+const std::string MQMessageConst::PROPERTY_MESSAGE_TYPE = "MSG_TYPE";
+
+const std::string MQMessageConst::PROPERTY_ALREADY_COMPRESSED_FLAG = "__ALREADY_COMPRESSED__";
+
+const std::string MQMessageConst::KEY_SEPARATOR = " ";
+
+} // namespace rocketmq
diff --git a/src/message/MQMessageExt.cpp b/src/message/MQMessageExt.cpp
index e76a844..09b7e2c 100644
--- a/src/message/MQMessageExt.cpp
+++ b/src/message/MQMessageExt.cpp
@@ -16,184 +16,133 @@
*/
#include "MQMessageExt.h"
-#include <cstring>
-
-#include "MessageClientIDSetter.h"
-#include "MessageSysFlag.h"
-#include "SocketUtil.h"
-#include "TopicFilterType.h"
+#include "MessageExtImpl.h"
+#include "UtilAll.h" // null
namespace rocketmq {
-MQMessageExt::MQMessageExt() : MQMessageExt(0, 0, nullptr, 0, nullptr, "") {}
+MQMessageExt::MQMessageExt() : MQMessageExt(0, 0, nullptr, 0, nullptr, null) {}
MQMessageExt::MQMessageExt(int queueId,
int64_t bornTimestamp,
const struct sockaddr* bornHost,
int64_t storeTimestamp,
const struct sockaddr* storeHost,
- std::string msgId)
- : m_storeSize(0),
- m_bodyCRC(0),
- m_queueId(queueId),
- m_queueOffset(0),
- m_commitLogOffset(0),
- m_sysFlag(0),
- m_bornTimestamp(bornTimestamp),
- m_bornHost(nullptr),
- m_storeTimestamp(storeTimestamp),
- m_storeHost(nullptr),
- m_reconsumeTimes(3),
- m_preparedTransactionOffset(0),
- m_msgId(msgId) {
- m_bornHost = copySocketAddress(m_bornHost, bornHost);
- m_storeHost = copySocketAddress(m_storeHost, storeHost);
-}
-
-MQMessageExt::~MQMessageExt() {
- free(m_bornHost);
- free(m_storeHost);
-}
-
-TopicFilterType MQMessageExt::parseTopicFilterType(int32_t sysFlag) {
- if ((sysFlag & MessageSysFlag::MultiTagsFlag) == MessageSysFlag::MultiTagsFlag) {
- return MULTI_TAG;
- }
- return SINGLE_TAG;
-}
+ const std::string& msgId)
+ : MQMessage(std::make_shared<MessageExtImpl>(queueId, bornTimestamp, bornHost, storeTimestamp, storeHost, msgId)) {}
+
+MQMessageExt::~MQMessageExt() = default;
int32_t MQMessageExt::getStoreSize() const {
- return m_storeSize;
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getStoreSize();
}
void MQMessageExt::setStoreSize(int32_t storeSize) {
- m_storeSize = storeSize;
+ std::dynamic_pointer_cast<MessageExt>(message_impl_)->setStoreSize(storeSize);
}
int32_t MQMessageExt::getBodyCRC() const {
- return m_bodyCRC;
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getBodyCRC();
}
void MQMessageExt::setBodyCRC(int32_t bodyCRC) {
- m_bodyCRC = bodyCRC;
+ std::dynamic_pointer_cast<MessageExt>(message_impl_)->setBodyCRC(bodyCRC);
}
int32_t MQMessageExt::getQueueId() const {
- return m_queueId;
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getQueueId();
}
void MQMessageExt::setQueueId(int32_t queueId) {
- m_queueId = queueId;
+ std::dynamic_pointer_cast<MessageExt>(message_impl_)->setQueueId(queueId);
}
int64_t MQMessageExt::getQueueOffset() const {
- return m_queueOffset;
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getQueueOffset();
}
void MQMessageExt::setQueueOffset(int64_t queueOffset) {
- m_queueOffset = queueOffset;
+ std::dynamic_pointer_cast<MessageExt>(message_impl_)->setQueueOffset(queueOffset);
}
int64_t MQMessageExt::getCommitLogOffset() const {
- return m_commitLogOffset;
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getCommitLogOffset();
}
void MQMessageExt::setCommitLogOffset(int64_t physicOffset) {
- m_commitLogOffset = physicOffset;
+ std::dynamic_pointer_cast<MessageExt>(message_impl_)->setCommitLogOffset(physicOffset);
}
int32_t MQMessageExt::getSysFlag() const {
- return m_sysFlag;
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getSysFlag();
}
void MQMessageExt::setSysFlag(int32_t sysFlag) {
- m_sysFlag = sysFlag;
+ std::dynamic_pointer_cast<MessageExt>(message_impl_)->setSysFlag(sysFlag);
}
int64_t MQMessageExt::getBornTimestamp() const {
- return m_bornTimestamp;
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getBornTimestamp();
}
void MQMessageExt::setBornTimestamp(int64_t bornTimestamp) {
- m_bornTimestamp = bornTimestamp;
+ std::dynamic_pointer_cast<MessageExt>(message_impl_)->setBornTimestamp(bornTimestamp);
}
-const struct sockaddr* MQMessageExt::getBornHost() const {
- return m_bornHost;
+std::string MQMessageExt::getBornHostString() const {
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getBornHostString();
}
-std::string MQMessageExt::getBornHostString() const {
- return socketAddress2String(m_bornHost);
+const struct sockaddr* MQMessageExt::getBornHost() const {
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getBornHost();
}
void MQMessageExt::setBornHost(const struct sockaddr* bornHost) {
- m_bornHost = copySocketAddress(m_bornHost, bornHost);
+ std::dynamic_pointer_cast<MessageExt>(message_impl_)->setBornHost(bornHost);
}
int64_t MQMessageExt::getStoreTimestamp() const {
- return m_storeTimestamp;
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getStoreTimestamp();
}
void MQMessageExt::setStoreTimestamp(int64_t storeTimestamp) {
- m_storeTimestamp = storeTimestamp;
-}
-
-const struct sockaddr* MQMessageExt::getStoreHost() const {
- return m_storeHost;
+ std::dynamic_pointer_cast<MessageExt>(message_impl_)->setStoreTimestamp(storeTimestamp);
}
std::string MQMessageExt::getStoreHostString() const {
- return socketAddress2String(m_storeHost);
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getStoreHostString();
}
-void MQMessageExt::setStoreHost(const struct sockaddr* storeHost) {
- m_storeHost = copySocketAddress(m_storeHost, storeHost);
-}
-
-const std::string& MQMessageExt::getMsgId() const {
- return m_msgId;
+const struct sockaddr* MQMessageExt::getStoreHost() const {
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getStoreHost();
}
-void MQMessageExt::setMsgId(const std::string& msgId) {
- m_msgId = msgId;
+void MQMessageExt::setStoreHost(const struct sockaddr* storeHost) {
+ std::dynamic_pointer_cast<MessageExt>(message_impl_)->setStoreHost(storeHost);
}
int32_t MQMessageExt::getReconsumeTimes() const {
- return m_reconsumeTimes;
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getReconsumeTimes();
}
void MQMessageExt::setReconsumeTimes(int32_t reconsumeTimes) {
- m_reconsumeTimes = reconsumeTimes;
+ std::dynamic_pointer_cast<MessageExt>(message_impl_)->setReconsumeTimes(reconsumeTimes);
}
int64_t MQMessageExt::getPreparedTransactionOffset() const {
- return m_preparedTransactionOffset;
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getPreparedTransactionOffset();
}
void MQMessageExt::setPreparedTransactionOffset(int64_t preparedTransactionOffset) {
- m_preparedTransactionOffset = preparedTransactionOffset;
+ std::dynamic_pointer_cast<MessageExt>(message_impl_)->setPreparedTransactionOffset(preparedTransactionOffset);
}
-const std::string& MQMessageClientExt::getOffsetMsgId() const {
- return MQMessageExt::getMsgId();
-}
-
-void MQMessageClientExt::setOffsetMsgId(const std::string& offsetMsgId) {
- return MQMessageExt::setMsgId(offsetMsgId);
-}
-
-const std::string& MQMessageClientExt::getMsgId() const {
- const auto& uniqID = MessageClientIDSetter::getUniqID(*this);
- if (uniqID.empty()) {
- return getOffsetMsgId();
- } else {
- return uniqID;
- }
+const std::string& MQMessageExt::getMsgId() const {
+ return std::dynamic_pointer_cast<MessageExt>(message_impl_)->getMsgId();
}
-void MQMessageClientExt::setMsgId(const std::string& msgId) {
- // DO NOTHING
- // MessageClientIDSetter::setUniqID(*this);
+void MQMessageExt::setMsgId(const std::string& msgId) {
+ std::dynamic_pointer_cast<MessageExt>(message_impl_)->setMsgId(msgId);
}
} // namespace rocketmq
diff --git a/src/message/MQMessageId.h b/src/message/MQMessageId.h
deleted file mode 100644
index 2e31e78..0000000
--- a/src/message/MQMessageId.h
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * 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.
- */
-#ifndef __MESSAGE_ID_H__
-#define __MESSAGE_ID_H__
-
-#include "SocketUtil.h"
-#include "UtilAll.h"
-
-namespace rocketmq {
-
-class MQMessageId {
- public:
- MQMessageId() : MQMessageId(nullptr, 0) {}
- MQMessageId(struct sockaddr* address, int64_t offset) : m_address(nullptr), m_offset(offset) { setAddress(address); }
- ~MQMessageId() { free(m_address); }
-
- MQMessageId& operator=(const MQMessageId& id) {
- if (&id == this) {
- return *this;
- }
- setAddress(id.m_address);
- this->m_offset = id.m_offset;
- return *this;
- }
-
- const struct sockaddr* getAddress() const { return m_address; }
-
- void setAddress(struct sockaddr* address) { m_address = copySocketAddress(m_address, address); }
-
- int64_t getOffset() const { return m_offset; }
-
- void setOffset(int64_t offset) { m_offset = offset; }
-
- private:
- struct sockaddr* m_address;
- int64_t m_offset;
-};
-
-} // namespace rocketmq
-
-#endif // __MESSAGE_ID_H__
diff --git a/src/message/MessageAccessor.h b/src/message/MessageAccessor.h
index c9bf2bd..ea6f6a2 100644
--- a/src/message/MessageAccessor.h
+++ b/src/message/MessageAccessor.h
@@ -17,31 +17,33 @@
#ifndef __MESSAGE_ACCESSOR_H__
#define __MESSAGE_ACCESSOR_H__
-#include "MQMessage.h"
+#include <algorithm>
+
+#include "Message.h"
namespace rocketmq {
class MessageAccessor {
public:
- static inline void clearProperty(MQMessage& msg, const std::string& name) { msg.clearProperty(name); }
+ static inline void clearProperty(Message& msg, const std::string& name) { msg.clearProperty(name); }
- static inline void setProperties(MQMessage& msg, std::map<std::string, std::string>&& properties) {
- msg.setProperties(std::forward<std::map<std::string, std::string>>(properties));
+ static inline void setProperties(Message& msg, std::map<std::string, std::string>&& properties) {
+ msg.setProperties(std::move(properties));
}
- static inline void putProperty(MQMessage& msg, const std::string& name, const std::string& value) {
+ static inline void putProperty(Message& msg, const std::string& name, const std::string& value) {
msg.putProperty(name, value);
}
- static inline const std::string& getReconsumeTime(MQMessage& msg) {
+ static inline const std::string& getReconsumeTime(Message& msg) {
return msg.getProperty(MQMessageConst::PROPERTY_RECONSUME_TIME);
}
- static inline const std::string& getMaxReconsumeTimes(MQMessage& msg) {
+ static inline const std::string& getMaxReconsumeTimes(Message& msg) {
return msg.getProperty(MQMessageConst::PROPERTY_MAX_RECONSUME_TIMES);
}
- static inline void setConsumeStartTimeStamp(MQMessage& msg, const std::string& propertyConsumeStartTimeStamp) {
+ static inline void setConsumeStartTimeStamp(Message& msg, const std::string& propertyConsumeStartTimeStamp) {
putProperty(msg, MQMessageConst::PROPERTY_CONSUME_START_TIMESTAMP, propertyConsumeStartTimeStamp);
}
};
diff --git a/src/message/MessageBatch.cpp b/src/message/MessageBatch.cpp
index 40ecee3..ab49b74 100644
--- a/src/message/MessageBatch.cpp
+++ b/src/message/MessageBatch.cpp
@@ -16,48 +16,47 @@
*/
#include "MessageBatch.h"
-#include <memory>
-
-#include "MQDecoder.h"
+#include "MessageDecoder.h"
#include "MessageClientIDSetter.h"
namespace rocketmq {
-std::string MessageBatch::encode() {
- return MQDecoder::encodeMessages(m_messages);
-}
-
-MessageBatch* MessageBatch::generateFromList(std::vector<MQMessagePtr>& messages) {
- bool isFirst = true;
+std::shared_ptr<MessageBatch> MessageBatch::generateFromList(std::vector<MQMessage>& messages) {
+ bool is_first = true;
std::string topic;
- bool waitStoreMsgOK = false;
+ bool wait_store_msg_ok = true;
+ // check messages
for (auto& message : messages) {
- if (message->getDelayTimeLevel() > 0) {
+ if (message.getDelayTimeLevel() > 0) {
THROW_MQEXCEPTION(MQClientException, "TimeDelayLevel in not supported for batching", -1);
}
- if (isFirst) {
- isFirst = false;
- topic = message->getTopic();
- waitStoreMsgOK = message->isWaitStoreMsgOK();
+ if (is_first) {
+ is_first = false;
+ topic = message.getTopic();
+ wait_store_msg_ok = message.isWaitStoreMsgOK();
if (UtilAll::isRetryTopic(topic)) {
THROW_MQEXCEPTION(MQClientException, "Retry Group is not supported for batching", -1);
}
} else {
- if (message->getTopic() != topic) {
+ if (message.getTopic() != topic) {
THROW_MQEXCEPTION(MQClientException, "The topic of the messages in one batch should be the same", -1);
}
- if (message->isWaitStoreMsgOK() != waitStoreMsgOK) {
+ if (message.isWaitStoreMsgOK() != wait_store_msg_ok) {
THROW_MQEXCEPTION(MQClientException, "The waitStoreMsgOK of the messages in one batch should the same", -2);
}
}
}
- MessageBatch* batchMessage = new MessageBatch(messages);
+ auto batchMessage = std::make_shared<MessageBatch>(messages);
batchMessage->setTopic(topic);
- batchMessage->setWaitStoreMsgOK(waitStoreMsgOK);
+ batchMessage->setWaitStoreMsgOK(wait_store_msg_ok);
return batchMessage;
}
+std::string MessageBatch::encode() {
+ return MessageDecoder::encodeMessages(m_messages);
+}
+
} // namespace rocketmq
diff --git a/src/message/MessageBatch.h b/src/message/MessageBatch.h
index 644d287..1495f72 100644
--- a/src/message/MessageBatch.h
+++ b/src/message/MessageBatch.h
@@ -17,28 +17,28 @@
#ifndef __MESSAGE_BATCH_H__
#define __MESSAGE_BATCH_H__
-#include <memory>
-#include <string>
-
#include "MQMessage.h"
+#include "MessageImpl.h"
namespace rocketmq {
-class MessageBatch : public MQMessage {
+class MessageBatch : public MessageImpl {
public:
- static MessageBatch* generateFromList(std::vector<MQMessagePtr>& messages);
+ static std::shared_ptr<MessageBatch> generateFromList(std::vector<MQMessage>& messages);
public:
- MessageBatch(std::vector<MQMessagePtr>& messages) : m_messages(messages) {}
+ MessageBatch(std::vector<MQMessage>& messages) : MessageImpl(), m_messages(messages) {}
- std::string encode();
+ public: // Message
+ bool isBatch() const override final { return true; }
- const std::vector<MQMessagePtr>& getMessages() const { return m_messages; }
+ public:
+ std::string encode();
- bool isBatch() override final { return true; }
+ const std::vector<MQMessage>& getMessages() const { return m_messages; }
protected:
- std::vector<MQMessagePtr> m_messages;
+ std::vector<MQMessage> m_messages;
};
} // namespace rocketmq
diff --git a/src/message/MessageClientIDSetter.h b/src/message/MessageClientIDSetter.h
index 9af959e..90f2ec5 100644
--- a/src/message/MessageClientIDSetter.h
+++ b/src/message/MessageClientIDSetter.h
@@ -43,13 +43,13 @@ class MessageClientIDSetter {
*/
static std::string createUniqID() { return getInstance().createUniqueID(); }
- static void setUniqID(MQMessage& msg) {
+ static void setUniqID(Message& msg) {
if (msg.getProperty(MQMessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX).empty()) {
MessageAccessor::putProperty(msg, MQMessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, createUniqID());
}
}
- static const std::string& getUniqID(const MQMessage& msg) {
+ static const std::string& getUniqID(const Message& msg) {
return msg.getProperty(MQMessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
}
diff --git a/src/message/MQDecoder.cpp b/src/message/MessageDecoder.cpp
similarity index 66%
rename from src/message/MQDecoder.cpp
rename to src/message/MessageDecoder.cpp
index 863fa41..e7e4e59 100644
--- a/src/message/MQDecoder.cpp
+++ b/src/message/MessageDecoder.cpp
@@ -14,83 +14,96 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include "MQDecoder.h"
+#include "MessageDecoder.h"
-#include <sstream>
+#include <cstring> // std::memcpy
+
+#include <algorithm> // std::move
+#include <sstream> // std::stringstream
#ifndef WIN32
-#include <netinet/in.h>
+#include <arpa/inet.h> // htons, htonl
+#include <netinet/in.h> // struct sockaddr, sockaddr_in, sockaddr_in6
#endif
#include "ByteOrder.h"
#include "Logging.h"
#include "MemoryOutputStream.h"
+#include "MessageExtImpl.h"
#include "MessageAccessor.h"
#include "MessageSysFlag.h"
#include "UtilAll.h"
namespace rocketmq {
-const int MQDecoder::MSG_ID_LENGTH = 8 + 8;
-
-const char MQDecoder::NAME_VALUE_SEPARATOR = 1;
-const char MQDecoder::PROPERTY_SEPARATOR = 2;
+const int MessageDecoder::MSG_ID_LENGTH = 8 + 8;
-int MQDecoder::MessageMagicCodePostion = 4;
-int MQDecoder::MessageFlagPostion = 16;
-int MQDecoder::MessagePhysicOffsetPostion = 28;
-int MQDecoder::MessageStoreTimestampPostion = 56;
+const char MessageDecoder::NAME_VALUE_SEPARATOR = 1;
+const char MessageDecoder::PROPERTY_SEPARATOR = 2;
-std::string MQDecoder::createMessageId(const struct sockaddr* sa, int64_t offset) {
- const struct sockaddr_in* sin = (struct sockaddr_in*)sa;
+int MessageDecoder::MessageMagicCodePostion = 4;
+int MessageDecoder::MessageFlagPostion = 16;
+int MessageDecoder::MessagePhysicOffsetPostion = 28;
+int MessageDecoder::MessageStoreTimestampPostion = 56;
- MemoryOutputStream outputmen(MSG_ID_LENGTH);
- outputmen.writeIntBigEndian(sin->sin_addr.s_addr);
- outputmen.writeRepeatedByte(0, 2);
- outputmen.write(&(sin->sin_port), 2);
- outputmen.writeInt64BigEndian(offset);
+std::string MessageDecoder::createMessageId(const struct sockaddr* sa, int64_t offset) {
+ MemoryOutputStream mos(sa->sa_family == AF_INET ? 16 : 28);
+ if (sa->sa_family == AF_INET) {
+ const struct sockaddr_in* sin = (struct sockaddr_in*)sa;
+ mos.write(&sin->sin_addr.s_addr, 4);
+ mos.writeRepeatedByte(0, 2);
+ mos.write(&sin->sin_port, 2);
+ mos.writeInt64BigEndian(offset);
+ } else {
+ const struct sockaddr_in6* sin6 = (struct sockaddr_in6*)sa;
+ mos.write(&sin6->sin6_addr, 16);
+ mos.writeRepeatedByte(0, 2);
+ mos.write(&sin6->sin6_port, 2);
+ mos.writeInt64BigEndian(offset);
+ }
- const char* bytes = static_cast<const char*>(outputmen.getData());
- int len = outputmen.getDataSize();
+ const char* bytes = static_cast<const char*>(mos.getData());
+ int len = mos.getDataSize();
return UtilAll::bytes2string(bytes, len);
}
-MQMessageId MQDecoder::decodeMessageId(const std::string& msgId) {
- std::string ipStr = msgId.substr(0, 8);
- std::string portStr = msgId.substr(8, 8);
- std::string offsetStr = msgId.substr(16);
-
- size_t pos;
- int ipInt = std::stoul(ipStr, &pos, 16);
- int portInt = std::stoul(portStr, &pos, 16);
+MessageId MessageDecoder::decodeMessageId(const std::string& msgId) {
+ size_t ip_length = msgId.length() == 32 ? 4 * 2 : 16 * 2;
- uint64_t offset = UtilAll::hexstr2ull(offsetStr.c_str());
+ std::string ip = msgId.substr(0, ip_length);
+ std::string port = msgId.substr(ip_length, 8);
+ std::string offset = msgId.substr(ip_length + 8);
- struct sockaddr_in sin;
- sin.sin_family = AF_INET;
- sin.sin_port = htons(portInt);
- sin.sin_addr.s_addr = htonl(ipInt);
+ uint16_t portInt = (uint16_t)std::stoul(port, nullptr, 16);
+ uint64_t offsetInt = std::stoull(offset, nullptr, 16);
- return MQMessageId((struct sockaddr*)&sin, offset);
+ if (ip_length == 32) {
+ int32_t ipInt = std::stoul(ip, nullptr, 16);
+ struct sockaddr_in sin;
+ sin.sin_family = AF_INET;
+ sin.sin_port = htons(portInt);
+ sin.sin_addr.s_addr = htonl(ipInt);
+ return MessageId((struct sockaddr*)&sin, offsetInt);
+ } else {
+ struct sockaddr_in6 sin6;
+ sin6.sin6_family = AF_INET6;
+ sin6.sin6_port = htons(portInt);
+ UtilAll::string2bytes((char*)&sin6.sin6_addr, ip);
+ return MessageId((struct sockaddr*)&sin6, offsetInt);
+ }
}
-MQMessageExtPtr MQDecoder::clientDecode(MemoryInputStream& byteBuffer, bool readBody) {
+MessageExtPtr MessageDecoder::clientDecode(MemoryInputStream& byteBuffer, bool readBody) {
return decode(byteBuffer, readBody, true, true);
}
-MQMessageExtPtr MQDecoder::decode(MemoryInputStream& byteBuffer, bool readBody) {
+MessageExtPtr MessageDecoder::decode(MemoryInputStream& byteBuffer, bool readBody) {
return decode(byteBuffer, readBody, true, false);
}
-MQMessageExtPtr MQDecoder::decode(MemoryInputStream& byteBuffer, bool readBody, bool deCompressBody, bool isClient) {
- MQMessageExtPtr msgExt;
-
- if (isClient) {
- msgExt = new MQMessageClientExt();
- } else {
- msgExt = new MQMessageExt();
- }
+MessageExtPtr MessageDecoder::decode(MemoryInputStream& byteBuffer, bool readBody, bool deCompressBody, bool isClient) {
+ auto msgExt = isClient ? std::make_shared<MessageClientExtImpl>() : std::make_shared<MessageExtImpl>();
// 1 TOTALSIZE
int32_t storeSize = byteBuffer.readIntBigEndian();
@@ -194,7 +207,7 @@ MQMessageExtPtr MQDecoder::decode(MemoryInputStream& byteBuffer, bool readBody,
// 18 msg ID
std::string msgId = createMessageId(msgExt->getStoreHost(), (int64_t)msgExt->getCommitLogOffset());
- msgExt->MQMessageExt::setMsgId(msgId);
+ msgExt->MessageExtImpl::setMsgId(msgId);
if (uncompress_failed) {
LOG_WARN_NEW("can not uncompress message, id:{}", msgExt->getMsgId());
@@ -203,35 +216,33 @@ MQMessageExtPtr MQDecoder::decode(MemoryInputStream& byteBuffer, bool readBody,
return msgExt;
}
-MQMessageExtPtr2 MQDecoder::decode(MemoryBlock& mem) {
+MessageExtPtr MessageDecoder::decode(MemoryBlock& mem) {
return decode(mem, true);
}
-MQMessageExtPtr2 MQDecoder::decode(MemoryBlock& mem, bool readBody) {
+MessageExtPtr MessageDecoder::decode(MemoryBlock& mem, bool readBody) {
MemoryInputStream rawInput(mem, true);
- MQMessageExtPtr2 message(decode(rawInput, readBody));
- return message;
+ return decode(rawInput, readBody);
}
-std::vector<MQMessageExtPtr2> MQDecoder::decodes(MemoryBlock& mem) {
+std::vector<MessageExtPtr> MessageDecoder::decodes(MemoryBlock& mem) {
return decodes(mem, true);
}
-std::vector<MQMessageExtPtr2> MQDecoder::decodes(MemoryBlock& mem, bool readBody) {
- std::vector<MQMessageExtPtr2> msgExts;
+std::vector<MessageExtPtr> MessageDecoder::decodes(MemoryBlock& mem, bool readBody) {
+ std::vector<MessageExtPtr> msgExts;
MemoryInputStream rawInput(mem, true);
while (rawInput.getNumBytesRemaining() > 0) {
- auto* msgExt = clientDecode(rawInput, readBody);
- if (msgExt != nullptr) {
- msgExts.emplace_back(msgExt);
- } else {
+ auto msgExt = clientDecode(rawInput, readBody);
+ if (nullptr == msgExt) {
break;
}
+ msgExts.emplace_back(msgExt);
}
return msgExts;
}
-std::string MQDecoder::messageProperties2String(const std::map<std::string, std::string>& properties) {
+std::string MessageDecoder::messageProperties2String(const std::map<std::string, std::string>& properties) {
std::string os;
for (const auto& it : properties) {
@@ -252,7 +263,7 @@ std::string MQDecoder::messageProperties2String(const std::map<std::string, std:
return os;
}
-std::map<std::string, std::string> MQDecoder::string2messageProperties(const std::string& properties) {
+std::map<std::string, std::string> MessageDecoder::string2messageProperties(const std::string& properties) {
std::vector<std::string> out;
UtilAll::Split(out, properties, PROPERTY_SEPARATOR);
@@ -268,10 +279,10 @@ std::map<std::string, std::string> MQDecoder::string2messageProperties(const std
return map;
}
-std::string MQDecoder::encodeMessage(MQMessage& message) {
+std::string MessageDecoder::encodeMessage(Message& message) {
const std::string& body = message.getBody();
uint32_t bodyLen = body.length();
- std::string properties = MQDecoder::messageProperties2String(message.getProperties());
+ std::string properties = MessageDecoder::messageProperties2String(message.getProperties());
uint16_t propertiesLength = (int16_t)properties.length();
uint32_t storeSize = 4 // 1 TOTALSIZE
+ 4 // 2 MAGICCODE
@@ -315,10 +326,10 @@ std::string MQDecoder::encodeMessage(MQMessage& message) {
return encodeMsg;
}
-std::string MQDecoder::encodeMessages(std::vector<MQMessagePtr>& msgs) {
+std::string MessageDecoder::encodeMessages(std::vector<MQMessage>& msgs) {
std::string encodedBody;
- for (auto* message : msgs) {
- encodedBody.append(encodeMessage(*message));
+ for (auto msg : msgs) {
+ encodedBody.append(encodeMessage(msg));
}
return encodedBody;
}
diff --git a/src/message/MQDecoder.h b/src/message/MessageDecoder.h
similarity index 62%
rename from src/message/MQDecoder.h
rename to src/message/MessageDecoder.h
index 8b9eea3..2bc935a 100644
--- a/src/message/MQDecoder.h
+++ b/src/message/MessageDecoder.h
@@ -14,46 +14,44 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef __MESSAGE_DECODER_H__
-#define __MESSAGE_DECODER_H__
-
-#include <map>
-#include <memory>
-#include <string>
+#ifndef ROCKETMQ_MESSAGE_MESSAGEDECODER_H_
+#define ROCKETMQ_MESSAGE_MESSAGEDECODER_H_
#include "MQClientException.h"
#include "MQMessageExt.h"
-#include "MQMessageId.h"
+#include "MessageId.h"
#include "MemoryInputStream.h"
namespace rocketmq {
-class MQDecoder {
+class MessageDecoder {
public:
static std::string createMessageId(const struct sockaddr* sa, int64_t offset);
- static MQMessageId decodeMessageId(const std::string& msgId);
+ static MessageId decodeMessageId(const std::string& msgId);
- static MQMessageExtPtr2 decode(MemoryBlock& mem);
- static MQMessageExtPtr2 decode(MemoryBlock& mem, bool readBody);
+ static MessageExtPtr decode(MemoryBlock& mem);
+ static MessageExtPtr decode(MemoryBlock& mem, bool readBody);
- static std::vector<MQMessageExtPtr2> decodes(MemoryBlock& mem);
- static std::vector<MQMessageExtPtr2> decodes(MemoryBlock& mem, bool readBody);
+ static std::vector<MessageExtPtr> decodes(MemoryBlock& mem);
+ static std::vector<MessageExtPtr> decodes(MemoryBlock& mem, bool readBody);
static std::string messageProperties2String(const std::map<std::string, std::string>& properties);
static std::map<std::string, std::string> string2messageProperties(const std::string& properties);
- static std::string encodeMessage(MQMessage& message);
- static std::string encodeMessages(std::vector<MQMessagePtr>& msgs);
+ static std::string encodeMessage(Message& message);
+ static std::string encodeMessages(std::vector<MQMessage>& msgs);
private:
- static MQMessageExtPtr clientDecode(MemoryInputStream& byteBuffer, bool readBody);
- static MQMessageExtPtr decode(MemoryInputStream& byteBuffer, bool readBody);
- static MQMessageExtPtr decode(MemoryInputStream& byteBuffer, bool readBody, bool deCompressBody, bool isClient);
+ static MessageExtPtr clientDecode(MemoryInputStream& byteBuffer, bool readBody);
+
+ static MessageExtPtr decode(MemoryInputStream& byteBuffer, bool readBody);
+ static MessageExtPtr decode(MemoryInputStream& byteBuffer, bool readBody, bool deCompressBody, bool isClient);
public:
static const char NAME_VALUE_SEPARATOR;
static const char PROPERTY_SEPARATOR;
static const int MSG_ID_LENGTH;
+
static int MessageMagicCodePostion;
static int MessageFlagPostion;
static int MessagePhysicOffsetPostion;
@@ -62,4 +60,4 @@ class MQDecoder {
} // namespace rocketmq
-#endif // __MESSAGE_DECODER_H__
+#endif // ROCKETMQ_MESSAGE_MESSAGEDECODER_H_
diff --git a/src/message/MessageExtImpl.cpp b/src/message/MessageExtImpl.cpp
new file mode 100644
index 0000000..ea36b35
--- /dev/null
+++ b/src/message/MessageExtImpl.cpp
@@ -0,0 +1,214 @@
+/*
+ * 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 "MessageExtImpl.h"
+
+#include <sstream> // std::stringstream
+
+#include "MessageClientIDSetter.h"
+#include "MessageSysFlag.h"
+#include "SocketUtil.h"
+#include "UtilAll.h"
+
+namespace rocketmq {
+
+// ============================
+// MessageExtImpl
+// ============================
+
+MessageExtImpl::MessageExtImpl() : MessageExtImpl(0, 0, nullptr, 0, nullptr, null) {}
+
+MessageExtImpl::MessageExtImpl(int queueId,
+ int64_t bornTimestamp,
+ const struct sockaddr* bornHost,
+ int64_t storeTimestamp,
+ const struct sockaddr* storeHost,
+ const std::string& msgId)
+ : store_size_(0),
+ body_crc_(0),
+ queue_id_(queueId),
+ queue_offset_(0),
+ commit_log_offset_(0),
+ sys_flag_(0),
+ born_timestamp_(bornTimestamp),
+ born_host_(nullptr),
+ store_timestamp_(storeTimestamp),
+ store_host_(nullptr),
+ reconsume_times_(3),
+ prepared_transaction_offset_(0),
+ msg_id_(msgId) {
+ born_host_ = copySocketAddress(born_host_, bornHost);
+ store_host_ = copySocketAddress(store_host_, storeHost);
+}
+
+MessageExtImpl::~MessageExtImpl() {
+ free(born_host_);
+ free(store_host_);
+}
+
+TopicFilterType MessageExtImpl::parseTopicFilterType(int32_t sysFlag) {
+ if ((sysFlag & MessageSysFlag::MultiTagsFlag) == MessageSysFlag::MultiTagsFlag) {
+ return MULTI_TAG;
+ }
+ return SINGLE_TAG;
+}
+
+int32_t MessageExtImpl::getStoreSize() const {
+ return store_size_;
+}
+
+void MessageExtImpl::setStoreSize(int32_t storeSize) {
+ store_size_ = storeSize;
+}
+
+int32_t MessageExtImpl::getBodyCRC() const {
+ return body_crc_;
+}
+
+void MessageExtImpl::setBodyCRC(int32_t bodyCRC) {
+ body_crc_ = bodyCRC;
+}
+
+int32_t MessageExtImpl::getQueueId() const {
+ return queue_id_;
+}
+
+void MessageExtImpl::setQueueId(int32_t queueId) {
+ queue_id_ = queueId;
+}
+
+int64_t MessageExtImpl::getQueueOffset() const {
+ return queue_offset_;
+}
+
+void MessageExtImpl::setQueueOffset(int64_t queueOffset) {
+ queue_offset_ = queueOffset;
+}
+
+int64_t MessageExtImpl::getCommitLogOffset() const {
+ return commit_log_offset_;
+}
+
+void MessageExtImpl::setCommitLogOffset(int64_t physicOffset) {
+ commit_log_offset_ = physicOffset;
+}
+
+int32_t MessageExtImpl::getSysFlag() const {
+ return sys_flag_;
+}
+
+void MessageExtImpl::setSysFlag(int32_t sysFlag) {
+ sys_flag_ = sysFlag;
+}
+
+int64_t MessageExtImpl::getBornTimestamp() const {
+ return born_timestamp_;
+}
+
+void MessageExtImpl::setBornTimestamp(int64_t bornTimestamp) {
+ born_timestamp_ = bornTimestamp;
+}
+
+const struct sockaddr* MessageExtImpl::getBornHost() const {
+ return born_host_;
+}
+
+std::string MessageExtImpl::getBornHostString() const {
+ return socketAddress2String(born_host_);
+}
+
+void MessageExtImpl::setBornHost(const struct sockaddr* bornHost) {
+ born_host_ = copySocketAddress(born_host_, bornHost);
+}
+
+int64_t MessageExtImpl::getStoreTimestamp() const {
+ return store_timestamp_;
+}
+
+void MessageExtImpl::setStoreTimestamp(int64_t storeTimestamp) {
+ store_timestamp_ = storeTimestamp;
+}
+
+const struct sockaddr* MessageExtImpl::getStoreHost() const {
+ return store_host_;
+}
+
+std::string MessageExtImpl::getStoreHostString() const {
+ return socketAddress2String(store_host_);
+}
+
+void MessageExtImpl::setStoreHost(const struct sockaddr* storeHost) {
+ store_host_ = copySocketAddress(store_host_, storeHost);
+}
+
+const std::string& MessageExtImpl::getMsgId() const {
+ return msg_id_;
+}
+
+void MessageExtImpl::setMsgId(const std::string& msgId) {
+ msg_id_ = msgId;
+}
+
+int32_t MessageExtImpl::getReconsumeTimes() const {
+ return reconsume_times_;
+}
+
+void MessageExtImpl::setReconsumeTimes(int32_t reconsumeTimes) {
+ reconsume_times_ = reconsumeTimes;
+}
+
+int64_t MessageExtImpl::getPreparedTransactionOffset() const {
+ return prepared_transaction_offset_;
+}
+
+void MessageExtImpl::setPreparedTransactionOffset(int64_t preparedTransactionOffset) {
+ prepared_transaction_offset_ = preparedTransactionOffset;
+}
+
+std::string MessageExtImpl::toString() const {
+ std::stringstream ss;
+ ss << "MessageExt [queueId=" << queue_id_ << ", storeSize=" << store_size_ << ", queueOffset=" << queue_offset_
+ << ", sysFlag=" << sys_flag_ << ", bornTimestamp=" << born_timestamp_ << ", bornHost=" << getBornHostString()
+ << ", storeTimestamp=" << store_timestamp_ << ", storeHost=" << getStoreHostString() << ", msgId=" << getMsgId()
+ << ", commitLogOffset=" << commit_log_offset_ << ", bodyCRC=" << body_crc_
+ << ", reconsumeTimes=" << reconsume_times_ << ", preparedTransactionOffset=" << prepared_transaction_offset_
+ << ", toString()=" << MessageImpl::toString() << "]";
+ return ss.str();
+}
+
+// ============================
+// MessageClientExtImpl
+// ============================
+
+const std::string& MessageClientExtImpl::getMsgId() const {
+ const auto& unique_id = MessageClientIDSetter::getUniqID(*this);
+ return unique_id.empty() ? getOffsetMsgId() : unique_id;
+}
+
+void MessageClientExtImpl::setMsgId(const std::string& msgId) {
+ // DO NOTHING
+ // MessageClientIDSetter::setUniqID(*this);
+}
+
+const std::string& MessageClientExtImpl::getOffsetMsgId() const {
+ return MessageExtImpl::getMsgId();
+}
+
+void MessageClientExtImpl::setOffsetMsgId(const std::string& offsetMsgId) {
+ return MessageExtImpl::setMsgId(offsetMsgId);
+}
+
+} // namespace rocketmq
diff --git a/src/message/MessageExtImpl.h b/src/message/MessageExtImpl.h
new file mode 100644
index 0000000..af9e1ff
--- /dev/null
+++ b/src/message/MessageExtImpl.h
@@ -0,0 +1,117 @@
+/*
+ * 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.
+ */
+#ifndef ROCKETMQ_MESSAGE_MESSAGEEXT_H_
+#define ROCKETMQ_MESSAGE_MESSAGEEXT_H_
+
+#include "MessageExt.h"
+#include "MessageImpl.h"
+#include "TopicFilterType.h"
+
+namespace rocketmq {
+
+/**
+ * MessageExtImpl - MessageExt implement
+ */
+class MessageExtImpl : public MessageImpl, // base
+ virtual public MessageExt // interface
+{
+ public:
+ MessageExtImpl();
+ MessageExtImpl(int queueId,
+ int64_t bornTimestamp,
+ const struct sockaddr* bornHost,
+ int64_t storeTimestamp,
+ const struct sockaddr* storeHost,
+ const std::string& msgId);
+
+ virtual ~MessageExtImpl();
+
+ static TopicFilterType parseTopicFilterType(int32_t sysFlag);
+
+ public: // MessageExt
+ int32_t getStoreSize() const override;
+ void setStoreSize(int32_t storeSize) override;
+
+ int32_t getBodyCRC() const override;
+ void setBodyCRC(int32_t bodyCRC) override;
+
+ int32_t getQueueId() const override;
+ void setQueueId(int32_t queueId) override;
+
+ int64_t getQueueOffset() const override;
+ void setQueueOffset(int64_t queueOffset) override;
+
+ int64_t getCommitLogOffset() const override;
+ void setCommitLogOffset(int64_t physicOffset) override;
+
+ int32_t getSysFlag() const override;
+ void setSysFlag(int32_t sysFlag) override;
+
+ int64_t getBornTimestamp() const override;
+ void setBornTimestamp(int64_t bornTimestamp) override;
+
+ const struct sockaddr* getBornHost() const override;
+ std::string getBornHostString() const override;
+ void setBornHost(const struct sockaddr* bornHost) override;
+
+ int64_t getStoreTimestamp() const override;
+ void setStoreTimestamp(int64_t storeTimestamp) override;
+
+ const struct sockaddr* getStoreHost() const override;
+ std::string getStoreHostString() const override;
+ void setStoreHost(const struct sockaddr* storeHost) override;
+
+ int32_t getReconsumeTimes() const override;
+ void setReconsumeTimes(int32_t reconsumeTimes) override;
+
+ int64_t getPreparedTransactionOffset() const override;
+ void setPreparedTransactionOffset(int64_t preparedTransactionOffset) override;
+
+ const std::string& getMsgId() const override;
+ void setMsgId(const std::string& msgId) override;
+
+ std::string toString() const override;
+
+ private:
+ int32_t store_size_;
+ int32_t body_crc_;
+ int32_t queue_id_;
+ int64_t queue_offset_;
+ int64_t commit_log_offset_;
+ int32_t sys_flag_;
+ int64_t born_timestamp_;
+ struct sockaddr* born_host_;
+ int64_t store_timestamp_;
+ struct sockaddr* store_host_;
+ int32_t reconsume_times_;
+ int64_t prepared_transaction_offset_;
+ std::string msg_id_;
+};
+
+class MessageClientExtImpl : public MessageExtImpl {
+ public: // MessageExt
+ const std::string& getMsgId() const override;
+ void setMsgId(const std::string& msgId) override;
+
+ public:
+ const std::string& getOffsetMsgId() const;
+ void setOffsetMsgId(const std::string& offsetMsgId);
+};
+
+} // namespace rocketmq
+
+#endif // ROCKETMQ_MESSAGE_MESSAGEEXT_H_
diff --git a/src/message/MessageId.h b/src/message/MessageId.h
new file mode 100644
index 0000000..64e4603
--- /dev/null
+++ b/src/message/MessageId.h
@@ -0,0 +1,58 @@
+/*
+ * 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.
+ */
+#ifndef ROCKETMQ_MESSAGE_MESSAGEID_H_
+#define ROCKETMQ_MESSAGE_MESSAGEID_H_
+
+#include <cstdlib>
+
+#include "SocketUtil.h"
+#include "UtilAll.h"
+
+namespace rocketmq {
+
+class MessageId {
+ public:
+ MessageId() : MessageId(nullptr, 0) {}
+ MessageId(struct sockaddr* address, int64_t offset) : address_(nullptr), offset_(offset) { setAddress(address); }
+
+ MessageId(const MessageId& other) : MessageId(other.address_, other.offset_) {}
+ MessageId(MessageId&& other) : address_(other.address_), offset_(other.offset_) { other.address_ = nullptr; }
+
+ virtual ~MessageId() { std::free(address_); }
+
+ MessageId& operator=(const MessageId& other) {
+ if (&other != this) {
+ setAddress(other.address_);
+ this->offset_ = other.offset_;
+ }
+ return *this;
+ }
+
+ const struct sockaddr* getAddress() const { return address_; }
+ void setAddress(struct sockaddr* address) { address_ = copySocketAddress(address_, address); }
+
+ int64_t getOffset() const { return offset_; }
+ void setOffset(int64_t offset) { offset_ = offset; }
+
+ private:
+ struct sockaddr* address_;
+ int64_t offset_;
+};
+
+} // namespace rocketmq
+
+#endif // ROCKETMQ_MESSAGE_MESSAGEID_H_
diff --git a/src/message/MessageImpl.cpp b/src/message/MessageImpl.cpp
new file mode 100644
index 0000000..f3f9def
--- /dev/null
+++ b/src/message/MessageImpl.cpp
@@ -0,0 +1,187 @@
+/*
+ * 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 "MessageImpl.h"
+
+#include <algorithm> // std::move
+#include <sstream> // std::stringstream
+
+#include "MQMessageConst.h"
+#include "MessageSysFlag.h"
+#include "UtilAll.h"
+
+namespace rocketmq {
+
+MessageImpl::MessageImpl() : MessageImpl(null, null) {}
+
+MessageImpl::MessageImpl(const std::string& topic, const std::string& body)
+ : MessageImpl(topic, null, null, 0, body, true) {}
+
+MessageImpl::MessageImpl(const std::string& topic,
+ const std::string& tags,
+ const std::string& keys,
+ int32_t flag,
+ const std::string& body,
+ bool waitStoreMsgOK)
+ : topic_(topic), flag_(flag), body_(body) {
+ if (tags.length() > 0) {
+ setTags(tags);
+ }
+
+ if (keys.length() > 0) {
+ setKeys(keys);
+ }
+
+ setWaitStoreMsgOK(waitStoreMsgOK);
+}
+
+MessageImpl::~MessageImpl() = default;
+
+const std::string& MessageImpl::getProperty(const std::string& name) const {
+ const auto& it = properties_.find(name);
+ if (it != properties_.end()) {
+ return it->second;
+ }
+ return null;
+}
+
+void MessageImpl::putProperty(const std::string& name, const std::string& value) {
+ properties_[name] = value;
+}
+
+void MessageImpl::clearProperty(const std::string& name) {
+ properties_.erase(name);
+}
+
+const std::string& MessageImpl::getTopic() const {
+ return topic_;
+}
+
+void MessageImpl::setTopic(const std::string& topic) {
+ topic_ = topic;
+}
+
+void MessageImpl::setTopic(const char* body, int len) {
+ topic_.clear();
+ topic_.append(body, len);
+}
+
+const std::string& MessageImpl::getTags() const {
+ return getProperty(MQMessageConst::PROPERTY_TAGS);
+}
+
+void MessageImpl::setTags(const std::string& tags) {
+ putProperty(MQMessageConst::PROPERTY_TAGS, tags);
+}
+
+const std::string& MessageImpl::getKeys() const {
+ return getProperty(MQMessageConst::PROPERTY_KEYS);
+}
+
+void MessageImpl::setKeys(const std::string& keys) {
+ putProperty(MQMessageConst::PROPERTY_KEYS, keys);
+}
+
+void MessageImpl::setKeys(const std::vector<std::string>& keys) {
+ if (keys.empty()) {
+ return;
+ }
+
+ std::string strKeys;
+ auto it = keys.begin();
+ strKeys += *it;
+ for (it++; it != keys.end(); it++) {
+ strKeys += MQMessageConst::KEY_SEPARATOR;
+ strKeys += *it;
+ }
+
+ setKeys(strKeys);
+}
+
+int MessageImpl::getDelayTimeLevel() const {
+ std::string tmp = getProperty(MQMessageConst::PROPERTY_DELAY_TIME_LEVEL);
+ if (!tmp.empty()) {
+ return atoi(tmp.c_str());
+ }
+ return 0;
+}
+
+void MessageImpl::setDelayTimeLevel(int level) {
+ putProperty(MQMessageConst::PROPERTY_DELAY_TIME_LEVEL, UtilAll::to_string(level));
+}
+
+bool MessageImpl::isWaitStoreMsgOK() const {
+ std::string tmp = getProperty(MQMessageConst::PROPERTY_WAIT_STORE_MSG_OK);
+ return tmp.empty() || UtilAll::stob(tmp);
+}
+
+void MessageImpl::setWaitStoreMsgOK(bool waitStoreMsgOK) {
+ putProperty(MQMessageConst::PROPERTY_WAIT_STORE_MSG_OK, UtilAll::to_string(waitStoreMsgOK));
+}
+
+int32_t MessageImpl::getFlag() const {
+ return flag_;
+}
+
+void MessageImpl::setFlag(int32_t flag) {
+ flag_ = flag;
+}
+
+const std::string& MessageImpl::getBody() const {
+ return body_;
+}
+
+void MessageImpl::setBody(const char* body, int len) {
+ body_.clear();
+ body_.append(body, len);
+}
+
+void MessageImpl::setBody(const std::string& body) {
+ body_ = body;
+}
+
+void MessageImpl::setBody(std::string&& body) {
+ body_ = std::move(body);
+}
+
+const std::string& MessageImpl::getTransactionId() const {
+ return transaction_id_;
+}
+
+void MessageImpl::setTransactionId(const std::string& transactionId) {
+ transaction_id_ = transactionId;
+}
+
+const std::map<std::string, std::string>& MessageImpl::getProperties() const {
+ return properties_;
+}
+
+void MessageImpl::setProperties(const std::map<std::string, std::string>& properties) {
+ properties_ = properties;
+}
+
+void MessageImpl::setProperties(std::map<std::string, std::string>&& properties) {
+ properties_ = std::move(properties);
+}
+
+std::string MessageImpl::toString() const {
+ std::stringstream ss;
+ ss << "Message [topic=" << topic_ << ", flag=" << flag_ << ", tag=" << getTags() << ", transactionId='"
+ << transaction_id_ + "']";
+ return ss.str();
+}
+
+} // namespace rocketmq
diff --git a/src/message/MessageImpl.h b/src/message/MessageImpl.h
new file mode 100644
index 0000000..bc4424f
--- /dev/null
+++ b/src/message/MessageImpl.h
@@ -0,0 +1,91 @@
+/*
+ * 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.
+ */
+#ifndef ROCKETMQ_MESSAGE_MESSAGEIMPL_H_
+#define ROCKETMQ_MESSAGE_MESSAGEIMPL_H_
+
+#include "Message.h"
+#include "noncopyable.h"
+
+namespace rocketmq {
+
+/**
+ * MessageImpl - Default Message implement
+ */
+class MessageImpl : public noncopyable, // base
+ virtual public Message // interface
+{
+ public:
+ MessageImpl();
+ MessageImpl(const std::string& topic, const std::string& body);
+ MessageImpl(const std::string& topic,
+ const std::string& tags,
+ const std::string& keys,
+ int32_t flag,
+ const std::string& body,
+ bool waitStoreMsgOK);
+
+ virtual ~MessageImpl();
+
+ const std::string& getProperty(const std::string& name) const override;
+ void putProperty(const std::string& name, const std::string& value) override;
+ void clearProperty(const std::string& name) override;
+
+ const std::string& getTopic() const override;
+ void setTopic(const std::string& topic) override;
+ void setTopic(const char* body, int len) override;
+
+ const std::string& getTags() const override;
+ void setTags(const std::string& tags) override;
+
+ const std::string& getKeys() const override;
+ void setKeys(const std::string& keys) override;
+ void setKeys(const std::vector<std::string>& keys) override;
+
+ int getDelayTimeLevel() const override;
+ void setDelayTimeLevel(int level) override;
+
+ bool isWaitStoreMsgOK() const override;
+ void setWaitStoreMsgOK(bool waitStoreMsgOK) override;
+
+ int32_t getFlag() const override;
+ void setFlag(int32_t flag) override;
+
+ const std::string& getBody() const override;
+ void setBody(const char* body, int len) override;
+ void setBody(const std::string& body) override;
+ void setBody(std::string&& body) override;
+
+ const std::string& getTransactionId() const override;
+ void setTransactionId(const std::string& transactionId) override;
+
+ const std::map<std::string, std::string>& getProperties() const override;
+ void setProperties(const std::map<std::string, std::string>& properties) override;
+ void setProperties(std::map<std::string, std::string>&& properties) override;
+
+ std::string toString() const override;
+
+ protected:
+ std::string topic_;
+ int32_t flag_;
+ std::map<std::string, std::string> properties_;
+ std::string body_;
+ std::string transaction_id_;
+};
+
+} // namespace rocketmq
+
+#endif // ROCKETMQ_MESSAGE_MESSAGEIMPL_H_
diff --git a/src/message/MessageUtil.cpp b/src/message/MessageUtil.cpp
index c5d2832..dc7e0a0 100644
--- a/src/message/MessageUtil.cpp
+++ b/src/message/MessageUtil.cpp
@@ -16,45 +16,37 @@
*/
#include "MessageUtil.h"
-#include <memory>
-
#include "ClientErrorCode.h"
-#include "MQMessageConst.h"
+#include "MessageImpl.h"
#include "MessageAccessor.h"
+#include "MQMessageConst.h"
#include "UtilAll.h"
namespace rocketmq {
-MQMessagePtr MessageUtil::createReplyMessage(const MQMessagePtr requestMessage,
- const std::string& body) throw(MQClientException) {
- if (requestMessage != nullptr) {
- std::unique_ptr<MQMessage> replyMessage(new MQMessage());
- const auto& cluster = requestMessage->getProperty(MQMessageConst::PROPERTY_CLUSTER);
- const auto& replyTo = requestMessage->getProperty(MQMessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT);
- const auto& correlationId = requestMessage->getProperty(MQMessageConst::PROPERTY_CORRELATION_ID);
- const auto& ttl = requestMessage->getProperty(MQMessageConst::PROPERTY_MESSAGE_TTL);
- replyMessage->setBody(body);
- if (!cluster.empty()) {
- auto replyTopic = UtilAll::getReplyTopic(cluster);
- replyMessage->setTopic(replyTopic);
- MessageAccessor::putProperty(*replyMessage, MQMessageConst::PROPERTY_MESSAGE_TYPE, REPLY_MESSAGE_FLAG);
- MessageAccessor::putProperty(*replyMessage, MQMessageConst::PROPERTY_CORRELATION_ID, correlationId);
- MessageAccessor::putProperty(*replyMessage, MQMessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT, replyTo);
- MessageAccessor::putProperty(*replyMessage, MQMessageConst::PROPERTY_MESSAGE_TTL, ttl);
-
- return replyMessage.release();
- } else {
- THROW_MQEXCEPTION(MQClientException, "create reply message fail, requestMessage error, property[" +
- MQMessageConst::PROPERTY_CLUSTER + "] is null.",
- ClientErrorCode::CREATE_REPLY_MESSAGE_EXCEPTION);
- }
+MQMessage MessageUtil::createReplyMessage(const Message& requestMessage,
+ const std::string& body) throw(MQClientException) {
+ const auto& cluster = requestMessage.getProperty(MQMessageConst::PROPERTY_CLUSTER);
+ if (!cluster.empty()) {
+ auto replyMessage = std::make_shared<MessageImpl>(UtilAll::getReplyTopic(cluster), body);
+ // set properties
+ MessageAccessor::putProperty(*replyMessage, MQMessageConst::PROPERTY_MESSAGE_TYPE, REPLY_MESSAGE_FLAG);
+ const auto& correlationId = requestMessage.getProperty(MQMessageConst::PROPERTY_CORRELATION_ID);
+ MessageAccessor::putProperty(*replyMessage, MQMessageConst::PROPERTY_CORRELATION_ID, correlationId);
+ const auto& replyTo = requestMessage.getProperty(MQMessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT);
+ MessageAccessor::putProperty(*replyMessage, MQMessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT, replyTo);
+ const auto& ttl = requestMessage.getProperty(MQMessageConst::PROPERTY_MESSAGE_TTL);
+ MessageAccessor::putProperty(*replyMessage, MQMessageConst::PROPERTY_MESSAGE_TTL, ttl);
+ return MQMessage(replyMessage);
+ } else {
+ THROW_MQEXCEPTION(MQClientException, "create reply message fail, requestMessage error, property[" +
+ MQMessageConst::PROPERTY_CLUSTER + "] is null.",
+ ClientErrorCode::CREATE_REPLY_MESSAGE_EXCEPTION);
}
- THROW_MQEXCEPTION(MQClientException, "create reply message fail, requestMessage cannot be null.",
- ClientErrorCode::CREATE_REPLY_MESSAGE_EXCEPTION);
}
-const std::string& MessageUtil::getReplyToClient(const MQMessagePtr msg) {
- return msg->getProperty(MQMessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT);
+const std::string& MessageUtil::getReplyToClient(const Message& msg) {
+ return msg.getProperty(MQMessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT);
}
} // namespace rocketmq
diff --git a/src/producer/DefaultMQProducer.cpp b/src/producer/DefaultMQProducer.cpp
index ce7e1a5..bf1c7ab 100644
--- a/src/producer/DefaultMQProducer.cpp
+++ b/src/producer/DefaultMQProducer.cpp
@@ -51,65 +51,65 @@ void DefaultMQProducer::shutdown() {
m_producerDelegate->shutdown();
}
-SendResult DefaultMQProducer::send(MQMessagePtr msg) {
+SendResult DefaultMQProducer::send(MQMessage& msg) {
return m_producerDelegate->send(msg);
}
-SendResult DefaultMQProducer::send(MQMessagePtr msg, long timeout) {
+SendResult DefaultMQProducer::send(MQMessage& msg, long timeout) {
return m_producerDelegate->send(msg, timeout);
}
-SendResult DefaultMQProducer::send(MQMessagePtr msg, const MQMessageQueue& mq) {
+SendResult DefaultMQProducer::send(MQMessage& msg, const MQMessageQueue& mq) {
return m_producerDelegate->send(msg, mq);
}
-SendResult DefaultMQProducer::send(MQMessagePtr msg, const MQMessageQueue& mq, long timeout) {
+SendResult DefaultMQProducer::send(MQMessage& msg, const MQMessageQueue& mq, long timeout) {
return m_producerDelegate->send(msg, mq, timeout);
}
-void DefaultMQProducer::send(MQMessagePtr msg, SendCallback* sendCallback) noexcept {
+void DefaultMQProducer::send(MQMessage& msg, SendCallback* sendCallback) noexcept {
m_producerDelegate->send(msg, sendCallback, getSendMsgTimeout());
}
-void DefaultMQProducer::send(MQMessagePtr msg, SendCallback* sendCallback, long timeout) noexcept {
+void DefaultMQProducer::send(MQMessage& msg, SendCallback* sendCallback, long timeout) noexcept {
m_producerDelegate->send(msg, sendCallback, timeout);
}
-void DefaultMQProducer::send(MQMessagePtr msg, const MQMessageQueue& mq, SendCallback* sendCallback) noexcept {
+void DefaultMQProducer::send(MQMessage& msg, const MQMessageQueue& mq, SendCallback* sendCallback) noexcept {
return m_producerDelegate->send(msg, mq, sendCallback);
}
-void DefaultMQProducer::send(MQMessagePtr msg,
+void DefaultMQProducer::send(MQMessage& msg,
const MQMessageQueue& mq,
SendCallback* sendCallback,
long timeout) noexcept {
m_producerDelegate->send(msg, mq, sendCallback, timeout);
}
-void DefaultMQProducer::sendOneway(MQMessagePtr msg) {
+void DefaultMQProducer::sendOneway(MQMessage& msg) {
m_producerDelegate->sendOneway(msg);
}
-void DefaultMQProducer::sendOneway(MQMessagePtr msg, const MQMessageQueue& mq) {
+void DefaultMQProducer::sendOneway(MQMessage& msg, const MQMessageQueue& mq) {
m_producerDelegate->sendOneway(msg, mq);
}
-SendResult DefaultMQProducer::send(MQMessagePtr msg, MessageQueueSelector* selector, void* arg) {
+SendResult DefaultMQProducer::send(MQMessage& msg, MessageQueueSelector* selector, void* arg) {
return m_producerDelegate->send(msg, selector, arg);
}
-SendResult DefaultMQProducer::send(MQMessagePtr msg, MessageQueueSelector* selector, void* arg, long timeout) {
+SendResult DefaultMQProducer::send(MQMessage& msg, MessageQueueSelector* selector, void* arg, long timeout) {
return m_producerDelegate->send(msg, selector, arg, timeout);
}
-void DefaultMQProducer::send(MQMessagePtr msg,
+void DefaultMQProducer::send(MQMessage& msg,
MessageQueueSelector* selector,
void* arg,
SendCallback* sendCallback) noexcept {
return m_producerDelegate->send(msg, selector, arg, sendCallback);
}
-void DefaultMQProducer::send(MQMessagePtr msg,
+void DefaultMQProducer::send(MQMessage& msg,
MessageQueueSelector* selector,
void* arg,
SendCallback* sendCallback,
@@ -117,32 +117,32 @@ void DefaultMQProducer::send(MQMessagePtr msg,
m_producerDelegate->send(msg, selector, arg, sendCallback, timeout);
}
-void DefaultMQProducer::sendOneway(MQMessagePtr msg, MessageQueueSelector* selector, void* arg) {
+void DefaultMQProducer::sendOneway(MQMessage& msg, MessageQueueSelector* selector, void* arg) {
m_producerDelegate->sendOneway(msg, selector, arg);
}
-TransactionSendResult DefaultMQProducer::sendMessageInTransaction(MQMessagePtr msg, void* arg) {
+TransactionSendResult DefaultMQProducer::sendMessageInTransaction(MQMessage& msg, void* arg) {
THROW_MQEXCEPTION(MQClientException, "sendMessageInTransaction not implement, please use TransactionMQProducer class",
-1);
}
-SendResult DefaultMQProducer::send(std::vector<MQMessagePtr>& msgs) {
+SendResult DefaultMQProducer::send(std::vector<MQMessage>& msgs) {
return m_producerDelegate->send(msgs);
}
-SendResult DefaultMQProducer::send(std::vector<MQMessagePtr>& msgs, long timeout) {
+SendResult DefaultMQProducer::send(std::vector<MQMessage>& msgs, long timeout) {
return m_producerDelegate->send(msgs, timeout);
}
-SendResult DefaultMQProducer::send(std::vector<MQMessagePtr>& msgs, const MQMessageQueue& mq) {
+SendResult DefaultMQProducer::send(std::vector<MQMessage>& msgs, const MQMessageQueue& mq) {
return m_producerDelegate->send(msgs, mq);
}
-SendResult DefaultMQProducer::send(std::vector<MQMessagePtr>& msgs, const MQMessageQueue& mq, long timeout) {
+SendResult DefaultMQProducer::send(std::vector<MQMessage>& msgs, const MQMessageQueue& mq, long timeout) {
return m_producerDelegate->send(msgs, mq, timeout);
}
-MQMessagePtr DefaultMQProducer::request(MQMessagePtr msg, long timeout) {
+MQMessage DefaultMQProducer::request(MQMessage& msg, long timeout) {
return m_producerDelegate->request(msg, timeout);
}
diff --git a/src/producer/DefaultMQProducerImpl.cpp b/src/producer/DefaultMQProducerImpl.cpp
index 3baa276..4fea984 100644
--- a/src/producer/DefaultMQProducerImpl.cpp
+++ b/src/producer/DefaultMQProducerImpl.cpp
@@ -31,7 +31,7 @@
#include "MQClientException.h"
#include "MQClientInstance.h"
#include "MQClientManager.h"
-#include "MQDecoder.h"
+#include "MessageDecoder.h"
#include "MQFaultStrategy.h"
#include "MQProtos.h"
#include "MessageBatch.h"
@@ -79,9 +79,8 @@ void DefaultMQProducerImpl::start() {
bool registerOK = m_clientInstance->registerProducer(m_producerConfig->getGroupName(), this);
if (!registerOK) {
m_serviceState = CREATE_JUST;
- THROW_MQEXCEPTION(MQClientException,
- "The producer group[" + m_producerConfig->getGroupName() +
- "] has been created before, specify another name please.",
+ THROW_MQEXCEPTION(MQClientException, "The producer group[" + m_producerConfig->getGroupName() +
+ "] has been created before, specify another name please.",
-1);
}
@@ -130,13 +129,13 @@ void DefaultMQProducerImpl::destroyTransactionEnv() {
m_checkTransactionExecutor->shutdown();
}
-SendResult DefaultMQProducerImpl::send(MQMessagePtr msg) {
+SendResult DefaultMQProducerImpl::send(MQMessage& msg) {
return send(msg, m_producerConfig->getSendMsgTimeout());
}
-SendResult DefaultMQProducerImpl::send(MQMessagePtr msg, long timeout) {
+SendResult DefaultMQProducerImpl::send(MQMessage& msg, long timeout) {
try {
- std::unique_ptr<SendResult> sendResult(sendDefaultImpl(msg, ComMode_SYNC, nullptr, timeout));
+ std::unique_ptr<SendResult> sendResult(sendDefaultImpl(msg.getMessageImpl(), ComMode_SYNC, nullptr, timeout));
return *sendResult;
} catch (MQException& e) {
LOG_ERROR_NEW("send failed, exception:{}", e.what());
@@ -144,19 +143,20 @@ SendResult DefaultMQProducerImpl::send(MQMessagePtr msg, long timeout) {
}
}
-SendResult DefaultMQProducerImpl::send(MQMessagePtr msg, const MQMessageQueue& mq) {
+SendResult DefaultMQProducerImpl::send(MQMessage& msg, const MQMessageQueue& mq) {
return send(msg, mq, m_producerConfig->getSendMsgTimeout());
}
-SendResult DefaultMQProducerImpl::send(MQMessagePtr msg, const MQMessageQueue& mq, long timeout) {
- Validators::checkMessage(*msg, m_producerConfig->getMaxMessageSize());
+SendResult DefaultMQProducerImpl::send(MQMessage& msg, const MQMessageQueue& mq, long timeout) {
+ Validators::checkMessage(msg, m_producerConfig->getMaxMessageSize());
- if (msg->getTopic() != mq.getTopic()) {
+ if (msg.getTopic() != mq.getTopic()) {
THROW_MQEXCEPTION(MQClientException, "message's topic not equal mq's topic", -1);
}
try {
- std::unique_ptr<SendResult> sendResult(sendKernelImpl(msg, mq, ComMode_SYNC, nullptr, nullptr, timeout));
+ std::unique_ptr<SendResult> sendResult(
+ sendKernelImpl(msg.getMessageImpl(), mq, ComMode_SYNC, nullptr, nullptr, timeout));
return *sendResult;
} catch (MQException& e) {
LOG_ERROR_NEW("send failed, exception:{}", e.what());
@@ -164,13 +164,13 @@ SendResult DefaultMQProducerImpl::send(MQMessagePtr msg, const MQMessageQueue& m
}
}
-void DefaultMQProducerImpl::send(MQMessagePtr msg, SendCallback* sendCallback) noexcept {
+void DefaultMQProducerImpl::send(MQMessage& msg, SendCallback* sendCallback) noexcept {
return send(msg, sendCallback, m_producerConfig->getSendMsgTimeout());
}
-void DefaultMQProducerImpl::send(MQMessagePtr msg, SendCallback* sendCallback, long timeout) noexcept {
+void DefaultMQProducerImpl::send(MQMessage& msg, SendCallback* sendCallback, long timeout) noexcept {
try {
- (void)sendDefaultImpl(msg, ComMode_ASYNC, sendCallback, timeout);
+ (void)sendDefaultImpl(msg.getMessageImpl(), ComMode_ASYNC, sendCallback, timeout);
} catch (MQException& e) {
LOG_ERROR_NEW("send failed, exception:{}", e.what());
sendCallback->onException(e);
@@ -183,23 +183,23 @@ void DefaultMQProducerImpl::send(MQMessagePtr msg, SendCallback* sendCallback, l
}
}
-void DefaultMQProducerImpl::send(MQMessagePtr msg, const MQMessageQueue& mq, SendCallback* sendCallback) noexcept {
+void DefaultMQProducerImpl::send(MQMessage& msg, const MQMessageQueue& mq, SendCallback* sendCallback) noexcept {
return send(msg, mq, sendCallback, m_producerConfig->getSendMsgTimeout());
}
-void DefaultMQProducerImpl::send(MQMessagePtr msg,
+void DefaultMQProducerImpl::send(MQMessage& msg,
const MQMessageQueue& mq,
SendCallback* sendCallback,
long timeout) noexcept {
try {
- Validators::checkMessage(*msg, m_producerConfig->getMaxMessageSize());
+ Validators::checkMessage(msg, m_producerConfig->getMaxMessageSize());
- if (msg->getTopic() != mq.getTopic()) {
+ if (msg.getTopic() != mq.getTopic()) {
THROW_MQEXCEPTION(MQClientException, "message's topic not equal mq's topic", -1);
}
try {
- sendKernelImpl(msg, mq, ComMode_ASYNC, sendCallback, nullptr, timeout);
+ sendKernelImpl(msg.getMessageImpl(), mq, ComMode_ASYNC, sendCallback, nullptr, timeout);
} catch (MQBrokerException& e) {
std::string info = std::string("unknown exception, ") + e.what();
THROW_MQEXCEPTION(MQClientException, info, e.GetError());
@@ -216,37 +216,38 @@ void DefaultMQProducerImpl::send(MQMessagePtr msg,
}
}
-void DefaultMQProducerImpl::sendOneway(MQMessagePtr msg) {
+void DefaultMQProducerImpl::sendOneway(MQMessage& msg) {
try {
- sendDefaultImpl(msg, ComMode_ONEWAY, nullptr, m_producerConfig->getSendMsgTimeout());
+ sendDefaultImpl(msg.getMessageImpl(), ComMode_ONEWAY, nullptr, m_producerConfig->getSendMsgTimeout());
} catch (MQBrokerException e) {
std::string info = std::string("unknown exception, ") + e.what();
THROW_MQEXCEPTION(MQClientException, info, e.GetError());
}
}
-void DefaultMQProducerImpl::sendOneway(MQMessagePtr msg, const MQMessageQueue& mq) {
- Validators::checkMessage(*msg, m_producerConfig->getMaxMessageSize());
+void DefaultMQProducerImpl::sendOneway(MQMessage& msg, const MQMessageQueue& mq) {
+ Validators::checkMessage(msg, m_producerConfig->getMaxMessageSize());
- if (msg->getTopic() != mq.getTopic()) {
+ if (msg.getTopic() != mq.getTopic()) {
THROW_MQEXCEPTION(MQClientException, "message's topic not equal mq's topic", -1);
}
try {
- sendKernelImpl(msg, mq, ComMode_ONEWAY, nullptr, nullptr, m_producerConfig->getSendMsgTimeout());
+ sendKernelImpl(msg.getMessageImpl(), mq, ComMode_ONEWAY, nullptr, nullptr, m_producerConfig->getSendMsgTimeout());
} catch (MQBrokerException e) {
std::string info = std::string("unknown exception, ") + e.what();
THROW_MQEXCEPTION(MQClientException, info, e.GetError());
}
}
-SendResult DefaultMQProducerImpl::send(MQMessagePtr msg, MessageQueueSelector* selector, void* arg) {
+SendResult DefaultMQProducerImpl::send(MQMessage& msg, MessageQueueSelector* selector, void* arg) {
return send(msg, selector, arg, m_producerConfig->getSendMsgTimeout());
}
-SendResult DefaultMQProducerImpl::send(MQMessagePtr msg, MessageQueueSelector* selector, void* arg, long timeout) {
+SendResult DefaultMQProducerImpl::send(MQMessage& msg, MessageQueueSelector* selector, void* arg, long timeout) {
try {
- std::unique_ptr<SendResult> result(sendSelectImpl(msg, selector, arg, ComMode_SYNC, nullptr, timeout));
+ std::unique_ptr<SendResult> result(
+ sendSelectImpl(msg.getMessageImpl(), selector, arg, ComMode_SYNC, nullptr, timeout));
return *result.get();
} catch (MQException& e) {
LOG_ERROR_NEW("send failed, exception:{}", e.what());
@@ -254,21 +255,21 @@ SendResult DefaultMQProducerImpl::send(MQMessagePtr msg, MessageQueueSelector* s
}
}
-void DefaultMQProducerImpl::send(MQMessagePtr msg,
+void DefaultMQProducerImpl::send(MQMessage& msg,
MessageQueueSelector* selector,
void* arg,
SendCallback* sendCallback) noexcept {
return send(msg, selector, arg, sendCallback, m_producerConfig->getSendMsgTimeout());
}
-void DefaultMQProducerImpl::send(MQMessagePtr msg,
+void DefaultMQProducerImpl::send(MQMessage& msg,
MessageQueueSelector* selector,
void* arg,
SendCallback* sendCallback,
long timeout) noexcept {
try {
try {
- sendSelectImpl(msg, selector, arg, ComMode_ASYNC, sendCallback, timeout);
+ sendSelectImpl(msg.getMessageImpl(), selector, arg, ComMode_ASYNC, sendCallback, timeout);
} catch (MQBrokerException& e) {
std::string info = std::string("unknown exception, ") + e.what();
THROW_MQEXCEPTION(MQClientException, info, e.GetError());
@@ -285,19 +286,19 @@ void DefaultMQProducerImpl::send(MQMessagePtr msg,
}
}
-void DefaultMQProducerImpl::sendOneway(MQMessagePtr msg, MessageQueueSelector* selector, void* arg) {
+void DefaultMQProducerImpl::sendOneway(MQMessage& msg, MessageQueueSelector* selector, void* arg) {
try {
- sendSelectImpl(msg, selector, arg, ComMode_ONEWAY, nullptr, m_producerConfig->getSendMsgTimeout());
+ sendSelectImpl(msg.getMessageImpl(), selector, arg, ComMode_ONEWAY, nullptr, m_producerConfig->getSendMsgTimeout());
} catch (MQBrokerException e) {
std::string info = std::string("unknown exception, ") + e.what();
THROW_MQEXCEPTION(MQClientException, info, e.GetError());
}
}
-TransactionSendResult DefaultMQProducerImpl::sendMessageInTransaction(MQMessagePtr msg, void* arg) {
+TransactionSendResult DefaultMQProducerImpl::sendMessageInTransaction(MQMessage& msg, void* arg) {
try {
std::unique_ptr<TransactionSendResult> sendResult(
- sendMessageInTransactionImpl(msg, arg, m_producerConfig->getSendMsgTimeout()));
+ sendMessageInTransactionImpl(msg.getMessageImpl(), arg, m_producerConfig->getSendMsgTimeout()));
return *sendResult;
} catch (MQException& e) {
LOG_ERROR_NEW("sendMessageInTransaction failed, exception:{}", e.what());
@@ -305,39 +306,39 @@ TransactionSendResult DefaultMQProducerImpl::sendMessageInTransaction(MQMessageP
}
}
-SendResult DefaultMQProducerImpl::send(std::vector<MQMessagePtr>& msgs) {
- std::unique_ptr<MessageBatch> batchMessage(batch(msgs));
- return send(batchMessage.get());
+SendResult DefaultMQProducerImpl::send(std::vector<MQMessage>& msgs) {
+ MQMessage batchMessage(batch(msgs));
+ return send(batchMessage);
}
-SendResult DefaultMQProducerImpl::send(std::vector<MQMessagePtr>& msgs, long timeout) {
- std::unique_ptr<MessageBatch> batchMessage(batch(msgs));
- return send(batchMessage.get(), timeout);
+SendResult DefaultMQProducerImpl::send(std::vector<MQMessage>& msgs, long timeout) {
+ MQMessage batchMessage(batch(msgs));
+ return send(batchMessage, timeout);
}
-SendResult DefaultMQProducerImpl::send(std::vector<MQMessagePtr>& msgs, const MQMessageQueue& mq) {
- std::unique_ptr<MessageBatch> batchMessage(batch(msgs));
- return send(batchMessage.get(), mq);
+SendResult DefaultMQProducerImpl::send(std::vector<MQMessage>& msgs, const MQMessageQueue& mq) {
+ MQMessage batchMessage(batch(msgs));
+ return send(batchMessage, mq);
}
-SendResult DefaultMQProducerImpl::send(std::vector<MQMessagePtr>& msgs, const MQMessageQueue& mq, long timeout) {
- std::unique_ptr<MessageBatch> batchMessage(batch(msgs));
- return send(batchMessage.get(), mq, timeout);
+SendResult DefaultMQProducerImpl::send(std::vector<MQMessage>& msgs, const MQMessageQueue& mq, long timeout) {
+ MQMessage batchMessage(batch(msgs));
+ return send(batchMessage, mq, timeout);
}
-MessageBatch* DefaultMQProducerImpl::batch(std::vector<MQMessagePtr>& msgs) {
+MessagePtr DefaultMQProducerImpl::batch(std::vector<MQMessage>& msgs) {
if (msgs.size() < 1) {
THROW_MQEXCEPTION(MQClientException, "msgs need one message at least", -1);
}
try {
- std::unique_ptr<MessageBatch> msgBatch(MessageBatch::generateFromList(msgs));
- for (auto& message : msgBatch->getMessages()) {
- Validators::checkMessage(*message, m_producerConfig->getMaxMessageSize());
- MessageClientIDSetter::setUniqID(*message);
+ auto messageBatch = MessageBatch::generateFromList(msgs);
+ for (auto& message : messageBatch->getMessages()) {
+ Validators::checkMessage(message, m_producerConfig->getMaxMessageSize());
+ MessageClientIDSetter::setUniqID(const_cast<MQMessage&>(message));
}
- msgBatch->setBody(msgBatch->encode());
- return msgBatch.release();
+ messageBatch->setBody(messageBatch->encode());
+ return messageBatch;
} catch (std::exception& e) {
THROW_MQEXCEPTION(MQClientException, "Failed to initiate the MessageBatch", -1);
}
@@ -359,29 +360,29 @@ class RequestSendCallback : public AutoDeleteSendCallback {
std::shared_ptr<RequestResponseFuture> m_requestFuture;
};
-MQMessagePtr DefaultMQProducerImpl::request(MQMessagePtr msg, long timeout) {
+MQMessage DefaultMQProducerImpl::request(MQMessage& msg, long timeout) {
auto beginTimestamp = UtilAll::currentTimeMillis();
prepareSendRequest(msg, timeout);
- const auto& correlationId = msg->getProperty(MQMessageConst::PROPERTY_CORRELATION_ID);
+ const auto& correlationId = msg.getProperty(MQMessageConst::PROPERTY_CORRELATION_ID);
std::exception_ptr eptr = nullptr;
- std::unique_ptr<MQMessage> responseMessage;
+ MessagePtr responseMessage;
try {
auto requestResponseFuture = std::make_shared<RequestResponseFuture>(correlationId, timeout, nullptr);
RequestFutureTable::putRequestFuture(correlationId, requestResponseFuture);
auto cost = UtilAll::currentTimeMillis() - beginTimestamp;
- sendDefaultImpl(msg, CommunicationMode::ComMode_ASYNC, new RequestSendCallback(requestResponseFuture),
- timeout - cost);
+ sendDefaultImpl(msg.getMessageImpl(), CommunicationMode::ComMode_ASYNC,
+ new RequestSendCallback(requestResponseFuture), timeout - cost);
- responseMessage.reset(requestResponseFuture->waitResponseMessage(timeout - cost));
+ responseMessage = requestResponseFuture->waitResponseMessage(timeout - cost);
if (responseMessage == nullptr) {
if (requestResponseFuture->isSendRequestOk()) {
- std::string info = "send request message to <" + msg->getTopic() + "> OK, but wait reply message timeout, " +
+ std::string info = "send request message to <" + msg.getTopic() + "> OK, but wait reply message timeout, " +
UtilAll::to_string(timeout) + " ms.";
THROW_MQEXCEPTION(RequestTimeoutException, info, ClientErrorCode::REQUEST_TIMEOUT_EXCEPTION);
} else {
- std::string info = "send request message to <" + msg->getTopic() + "> fail";
+ std::string info = "send request message to <" + msg.getTopic() + "> fail";
THROW_MQEXCEPTION2(MQClientException, info, -1, requestResponseFuture->getCause());
}
}
@@ -396,24 +397,24 @@ MQMessagePtr DefaultMQProducerImpl::request(MQMessagePtr msg, long timeout) {
std::rethrow_exception(eptr);
}
- return responseMessage.release();
+ return MQMessage(responseMessage);
}
-void DefaultMQProducerImpl::prepareSendRequest(MQMessagePtr msg, long timeout) {
+void DefaultMQProducerImpl::prepareSendRequest(Message& msg, long timeout) {
const auto correlationId = CorrelationIdUtil::createCorrelationId();
const auto& requestClientId = m_clientInstance->getClientId();
- MessageAccessor::putProperty(*msg, MQMessageConst::PROPERTY_CORRELATION_ID, correlationId);
- MessageAccessor::putProperty(*msg, MQMessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT, requestClientId);
- MessageAccessor::putProperty(*msg, MQMessageConst::PROPERTY_MESSAGE_TTL, UtilAll::to_string(timeout));
+ MessageAccessor::putProperty(msg, MQMessageConst::PROPERTY_CORRELATION_ID, correlationId);
+ MessageAccessor::putProperty(msg, MQMessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT, requestClientId);
+ MessageAccessor::putProperty(msg, MQMessageConst::PROPERTY_MESSAGE_TTL, UtilAll::to_string(timeout));
- auto hasRouteData = m_clientInstance->getTopicRouteData(msg->getTopic()) != nullptr;
+ auto hasRouteData = m_clientInstance->getTopicRouteData(msg.getTopic()) != nullptr;
if (!hasRouteData) {
auto beginTimestamp = UtilAll::currentTimeMillis();
- m_clientInstance->tryToFindTopicPublishInfo(msg->getTopic());
+ m_clientInstance->tryToFindTopicPublishInfo(msg.getTopic());
m_clientInstance->sendHeartbeatToAllBrokerWithLock();
auto cost = UtilAll::currentTimeMillis() - beginTimestamp;
if (cost > 500) {
- LOG_WARN_NEW("prepare send request for <{}> cost {} ms", msg->getTopic(), cost);
+ LOG_WARN_NEW("prepare send request for <{}> cost {} ms", msg.getTopic(), cost);
}
}
}
@@ -427,7 +428,7 @@ void DefaultMQProducerImpl::updateFaultItem(const std::string& brokerName, const
m_mqFaultStrategy->updateFaultItem(brokerName, currentLatency, isolation);
}
-SendResult* DefaultMQProducerImpl::sendDefaultImpl(MQMessagePtr msg,
+SendResult* DefaultMQProducerImpl::sendDefaultImpl(MessagePtr msg,
CommunicationMode communicationMode,
SendCallback* sendCallback,
long timeout) {
@@ -495,15 +496,15 @@ SendResult* DefaultMQProducerImpl::sendDefaultImpl(MQMessagePtr msg,
}
std::string info = "Send [" + UtilAll::to_string(times) + "] times, still failed, cost [" +
- UtilAll::to_string(UtilAll::currentTimeMillis() - beginTimestampFirst) +
- "]ms, Topic: " + msg->getTopic();
+ UtilAll::to_string(UtilAll::currentTimeMillis() - beginTimestampFirst) + "]ms, Topic: " +
+ msg->getTopic();
THROW_MQEXCEPTION(MQClientException, info, -1);
}
THROW_MQEXCEPTION(MQClientException, "No route info of this topic: " + msg->getTopic(), -1);
}
-SendResult* DefaultMQProducerImpl::sendKernelImpl(MQMessagePtr msg,
+SendResult* DefaultMQProducerImpl::sendKernelImpl(MessagePtr msg,
const MQMessageQueue& mq,
CommunicationMode communicationMode,
SendCallback* sendCallback,
@@ -547,7 +548,7 @@ SendResult* DefaultMQProducerImpl::sendKernelImpl(MQMessagePtr msg,
requestHeader->sysFlag = sysFlag;
requestHeader->bornTimestamp = UtilAll::currentTimeMillis();
requestHeader->flag = msg->getFlag();
- requestHeader->properties = MQDecoder::messageProperties2String(msg->getProperties());
+ requestHeader->properties = MessageDecoder::messageProperties2String(msg->getProperties());
requestHeader->reconsumeTimes = 0;
requestHeader->unitMode = false;
requestHeader->batch = msg->isBatch();
@@ -601,7 +602,7 @@ SendResult* DefaultMQProducerImpl::sendKernelImpl(MQMessagePtr msg,
THROW_MQEXCEPTION(MQClientException, "The broker[" + mq.getBrokerName() + "] not exist", -1);
}
-SendResult* DefaultMQProducerImpl::sendSelectImpl(MQMessagePtr msg,
+SendResult* DefaultMQProducerImpl::sendSelectImpl(MessagePtr msg,
MessageQueueSelector* selector,
void* arg,
CommunicationMode communicationMode,
@@ -612,7 +613,7 @@ SendResult* DefaultMQProducerImpl::sendSelectImpl(MQMessagePtr msg,
TopicPublishInfoPtr topicPublishInfo = m_clientInstance->tryToFindTopicPublishInfo(msg->getTopic());
if (topicPublishInfo != nullptr && topicPublishInfo->ok()) {
- MQMessageQueue mq = selector->select(topicPublishInfo->getMessageQueueList(), *msg, arg);
+ MQMessageQueue mq = selector->select(topicPublishInfo->getMessageQueueList(), MQMessage(msg), arg);
auto costTime = UtilAll::currentTimeMillis() - beginStartTime;
if (timeout < costTime) {
@@ -626,7 +627,7 @@ SendResult* DefaultMQProducerImpl::sendSelectImpl(MQMessagePtr msg,
THROW_MQEXCEPTION(MQClientException, info, -1);
}
-TransactionSendResult* DefaultMQProducerImpl::sendMessageInTransactionImpl(MQMessagePtr msg, void* arg, long timeout) {
+TransactionSendResult* DefaultMQProducerImpl::sendMessageInTransactionImpl(MessagePtr msg, void* arg, long timeout) {
auto* transactionListener = getCheckListener();
if (nullptr == transactionListener) {
THROW_MQEXCEPTION(MQClientException, "transactionListener is null", -1);
@@ -653,7 +654,7 @@ TransactionSendResult* DefaultMQProducerImpl::sendMessageInTransactionImpl(MQMes
if (!transactionId.empty()) {
msg->setTransactionId(transactionId);
}
- localTransactionState = transactionListener->executeLocalTransaction(*msg, arg);
+ localTransactionState = transactionListener->executeLocalTransaction(MQMessage(msg), arg);
if (localTransactionState != LocalTransactionState::COMMIT_MESSAGE) {
LOG_INFO_NEW("executeLocalTransaction return not COMMIT_MESSAGE, msg:{}", msg->toString());
}
@@ -695,7 +696,7 @@ TransactionListener* DefaultMQProducerImpl::getCheckListener() {
};
void DefaultMQProducerImpl::checkTransactionState(const std::string& addr,
- MQMessageExtPtr2 msg,
+ MessageExtPtr msg,
CheckTransactionStateRequestHeader* checkRequestHeader) {
long tranStateTableOffset = checkRequestHeader->tranStateTableOffset;
long commitLogOffset = checkRequestHeader->commitLogOffset;
@@ -709,7 +710,7 @@ void DefaultMQProducerImpl::checkTransactionState(const std::string& addr,
}
void DefaultMQProducerImpl::checkTransactionStateImpl(const std::string& addr,
- MQMessageExtPtr2 message,
+ MessageExtPtr message,
long tranStateTableOffset,
long commitLogOffset,
const std::string& msgId,
@@ -725,7 +726,7 @@ void DefaultMQProducerImpl::checkTransactionStateImpl(const std::string& addr,
LocalTransactionState localTransactionState = UNKNOWN;
std::exception_ptr exception = nullptr;
try {
- localTransactionState = transactionCheckListener->checkLocalTransaction(*message);
+ localTransactionState = transactionCheckListener->checkLocalTransaction(MQMessageExt(message));
} catch (MQException& e) {
LOG_ERROR_NEW("Broker call checkTransactionState, but checkLocalTransactionState exception, {}", e.what());
exception = std::current_exception();
@@ -775,12 +776,8 @@ void DefaultMQProducerImpl::checkTransactionStateImpl(const std::string& addr,
void DefaultMQProducerImpl::endTransaction(SendResult& sendResult,
LocalTransactionState localTransactionState,
std::exception_ptr& localException) {
- MQMessageId id;
- if (!sendResult.getOffsetMsgId().empty()) {
- id = MQDecoder::decodeMessageId(sendResult.getOffsetMsgId());
- } else {
- id = MQDecoder::decodeMessageId(sendResult.getMsgId());
- }
+ const auto& msg_id = !sendResult.getOffsetMsgId().empty() ? sendResult.getOffsetMsgId() : sendResult.getMsgId();
+ auto id = MessageDecoder::decodeMessageId(msg_id);
const auto& transactionId = sendResult.getTransactionId();
std::string brokerAddr = m_clientInstance->findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());
EndTransactionRequestHeader* requestHeader = new EndTransactionRequestHeader();
@@ -810,7 +807,7 @@ void DefaultMQProducerImpl::endTransaction(SendResult& sendResult,
m_clientInstance->getMQClientAPIImpl()->endTransactionOneway(brokerAddr, requestHeader, remark);
}
-bool DefaultMQProducerImpl::tryToCompressMessage(MQMessage& msg) {
+bool DefaultMQProducerImpl::tryToCompressMessage(Message& msg) {
if (msg.isBatch()) {
// batch dose not support compressing right now
return false;
diff --git a/src/producer/DefaultMQProducerImpl.h b/src/producer/DefaultMQProducerImpl.h
index 4a2f273..0b1a93a 100644
--- a/src/producer/DefaultMQProducerImpl.h
+++ b/src/producer/DefaultMQProducerImpl.h
@@ -57,48 +57,48 @@ class DefaultMQProducerImpl : public std::enable_shared_from_this<DefaultMQProdu
void shutdown() override;
// Sync: caller will be responsible for the lifecycle of messages.
- SendResult send(MQMessagePtr msg) override;
- SendResult send(MQMessagePtr msg, long timeout) override;
- SendResult send(MQMessagePtr msg, const MQMessageQueue& mq) override;
- SendResult send(MQMessagePtr msg, const MQMessageQueue& mq, long timeout) override;
+ SendResult send(MQMessage& msg) override;
+ SendResult send(MQMessage& msg, long timeout) override;
+ SendResult send(MQMessage& msg, const MQMessageQueue& mq) override;
+ SendResult send(MQMessage& msg, const MQMessageQueue& mq, long timeout) override;
// Async: don't delete msg object, until callback occur.
- void send(MQMessagePtr msg, SendCallback* sendCallback) noexcept override;
- void send(MQMessagePtr msg, SendCallback* sendCallback, long timeout) noexcept override;
- void send(MQMessagePtr msg, const MQMessageQueue& mq, SendCallback* sendCallback) noexcept override;
- void send(MQMessagePtr msg, const MQMessageQueue& mq, SendCallback* sendCallback, long timeout) noexcept override;
+ void send(MQMessage& msg, SendCallback* sendCallback) noexcept override;
+ void send(MQMessage& msg, SendCallback* sendCallback, long timeout) noexcept override;
+ void send(MQMessage& msg, const MQMessageQueue& mq, SendCallback* sendCallback) noexcept override;
+ void send(MQMessage& msg, const MQMessageQueue& mq, SendCallback* sendCallback, long timeout) noexcept override;
// Oneyway: same as sync send, but don't care its result.
- void sendOneway(MQMessagePtr msg) override;
- void sendOneway(MQMessagePtr msg, const MQMessageQueue& mq) override;
+ void sendOneway(MQMessage& msg) override;
+ void sendOneway(MQMessage& msg, const MQMessageQueue& mq) override;
// Select
- SendResult send(MQMessagePtr msg, MessageQueueSelector* selector, void* arg) override;
- SendResult send(MQMessagePtr msg, MessageQueueSelector* selector, void* arg, long timeout) override;
- void send(MQMessagePtr msg, MessageQueueSelector* selector, void* arg, SendCallback* sendCallback) noexcept override;
- void send(MQMessagePtr msg,
+ SendResult send(MQMessage& msg, MessageQueueSelector* selector, void* arg) override;
+ SendResult send(MQMessage& msg, MessageQueueSelector* selector, void* arg, long timeout) override;
+ void send(MQMessage& msg, MessageQueueSelector* selector, void* arg, SendCallback* sendCallback) noexcept override;
+ void send(MQMessage& msg,
MessageQueueSelector* selector,
void* arg,
SendCallback* sendCallback,
long timeout) noexcept override;
- void sendOneway(MQMessagePtr msg, MessageQueueSelector* selector, void* arg) override;
+ void sendOneway(MQMessage& msg, MessageQueueSelector* selector, void* arg) override;
// Transaction
- TransactionSendResult sendMessageInTransaction(MQMessagePtr msg, void* arg) override;
+ TransactionSendResult sendMessageInTransaction(MQMessage& msg, void* arg) override;
// Batch: power by sync send, caller will be responsible for the lifecycle of messages.
- SendResult send(std::vector<MQMessagePtr>& msgs) override;
- SendResult send(std::vector<MQMessagePtr>& msgs, long timeout) override;
- SendResult send(std::vector<MQMessagePtr>& msgs, const MQMessageQueue& mq) override;
- SendResult send(std::vector<MQMessagePtr>& msgs, const MQMessageQueue& mq, long timeout) override;
+ SendResult send(std::vector<MQMessage>& msgs) override;
+ SendResult send(std::vector<MQMessage>& msgs, long timeout) override;
+ SendResult send(std::vector<MQMessage>& msgs, const MQMessageQueue& mq) override;
+ SendResult send(std::vector<MQMessage>& msgs, const MQMessageQueue& mq, long timeout) override;
- MQMessagePtr request(MQMessagePtr msg, long timeout) override;
+ MQMessage request(MQMessage& msg, long timeout) override;
public: // MQProducerInner
TransactionListener* getCheckListener() override;
void checkTransactionState(const std::string& addr,
- MQMessageExtPtr2 msg,
+ MessageExtPtr msg,
CheckTransactionStateRequestHeader* checkRequestHeader) override;
public:
@@ -116,37 +116,37 @@ class DefaultMQProducerImpl : public std::enable_shared_from_this<DefaultMQProdu
void setSendLatencyFaultEnable(bool sendLatencyFaultEnable);
protected:
- SendResult* sendDefaultImpl(MQMessagePtr msg,
+ SendResult* sendDefaultImpl(MessagePtr msg,
CommunicationMode communicationMode,
SendCallback* sendCallback,
long timeout);
- SendResult* sendKernelImpl(MQMessagePtr msg,
+ SendResult* sendKernelImpl(MessagePtr msg,
const MQMessageQueue& mq,
CommunicationMode communicationMode,
SendCallback* sendCallback,
std::shared_ptr<const TopicPublishInfo> topicPublishInfo,
long timeout);
- SendResult* sendSelectImpl(MQMessagePtr msg,
+ SendResult* sendSelectImpl(MessagePtr msg,
MessageQueueSelector* selector,
void* arg,
CommunicationMode communicationMode,
SendCallback* sendCallback,
long timeout);
- TransactionSendResult* sendMessageInTransactionImpl(MQMessagePtr msg, void* arg, long timeout);
+ TransactionSendResult* sendMessageInTransactionImpl(MessagePtr msg, void* arg, long timeout);
void checkTransactionStateImpl(const std::string& addr,
- MQMessageExtPtr2 message,
+ MessageExtPtr message,
long tranStateTableOffset,
long commitLogOffset,
const std::string& msgId,
const std::string& transactionId,
const std::string& offsetMsgId);
- bool tryToCompressMessage(MQMessage& msg);
+ bool tryToCompressMessage(Message& msg);
- MessageBatch* batch(std::vector<MQMessagePtr>& msgs);
+ MessagePtr batch(std::vector<MQMessage>& msgs);
- void prepareSendRequest(MQMessagePtr msg, long timeout);
+ void prepareSendRequest(Message& msg, long timeout);
private:
DefaultMQProducerConfigPtr m_producerConfig;
diff --git a/src/producer/MQProducerInner.h b/src/producer/MQProducerInner.h
index 6e3bede..c0538c3 100644
--- a/src/producer/MQProducerInner.h
+++ b/src/producer/MQProducerInner.h
@@ -31,7 +31,7 @@ class MQProducerInner {
virtual TransactionListener* getCheckListener() = 0;
virtual void checkTransactionState(const std::string& addr,
- MQMessageExtPtr2 msg,
+ MessageExtPtr msg,
CheckTransactionStateRequestHeader* checkRequestHeader) = 0;
// virtual std::vector<std::string> getPublishTopicList() = 0;
diff --git a/src/producer/RequestResponseFuture.cpp b/src/producer/RequestResponseFuture.cpp
index 92e86a4..c3b28f9 100644
--- a/src/producer/RequestResponseFuture.cpp
+++ b/src/producer/RequestResponseFuture.cpp
@@ -68,17 +68,17 @@ bool RequestResponseFuture::isTimeout() {
return diff > m_timeoutMillis;
}
-MQMessagePtr RequestResponseFuture::waitResponseMessage(int64_t timeout) {
+MessagePtr RequestResponseFuture::waitResponseMessage(int64_t timeout) {
if (m_countDownLatch != nullptr) {
if (timeout < 0) {
timeout = 0;
}
m_countDownLatch->wait(timeout, time_unit::milliseconds);
}
- return m_responseMsg.release();
+ return m_responseMsg;
}
-void RequestResponseFuture::putResponseMessage(MQMessagePtr3 responseMsg) {
+void RequestResponseFuture::putResponseMessage(MessagePtr responseMsg) {
m_responseMsg = std::move(responseMsg);
if (m_countDownLatch != nullptr) {
m_countDownLatch->count_down();
diff --git a/src/producer/RequestResponseFuture.h b/src/producer/RequestResponseFuture.h
index 0ebb235..d4e3e68 100644
--- a/src/producer/RequestResponseFuture.h
+++ b/src/producer/RequestResponseFuture.h
@@ -20,7 +20,7 @@
#include <exception>
#include <string>
-#include "MQMessage.h"
+#include "Message.h"
#include "RequestCallback.h"
#include "concurrent/latch.hpp"
@@ -34,8 +34,8 @@ class RequestResponseFuture {
bool isTimeout();
- MQMessagePtr waitResponseMessage(int64_t timeout);
- void putResponseMessage(MQMessagePtr3 responseMsg);
+ MessagePtr waitResponseMessage(int64_t timeout);
+ void putResponseMessage(MessagePtr responseMsg);
bool isSendRequestOk();
void setSendRequestOk(bool sendRequestOk);
@@ -47,10 +47,10 @@ class RequestResponseFuture {
std::string m_correlationId;
RequestCallback* m_requestCallback;
uint64_t m_beginTimestamp;
- MQMessagePtr m_requestMsg;
+ MessagePtr m_requestMsg;
long m_timeoutMillis;
std::unique_ptr<latch> m_countDownLatch; // use for synchronization rpc
- MQMessagePtr3 m_responseMsg;
+ MessagePtr m_responseMsg;
bool m_sendRequestOk;
std::exception_ptr m_cause;
};
diff --git a/src/producer/TransactionMQProducer.cpp b/src/producer/TransactionMQProducer.cpp
index b258dad..cd2e6a7 100644
--- a/src/producer/TransactionMQProducer.cpp
+++ b/src/producer/TransactionMQProducer.cpp
@@ -54,7 +54,7 @@ void TransactionMQProducer::shutdown() {
std::dynamic_pointer_cast<DefaultMQProducerImpl>(m_producerDelegate)->destroyTransactionEnv();
}
-TransactionSendResult TransactionMQProducer::sendMessageInTransaction(MQMessagePtr msg, void* arg) {
+TransactionSendResult TransactionMQProducer::sendMessageInTransaction(MQMessage& msg, void* arg) {
return m_producerDelegate->sendMessageInTransaction(msg, arg);
}
diff --git a/src/transport/SocketUtil.cpp b/src/transport/SocketUtil.cpp
index e114128..e4810d8 100644
--- a/src/transport/SocketUtil.cpp
+++ b/src/transport/SocketUtil.cpp
@@ -16,12 +16,14 @@
*/
#include "SocketUtil.h"
-#include <event2/event.h>
+#include <cstdlib> // std::realloc
+#include <cstring> // std::memset, std::memcpy
-#include <cstring>
#include <sstream>
#include <stdexcept>
+#include <event2/event.h>
+
#ifndef WIN32
#include <netdb.h>
#include <unistd.h>
@@ -113,7 +115,7 @@ struct sockaddr* lookupNameServers(const std::string& hostname) {
struct evutil_addrinfo* answer = NULL;
/* Build the hints to tell getaddrinfo how to act. */
- memset(&hints, 0, sizeof(hints));
+ std::memset(&hints, 0, sizeof(hints));
hints.ai_family = AF_UNSPEC; /* v4 or v6 is fine. */
hints.ai_socktype = SOCK_STREAM; /* We want stream socket*/
hints.ai_protocol = IPPROTO_TCP; /* We want a TCP socket */
@@ -134,7 +136,7 @@ struct sockaddr* lookupNameServers(const std::string& hostname) {
continue;
}
sin = (struct sockaddr*)&sin_buf;
- memcpy(sin, ai_addr, sockaddr_size(ai_addr));
+ std::memcpy(sin, ai_addr, sockaddr_size(ai_addr));
break;
}
@@ -146,9 +148,9 @@ struct sockaddr* lookupNameServers(const std::string& hostname) {
struct sockaddr* copySocketAddress(struct sockaddr* dst, const struct sockaddr* src) {
if (src != nullptr) {
if (dst == nullptr || dst->sa_family != src->sa_family) {
- dst = (struct sockaddr*)realloc(dst, sizeof(union sockaddr_union));
+ dst = (struct sockaddr*)std::realloc(dst, sizeof(union sockaddr_union));
}
- memcpy(dst, src, sockaddr_size(src));
+ std::memcpy(dst, src, sockaddr_size(src));
} else {
free(dst);
dst = nullptr;
diff --git a/test/src/message/MessageBatchTest.cpp b/test/src/message/MessageBatchTest.cpp
index f7ba0f3..7d84592 100644
--- a/test/src/message/MessageBatchTest.cpp
+++ b/test/src/message/MessageBatchTest.cpp
@@ -19,7 +19,7 @@
#include <memory>
-#include "MQDecoder.h"
+#include "MessageDecoder.h"
#include "MQMessage.h"
#include "MessageBatch.h"
@@ -28,30 +28,26 @@ using testing::InitGoogleTest;
using testing::Return;
using rocketmq::MessageBatch;
-using rocketmq::MQDecoder;
+using rocketmq::MessageDecoder;
using rocketmq::MQMessage;
TEST(MessageBatchTest, Encode) {
- std::vector<MQMessage*> msgs;
- msgs.push_back(new MQMessage("topic", "*", "test1"));
- std::unique_ptr<MessageBatch> msgBatch(MessageBatch::generateFromList(msgs));
+ std::vector<MQMessage> msgs;
+ msgs.push_back(MQMessage("topic", "*", "test1"));
+ auto msgBatch = MessageBatch::generateFromList(msgs);
auto encodeMessage = msgBatch->encode();
- auto encodeMessage2 = MQDecoder::encodeMessages(msgs);
+ auto encodeMessage2 = MessageDecoder::encodeMessages(msgs);
EXPECT_EQ(encodeMessage, encodeMessage2);
// 20 + bodyLen(test1) + 2 + propertiesLength(TAGS:*;WAIT:true;);
EXPECT_EQ(encodeMessage.size(), 44);
- msgs.push_back(new MQMessage("topic", "*", "test2"));
- msgs.push_back(new MQMessage("topic", "*", "test3"));
- msgBatch.reset(MessageBatch::generateFromList(msgs));
+ msgs.push_back(MQMessage("topic", "*", "test2"));
+ msgs.push_back(MQMessage("topic", "*", "test3"));
+ msgBatch = MessageBatch::generateFromList(msgs);
encodeMessage = msgBatch->encode();
- encodeMessage2 = MQDecoder::encodeMessages(msgs);
+ encodeMessage2 = MessageDecoder::encodeMessages(msgs);
EXPECT_EQ(encodeMessage, encodeMessage2);
EXPECT_EQ(encodeMessage.size(), 132); // 44 * 3
-
- for (auto* msg : msgs) {
- delete msg;
- }
}
int main(int argc, char* argv[]) {
diff --git a/test/src/message/MQDecoderTest.cpp b/test/src/message/MessageDecoderTest.cpp
similarity index 87%
rename from test/src/message/MQDecoderTest.cpp
rename to test/src/message/MessageDecoderTest.cpp
index 2d43ac6..5acfc81 100644
--- a/test/src/message/MQDecoderTest.cpp
+++ b/test/src/message/MessageDecoderTest.cpp
@@ -22,11 +22,11 @@
#include <string>
#include <vector>
-#include "MQDecoder.h"
+#include "MessageDecoder.h"
#include "MQMessage.h"
#include "MQMessageConst.h"
#include "MQMessageExt.h"
-#include "MQMessageId.h"
+#include "MessageId.h"
#include "MemoryInputStream.h"
#include "MemoryOutputStream.h"
#include "MessageSysFlag.h"
@@ -42,21 +42,21 @@ using rocketmq::MemoryBlock;
using rocketmq::MemoryInputStream;
using rocketmq::MemoryOutputStream;
using rocketmq::MessageSysFlag;
-using rocketmq::MQDecoder;
+using rocketmq::MessageDecoder;
using rocketmq::MQMessage;
using rocketmq::MQMessageConst;
using rocketmq::MQMessageExt;
-using rocketmq::MQMessageId;
+using rocketmq::MessageId;
using rocketmq::RemotingCommand;
using rocketmq::SendMessageRequestHeader;
using rocketmq::UtilAll;
// TODO
TEST(MessageDecoderTest, MessageId) {
- std::string strMsgId = MQDecoder::createMessageId(rocketmq::string2SocketAddress("127.0.0.1:10091"), 1024LL);
+ std::string strMsgId = MessageDecoder::createMessageId(rocketmq::string2SocketAddress("127.0.0.1:10091"), 1024LL);
EXPECT_EQ(strMsgId, "0100007F0000276B0000000000000400");
- MQMessageId msgId = MQDecoder::decodeMessageId(strMsgId);
+ MessageId msgId = MessageDecoder::decodeMessageId(strMsgId);
EXPECT_EQ(msgId.getOffset(), 1024);
}
@@ -136,17 +136,17 @@ TEST(MessageDecoderTest, Decode) {
// 17 PROPERTIES 111=109+2
memoryOut->writeShortBigEndian(0);
- msgExt.setMsgId(MQDecoder::createMessageId(msgExt.getStoreHost(), (int64_t)msgExt.getCommitLogOffset()));
+ msgExt.setMsgId(MessageDecoder::createMessageId(msgExt.getStoreHost(), (int64_t)msgExt.getCommitLogOffset()));
auto block = memoryOut->getMemoryBlock();
- auto msgs = MQDecoder::decodes(*static_cast<MemoryBlock*>(&block));
+ auto msgs = MessageDecoder::decodes(*static_cast<MemoryBlock*>(&block));
EXPECT_EQ(msgs.size(), 1);
std::cout << msgs[0]->toString() << std::endl;
std::cout << msgExt.toString() << std::endl;
EXPECT_EQ(msgs[0]->toString(), msgExt.toString());
- msgs = MQDecoder::decodes(*static_cast<MemoryBlock*>(&block), false);
+ msgs = MessageDecoder::decodes(*static_cast<MemoryBlock*>(&block), false);
EXPECT_EQ(msgs[0]->getBody().size(), 0);
//===============================================================
@@ -173,7 +173,7 @@ TEST(MessageDecoderTest, Decode) {
std::map<std::string, std::string> properties;
properties["RocketMQ"] = "cpp-client";
properties[MQMessageConst::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX] = "123456";
- std::string props = MQDecoder::messageProperties2String(properties);
+ std::string props = MessageDecoder::messageProperties2String(properties);
memoryOut->writeShortBigEndian(props.size());
memoryOut->write(props.data(), props.size());
msgExt.setProperties(properties);
@@ -184,17 +184,17 @@ TEST(MessageDecoderTest, Decode) {
msgExt.setStoreSize(memoryOut->getDataSize());
block = memoryOut->getMemoryBlock();
- msgs = MQDecoder::decodes(*static_cast<MemoryBlock*>(&block));
+ msgs = MessageDecoder::decodes(*static_cast<MemoryBlock*>(&block));
EXPECT_EQ(msgs[0]->toString(), msgExt.toString());
}
TEST(MessageDecoderTest, MessagePropertiesAndToString) {
std::map<std::string, std::string> properties;
properties["RocketMQ"] = "cpp-client";
- std::string props = MQDecoder::messageProperties2String(properties);
+ std::string props = MessageDecoder::messageProperties2String(properties);
EXPECT_EQ(props, "RocketMQ\001cpp-client\002");
- auto properties2 = MQDecoder::string2messageProperties(props);
+ auto properties2 = MessageDecoder::string2messageProperties(props);
EXPECT_EQ(properties, properties2);
}
diff --git a/test/src/message/MQMessageIdTest.cpp b/test/src/message/MessageIdTest.cpp
similarity index 92%
rename from test/src/message/MQMessageIdTest.cpp
rename to test/src/message/MessageIdTest.cpp
index 2fd7779..aa09006 100644
--- a/test/src/message/MQMessageIdTest.cpp
+++ b/test/src/message/MessageIdTest.cpp
@@ -17,17 +17,17 @@
#include <gmock/gmock.h>
#include <gtest/gtest.h>
-#include "MQMessageId.h"
+#include "MessageId.h"
using namespace std;
using testing::InitGoogleMock;
using testing::InitGoogleTest;
using testing::Return;
-using rocketmq::MQMessageId;
+using rocketmq::MessageId;
TEST(MessageIdTest, MessageId) {
- MQMessageId msgId(rocketmq::string2SocketAddress("127.0.0.1:10091"), 1024);
+ MessageId msgId(rocketmq::string2SocketAddress("127.0.0.1:10091"), 1024);
EXPECT_EQ(rocketmq::socketAddress2String(msgId.getAddress()), "127.0.0.1:10091");
EXPECT_EQ(msgId.getOffset(), 1024);