feat(trace): add message trace shift for C style apis. (#281)

Author: ShannonDing

include/CCommon.h

@@ -83,6 +83,7 @@ typedef enum _CLogLevel_ {
 #endif
 
 typedef enum _CMessageModel_ { BROADCASTING, CLUSTERING } CMessageModel;
+typedef enum _CTraceModel_ { OPEN, CLOSE } CTraceModel;
 
 #ifdef __cplusplus
 }

include/CProducer.h

@@ -63,6 +63,7 @@ ROCKETMQCLIENT_API int SetProducerLogLevel(CProducer* producer, CLogLevel level)
 ROCKETMQCLIENT_API int SetProducerSendMsgTimeout(CProducer* producer, int timeout);
 ROCKETMQCLIENT_API int SetProducerCompressLevel(CProducer* producer, int level);
 ROCKETMQCLIENT_API int SetProducerMaxMessageSize(CProducer* producer, int size);
+ROCKETMQCLIENT_API int SetProducerMessageTrace(CProducer* consumer, CTraceModel openTrace);
 
 ROCKETMQCLIENT_API int SendMessageSync(CProducer* producer, CMessage* msg, CSendResult* result);
 ROCKETMQCLIENT_API int SendBatchMessage(CProducer* producer, CBatchMessage* msg, CSendResult* result);

include/CPushConsumer.h

@@ -59,6 +59,7 @@ ROCKETMQCLIENT_API int SetPushConsumerLogLevel(CPushConsumer* consumer, CLogLeve
 ROCKETMQCLIENT_API int SetPushConsumerMessageModel(CPushConsumer* consumer, CMessageModel messageModel);
 ROCKETMQCLIENT_API int SetPushConsumerMaxCacheMessageSize(CPushConsumer* consumer, int maxCacheSize);
 ROCKETMQCLIENT_API int SetPushConsumerMaxCacheMessageSizeInMb(CPushConsumer* consumer, int maxCacheSizeInMb);
+ROCKETMQCLIENT_API int SetPushConsumerMessageTrace(CPushConsumer* consumer, CTraceModel openTrace);
 
 #ifdef __cplusplus
 }

include/TransactionMQProducer.h

@@ -79,7 +79,8 @@ class ROCKETMQCLIENT_API TransactionMQProducer {
   void setLogLevel(elogLevel inputLevel);
   elogLevel getLogLevel();
   void setLogFileSizeAndNum(int fileNum, long perFileSize);  // perFileSize is MB unit
-
+  void setMessageTrace(bool messageTrace);
+  bool getMessageTrace() const;
   std::shared_ptr<TransactionListener> getTransactionListener();
   void setTransactionListener(TransactionListener* listener);
   TransactionSendResult sendMessageInTransaction(MQMessage& msg, void* arg);

src/extern/CProducer.cpp

@@ -797,6 +797,24 @@ int SetProducerMaxMessageSize(CProducer* producer, int size) {
   }
   return OK;
 }
+int SetProducerMessageTrace(CProducer* producer, CTraceModel openTrace) {
+  if (producer == NULL) {
+    return NULL_POINTER;
+  }
+  DefaultProducer* defaultMQProducer = (DefaultProducer*)producer;
+  bool messageTrace = openTrace == OPEN ? true : false;
+  try {
+    if (CAPI_C_PRODUCER_TYPE_TRANSACTION == defaultMQProducer->producerType) {
+      defaultMQProducer->innerTransactionProducer->setMessageTrace(messageTrace);
+    } else {
+      defaultMQProducer->innerProducer->setMessageTrace(messageTrace);
+    }
+  } catch (exception& e) {
+    MQClientErrorContainer::setErr(string(e.what()));
+    return PRODUCER_START_FAILED;
+  }
+  return OK;
+}
 #ifdef __cplusplus
 };
 #endif

src/extern/CPushConsumer.cpp

@@ -296,7 +296,14 @@ int SetPushConsumerLogLevel(CPushConsumer* consumer, CLogLevel level) {
   ((DefaultMQPushConsumer*)consumer)->setLogLevel((elogLevel)level);
   return OK;
 }
-
+int SetPushConsumerMessageTrace(CPushConsumer* consumer, CTraceModel openTrace) {
+  if (consumer == NULL) {
+    return NULL_POINTER;
+  }
+  bool messageTrace = openTrace == OPEN ? true : false;
+  ((DefaultMQPushConsumer*)consumer)->setMessageTrace(messageTrace);
+  return OK;
+}
 #ifdef __cplusplus
 };
 #endif

src/producer/TransactionMQProducer.cpp

@@ -160,7 +160,12 @@ void TransactionMQProducer::setUnitName(std::string unitName) {
 const std::string& TransactionMQProducer::getUnitName() const {
   return impl->getUnitName();
 }
-
+void TransactionMQProducer::setMessageTrace(bool messageTrace) {
+  impl->setMessageTrace(messageTrace);
+}
+bool TransactionMQProducer::getMessageTrace() const {
+  return impl->getMessageTrace();
+}
 std::shared_ptr<TransactionListener> TransactionMQProducer::getTransactionListener() {
   return impl->getTransactionListener();
 }

test/src/extern/CProducerTest.cpp

@@ -228,6 +228,10 @@ TEST(cProducer, info) {
   EXPECT_EQ(SetProducerSessionCredentials(cProducer, "accessKey", "secretKey", "channel"), OK);
   SessionCredentials sessionCredentials = defaultMQProducer->getSessionCredentials();
   EXPECT_EQ(sessionCredentials.getAccessKey(), "accessKey");
+
+  EXPECT_EQ(SetProducerMessageTrace(cProducer, OPEN), OK);
+  EXPECT_EQ(defaultMQProducer->getMessageTrace(), true);
+
   Mock::AllowLeak(defaultMQProducer);
 }
 

test/src/extern/CPushConsumerTest.cpp

@@ -118,6 +118,8 @@ TEST(cPushComsumer, info) {
   EXPECT_EQ(SetPushConsumerMessageModel(cpushConsumer, BROADCASTING), OK);
   EXPECT_EQ(mqPushConsumer->getMessageModel(), MessageModel::BROADCASTING);
 
+  EXPECT_EQ(SetPushConsumerMessageTrace(cpushConsumer, CLOSE), OK);
+  EXPECT_EQ(mqPushConsumer->getMessageTrace(), false);
   Mock::AllowLeak(mqPushConsumer);
 }