[ISSUE #275] Add trace message for pub and sub.

include/DefaultMQProducer.h

@@ -136,6 +136,8 @@ class ROCKETMQCLIENT_API DefaultMQProducer {
 
   void setUnitName(std::string unitName);
   const std::string& getUnitName() const;
+  void setMessageTrace(bool messageTrace);
+  bool getMessageTrace() const;
 
  private:
   DefaultMQProducerImpl* impl;

include/DefaultMQPushConsumer.h

@@ -132,6 +132,8 @@ class ROCKETMQCLIENT_API DefaultMQPushConsumer {
   const std::string& getUnitName() const;
 
   void setAsyncPull(bool asyncFlag);
+  void setMessageTrace(bool messageTrace);
+  bool getMessageTrace() const;
 
  private:
   DefaultMQPushConsumerImpl* impl;

src/MQClientFactory.cpp

@@ -77,6 +77,8 @@ void MQClientFactory::start() {
       m_serviceState = RUNNING;
       break;
     case RUNNING:
+      LOG_INFO("The Factory object:%s start before with now state:%d", m_clientId.c_str(), m_serviceState);
+      break;
     case SHUTDOWN_ALREADY:
     case START_FAILED:
       LOG_INFO("The Factory object:%s start failed with fault state:%d", m_clientId.c_str(), m_serviceState);

src/common/DefaultMQClient.cpp

@@ -46,6 +46,7 @@ DefaultMQClient::DefaultMQClient() {
   m_tcpConnectTimeout = 3000;        // 3s
   m_tcpTransportTryLockTimeout = 3;  // 3s
   m_unitName = "";
+  m_messageTrace = false;
 }
 
 DefaultMQClient::~DefaultMQClient() {}
@@ -216,6 +217,14 @@ const string& DefaultMQClient::getUnitName() const {
   return m_unitName;
 }
 
+bool DefaultMQClient::getMessageTrace() const {
+  return m_messageTrace;
+}
+
+void DefaultMQClient::setMessageTrace(bool mMessageTrace) {
+  m_messageTrace = mMessageTrace;
+}
+
 void DefaultMQClient::setSessionCredentials(const string& input_accessKey,
                                             const string& input_secretKey,
                                             const string& input_onsChannel) {
@@ -239,6 +248,7 @@ void DefaultMQClient::showClientConfigs() {
   LOG_WARN("PullThreadNum:%d", m_pullThreadNum);
   LOG_WARN("TcpConnectTimeout:%lld ms", m_tcpConnectTimeout);
   LOG_WARN("TcpTransportTryLockTimeout:%lld s", m_tcpTransportTryLockTimeout);
+  LOG_WARN("OpenMessageTrace:%s", m_messageTrace ? "true" : "false");
   // LOG_WARN("*****************************************************************************");
 }
 //<!************************************************************************

src/common/NameSpaceUtil.cpp

@@ -17,6 +17,7 @@
 
 #include "NameSpaceUtil.h"
 #include "Logging.h"
+#include "TraceContant.h"
 
 namespace rocketmq {
 
@@ -75,6 +76,15 @@ bool NameSpaceUtil::checkNameSpaceExistInNameServer(string nameServerAddr) {
   return false;
 }
 
+string NameSpaceUtil::withoutNameSpace(string source, string nameSpace) {
+  if (!nameSpace.empty()) {
+    auto index = source.find(nameSpace);
+    if (index != string::npos) {
+      return source.substr(index + nameSpace.length() + NAMESPACE_SPLIT_FLAG.length(), source.length());
+    }
+  }
+  return source;
+}
 string NameSpaceUtil::withNameSpace(string source, string ns) {
   if (!ns.empty()) {
     return ns + NAMESPACE_SPLIT_FLAG + source;
@@ -83,6 +93,10 @@ string NameSpaceUtil::withNameSpace(string source, string ns) {
 }
 
 bool NameSpaceUtil::hasNameSpace(string source, string ns) {
+  if (source.find(TraceContant::TRACE_TOPIC) != string::npos) {
+    LOG_DEBUG("Find Trace Topic [%s]", source.c_str());
+    return true;
+  }
   if (!ns.empty() && source.length() >= ns.length() && source.find(ns) != string::npos) {
     return true;
   }

src/common/NameSpaceUtil.h

@@ -43,6 +43,8 @@ class NameSpaceUtil {
 
   static string withNameSpace(string source, string ns);
 
+  static string withoutNameSpace(string source, string ns);
+
   static bool hasNameSpace(string source, string ns);
 };
 

src/consumer/ConsumeMessageConcurrentlyService.cpp

@@ -17,11 +17,13 @@
 #if !defined(WIN32) && !defined(__APPLE__)
 #include <sys/prctl.h>
 #endif
+
 #include "ConsumeMsgService.h"
 #include "DefaultMQPushConsumer.h"
 #include "Logging.h"
 #include "MessageAccessor.h"
 #include "UtilAll.h"
+
 namespace rocketmq {
 
 //<!************************************************************************
@@ -80,6 +82,7 @@ void ConsumeMessageConcurrentlyService::submitConsumeRequest(boost::weak_ptr<Pul
              request->m_messageQueue.toString().c_str());
   }
 }
+
 void ConsumeMessageConcurrentlyService::submitConsumeRequestLater(boost::weak_ptr<PullRequest> pullRequest,
                                                                   vector<MQMessageExt>& msgs,
                                                                   int millis) {
@@ -146,14 +149,26 @@ void ConsumeMessageConcurrentlyService::ConsumeRequest(boost::weak_ptr<PullReque
   if (request->isDropped()) {
     LOG_WARN("the pull request for %s Had been dropped before", request->m_messageQueue.toString().c_str());
     request->clearAllMsgs();  // add clear operation to avoid bad state when
-                              // dropped pullRequest returns normal
+    // dropped pullRequest returns normal
     return;
   }
   if (msgs.empty()) {
     LOG_WARN("the msg of pull result is NULL,its mq:%s", (request->m_messageQueue).toString().c_str());
     return;
   }
-
+  ConsumeMessageContext consumeMessageContext;
+  DefaultMQPushConsumerImpl* pConsumer = dynamic_cast<DefaultMQPushConsumerImpl*>(m_pConsumer);
+  if (pConsumer) {
+    if (pConsumer->getMessageTrace() && pConsumer->hasConsumeMessageHook()) {
+      consumeMessageContext.setDefaultMQPushConsumer(pConsumer);
+      consumeMessageContext.setConsumerGroup(pConsumer->getGroupName());
+      consumeMessageContext.setMessageQueue(request->m_messageQueue);
+      consumeMessageContext.setMsgList(msgs);
+      consumeMessageContext.setSuccess(false);
+      consumeMessageContext.setNameSpace(pConsumer->getNameSpace());
+      pConsumer->executeConsumeMessageHookBefore(&consumeMessageContext);
+    }
+  }
   ConsumeStatus status = CONSUME_SUCCESS;
   if (m_pMessageListener != NULL) {
     resetRetryTopic(msgs);
@@ -163,11 +178,48 @@ void ConsumeMessageConcurrentlyService::ConsumeRequest(boost::weak_ptr<PullReque
     if (m_pConsumer->isUseNameSpaceMode()) {
       MessageAccessor::withoutNameSpace(msgs, m_pConsumer->getNameSpace());
     }
-    try {
-      status = m_pMessageListener->consumeMessage(msgs);
-    } catch (...) {
-      status = RECONSUME_LATER;
-      LOG_ERROR("Consumer's code is buggy. Un-caught exception raised");
+
+    if (pConsumer->getMessageTrace() && pConsumer->hasConsumeMessageHook()) {
+      // For open trace message, consume message one by one.
+      for (size_t i = 0; i < msgs.size(); ++i) {
+        LOG_DEBUG("=====Trace Receive Messages,Topic[%s], MsgId[%s],Body[%s],RetryTimes[%d]",
+                  msgs[i].getTopic().c_str(), msgs[i].getMsgId().c_str(), msgs[i].getBody().c_str(),
+                  msgs[i].getReconsumeTimes());
+        std::vector<MQMessageExt> msgInner;
+        msgInner.push_back(msgs[i]);
+        if (status != CONSUME_SUCCESS) {
+          // all the Messages behind should be set to failed.
+          status = RECONSUME_LATER;
+          consumeMessageContext.setMsgIndex(i);
+          consumeMessageContext.setStatus("RECONSUME_LATER");
+          consumeMessageContext.setSuccess(false);
+          pConsumer->executeConsumeMessageHookAfter(&consumeMessageContext);
+          continue;
+        }
+        try {
+          status = m_pMessageListener->consumeMessage(msgInner);
+        } catch (...) {
+          status = RECONSUME_LATER;
+          LOG_ERROR("Consumer's code is buggy. Un-caught exception raised");
+        }
+        consumeMessageContext.setMsgIndex(i);  // indicate message position,not support batch consumer
+        if (status == CONSUME_SUCCESS) {
+          consumeMessageContext.setStatus("CONSUME_SUCCESS");
+          consumeMessageContext.setSuccess(true);
+        } else {
+          status = RECONSUME_LATER;
+          consumeMessageContext.setStatus("RECONSUME_LATER");
+          consumeMessageContext.setSuccess(false);
+        }
+        pConsumer->executeConsumeMessageHookAfter(&consumeMessageContext);
+      }
+    } else {
+      try {
+        status = m_pMessageListener->consumeMessage(msgs);
+      } catch (...) {
+        status = RECONSUME_LATER;
+        LOG_ERROR("Consumer's code is buggy. Un-caught exception raised");
+      }
     }
   }
 

src/consumer/ConsumeMessageHookImpl.cpp

@@ -0,0 +1,114 @@
+/*
+ * 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 "ConsumeMessageHookImpl.h"
+#include <memory>
+#include <string>
+#include "ConsumeMessageContext.h"
+#include "DefaultMQPushConsumerImpl.h"
+#include "Logging.h"
+#include "MQClientException.h"
+#include "NameSpaceUtil.h"
+#include "TraceContant.h"
+#include "TraceContext.h"
+#include "TraceTransferBean.h"
+#include "TraceUtil.h"
+#include "UtilAll.h"
+namespace rocketmq {
+
+class TraceMessageConsumeCallback : public SendCallback {
+  virtual void onSuccess(SendResult& sendResult) {
+    LOG_DEBUG("TraceMessageConsumeCallback, MsgId:[%s],OffsetMsgId[%s]", sendResult.getMsgId().c_str(),
+              sendResult.getOffsetMsgId().c_str());
+  }
+  virtual void onException(MQException& e) {}
+};
+static TraceMessageConsumeCallback* consumeTraceCallback = new TraceMessageConsumeCallback();
+std::string ConsumeMessageHookImpl::getHookName() {
+  return "RocketMQConsumeMessageHookImpl";
+}
+
+void ConsumeMessageHookImpl::executeHookBefore(ConsumeMessageContext* context) {
+  if (context == NULL || context->getMsgList().empty()) {
+    return;
+  }
+  TraceContext* traceContext = new TraceContext();
+  context->setTraceContext(traceContext);
+  traceContext->setTraceType(SubBefore);
+  traceContext->setGroupName(NameSpaceUtil::withoutNameSpace(context->getConsumerGroup(), context->getNameSpace()));
+  std::vector<TraceBean> beans;
+
+  std::vector<MQMessageExt> msgs = context->getMsgList();
+  std::vector<MQMessageExt>::iterator it = msgs.begin();
+  for (; it != msgs.end(); ++it) {
+    TraceBean bean;
+    bean.setTopic((*it).getTopic());
+    bean.setMsgId((*it).getMsgId());
+    bean.setTags((*it).getTags());
+    bean.setKeys((*it).getKeys());
+    bean.setStoreHost((*it).getStoreHostString());
+    bean.setStoreTime((*it).getStoreTimestamp());
+    bean.setBodyLength((*it).getStoreSize());
+    bean.setRetryTimes((*it).getReconsumeTimes());
+    std::string regionId = (*it).getProperty(MQMessage::PROPERTY_MSG_REGION);
+    if (regionId.empty()) {
+      regionId = TraceContant::DEFAULT_REDION;
+    }
+    traceContext->setRegionId(regionId);
+    traceContext->setTraceBean(bean);
+  }
+  traceContext->setTimeStamp(UtilAll::currentTimeMillis());
+
+  std::string topic = TraceContant::TRACE_TOPIC + traceContext->getRegionId();
+
+  TraceTransferBean ben = TraceUtil::CovertTraceContextToTransferBean(traceContext);
+  MQMessage message(topic, ben.getTransData());
+  message.setKeys(ben.getTransKey());
+
+  // send trace message async.
+  context->getDefaultMQPushConsumer()->submitSendTraceRequest(message, consumeTraceCallback);
+  return;
+}
+
+void ConsumeMessageHookImpl::executeHookAfter(ConsumeMessageContext* context) {
+  if (context == NULL || context->getMsgList().empty()) {
+    return;
+  }
+
+  std::shared_ptr<TraceContext> subBeforeContext = context->getTraceContext();
+  TraceContext subAfterContext;
+  subAfterContext.setTraceType(SubAfter);
+  subAfterContext.setRegionId(subBeforeContext->getRegionId());
+  subAfterContext.setGroupName(subBeforeContext->getGroupName());
+  subAfterContext.setRequestId(subBeforeContext->getRequestId());
+  subAfterContext.setStatus(context->getSuccess());
+  int costTime = static_cast<int>(UtilAll::currentTimeMillis() - subBeforeContext->getTimeStamp());
+  subAfterContext.setCostTime(costTime);
+  subAfterContext.setTraceBeanIndex(context->getMsgIndex());
+  TraceBean bean = subBeforeContext->getTraceBeans()[subAfterContext.getTraceBeanIndex()];
+  subAfterContext.setTraceBean(bean);
+
+  std::string topic = TraceContant::TRACE_TOPIC + subAfterContext.getRegionId();
+  TraceTransferBean ben = TraceUtil::CovertTraceContextToTransferBean(&subAfterContext);
+  MQMessage message(topic, ben.getTransData());
+  message.setKeys(ben.getTransKey());
+
+  // send trace message async.
+  context->getDefaultMQPushConsumer()->submitSendTraceRequest(message, consumeTraceCallback);
+  return;
+}
+}  // namespace rocketmq

src/consumer/ConsumeMessageHookImpl.h

@@ -0,0 +1,32 @@
+/*
+ * 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_CONSUME_MESSAGE_RPC_HOOK_IMPL_H__
+#define __ROCKETMQ_CONSUME_MESSAGE_RPC_HOOK_IMPL_H__
+
+#include <string>
+#include "ConsumeMessageContext.h"
+#include "ConsumeMessageHook.h"
+namespace rocketmq {
+class ConsumeMessageHookImpl : public ConsumeMessageHook {
+ public:
+  virtual ~ConsumeMessageHookImpl() {}
+  virtual std::string getHookName();
+  virtual void executeHookBefore(ConsumeMessageContext* context);
+  virtual void executeHookAfter(ConsumeMessageContext* context);
+};
+}  // namespace rocketmq
+#endif

src/consumer/ConsumeMessageOrderlyService.cpp

@@ -185,8 +185,27 @@ void ConsumeMessageOrderlyService::ConsumeRequest(boost::weak_ptr<PullRequest> p
           if (m_pConsumer->isUseNameSpaceMode()) {
             MessageAccessor::withoutNameSpace(msgs, m_pConsumer->getNameSpace());
           }
+          ConsumeMessageContext consumeMessageContext;
+          DefaultMQPushConsumerImpl* pConsumer = dynamic_cast<DefaultMQPushConsumerImpl*>(m_pConsumer);
+          if (pConsumer) {
+            if (pConsumer->getMessageTrace() && pConsumer->hasConsumeMessageHook()) {
+              consumeMessageContext.setDefaultMQPushConsumer(pConsumer);
+              consumeMessageContext.setConsumerGroup(pConsumer->getGroupName());
+              consumeMessageContext.setMessageQueue(request->m_messageQueue);
+              consumeMessageContext.setMsgList(msgs);
+              consumeMessageContext.setSuccess(false);
+              consumeMessageContext.setNameSpace(pConsumer->getNameSpace());
+              pConsumer->executeConsumeMessageHookBefore(&consumeMessageContext);
+            }
+          }
           ConsumeStatus consumeStatus = m_pMessageListener->consumeMessage(msgs);
           if (consumeStatus == RECONSUME_LATER) {
+            if (pConsumer) {
+              consumeMessageContext.setMsgIndex(0);
+              consumeMessageContext.setStatus("RECONSUME_LATER");
+              consumeMessageContext.setSuccess(false);
+              pConsumer->executeConsumeMessageHookAfter(&consumeMessageContext);
+            }
             if (msgs[0].getReconsumeTimes() <= 15) {
               msgs[0].setReconsumeTimes(msgs[0].getReconsumeTimes() + 1);
               request->makeMessageToCosumeAgain(msgs);
@@ -202,6 +221,12 @@ void ConsumeMessageOrderlyService::ConsumeRequest(boost::weak_ptr<PullRequest> p
               tryLockLaterAndReconsumeDelay(request, false, 5000);
             }
           } else {
+            if (pConsumer) {
+              consumeMessageContext.setMsgIndex(0);
+              consumeMessageContext.setStatus("CONSUME_SUCCESS");
+              consumeMessageContext.setSuccess(true);
+              pConsumer->executeConsumeMessageHookAfter(&consumeMessageContext);
+            }
             m_pConsumer->updateConsumeOffset(request->m_messageQueue, request->commit());
           }
         } else {