fix(send): try to use command v2 to send messages (#225)

[fix #189]
* fix(send): try to use command v2 to send messages

* fix(send): try to use command v2 to send messages

* fix(send): try to use command v2 to send messages

* fix(send): try to use command v2 to send messages

* fix(send): try to use command v2 to send messages

* fix(send): try to use command v2 to send messages

* fix(send): try to use command v2 to send messages

Author: ShannonDing

src/MQClientAPIImpl.cpp

@@ -230,7 +230,11 @@ SendResult MQClientAPIImpl::sendMessage(const string& addr,
                                         int communicationMode,
                                         SendCallback* pSendCallback,
                                         const SessionCredentials& sessionCredentials) {
-  RemotingCommand request(SEND_MESSAGE, pRequestHeader);
+  // RemotingCommand request(SEND_MESSAGE, pRequestHeader);
+  // Using MQ V2 Protocol to end messages.
+  SendMessageRequestHeaderV2* pRequestHeaderV2 = new SendMessageRequestHeaderV2(*pRequestHeader);
+  RemotingCommand request(SEND_MESSAGE_V2, pRequestHeaderV2);
+  delete pRequestHeader;  // delete to avoid memory leak.
   string body = msg.getBody();
   request.SetBody(body.c_str(), body.length());
   request.setMsgBody(body);

src/MQClientFactory.cpp

@@ -18,6 +18,7 @@
 #include "ConsumerRunningInfo.h"
 #include "Logging.h"
 #include "MQClientManager.h"
+#include "MQVersion.h"
 #include "PullRequest.h"
 #include "Rebalance.h"
 #include "TopicPublishInfo.h"
@@ -1161,7 +1162,9 @@ ConsumerRunningInfo* MQClientFactory::consumerRunningInfo(const string& consumer
       } else {
         runningInfo->setProperty(ConsumerRunningInfo::PROP_CONSUME_TYPE, "CONSUME_ACTIVELY");
       }
-      runningInfo->setProperty(ConsumerRunningInfo::PROP_CLIENT_VERSION, "V3_1_8");  // MQVersion::s_CurrentVersion ));
+      runningInfo->setProperty(
+          ConsumerRunningInfo::PROP_CLIENT_VERSION,
+          MQVersion::GetVersionDesc(MQVersion::s_CurrentVersion));  // MQVersion::s_CurrentVersion ));
 
       return runningInfo;
     }

src/common/AsyncCallbackWrap.cpp

@@ -20,7 +20,6 @@
 #include "MQClientAPIImpl.h"
 #include "MQDecoder.h"
 #include "MQMessageQueue.h"
-#include "MQProtos.h"
 #include "PullAPIWrapper.h"
 #include "PullResultExt.h"
 #include "ResponseFuture.h"

src/common/AsyncCallbackWrap.h

@@ -20,15 +20,14 @@
 #include "AsyncArg.h"
 #include "AsyncCallback.h"
 #include "MQMessage.h"
-#include "UtilAll.h"
 #include "RemotingCommand.h"
+#include "UtilAll.h"
 
 namespace rocketmq {
 
 class ResponseFuture;
 class MQClientAPIImpl;
 class DefaultMQProducer;
-class SendMessageRequestHeader;
 //<!***************************************************************************
 enum asyncCallBackType { asyncCallbackWrap = 0, sendCallbackWrap = 1, pullCallbackWarp = 2 };
 
@@ -77,5 +76,5 @@ class PullCallbackWarp : public AsyncCallbackWrap {
 };
 
 //<!***************************************************************************
-}  //<!end namespace;
-#endif  //<! _AsyncCallbackWrap_H_
+}  // namespace rocketmq
+#endif  // __ASYNCCALLBACKWRAP_H__

src/common/MQVersion.cpp

@@ -17,15 +17,19 @@
 #include "MQVersion.h"
 
 namespace rocketmq {
-int MQVersion::s_CurrentVersion = MQVersion::V3_1_8;
+int MQVersion::s_CurrentVersion = MQVersion::V4_6_0;
+std::string MQVersion::s_CurrentLanguage = "CPP";
 
 //<!************************************************************************
-const char* MQVersion::getVersionDesc(int value) {
-  switch (value) {
-    // case V1_0_0:
-    // return "V1_0_0";
+const char* MQVersion::GetVersionDesc(int value) {
+  int currentVersion = value;
+  if (value <= V3_0_0_SNAPSHOT) {
+    currentVersion = V3_0_0_SNAPSHOT;
   }
-  return "";
+  if (value >= HIGHER_VERSION) {
+    currentVersion = HIGHER_VERSION;
+  }
+  return RocketMQCPPClientVersion[currentVersion];
 }
 //<!***************************************************************************
-}  //<!end namespace;
+}  // namespace rocketmq