feat(transaction): add transaction message for C APIs

include/Arg_helper.h

@@ -37,6 +37,6 @@ class ROCKETMQCLIENT_API Arg_helper {
 };
 
 //<!***************************************************************************
-}  //<!end namespace;
+}  // namespace rocketmq
 
 #endif  //<!_ARG_HELPER_H_;

include/AsyncCallback.h

@@ -54,5 +54,5 @@ class ROCKETMQCLIENT_API PullCallback : public AsyncCallback {
   virtual void onException(MQException& e) = 0;
 };
 //<!***************************************************************************
-}  //<!end namespace;
+}  // namespace rocketmq
 #endif

include/BatchMessage.h

@@ -17,13 +17,13 @@
 
 #ifndef __BATCHMESSAGE_H__
 #define __BATCHMESSAGE_H__
-#include "MQMessage.h"
 #include <string>
+#include "MQMessage.h"
 namespace rocketmq {
 class BatchMessage : public MQMessage {
  public:
   static std::string encode(std::vector<MQMessage>& msgs);
   static std::string encode(MQMessage& message);
 };
-}
+}  // namespace rocketmq
 #endif

include/CBatchMessage.h

@@ -31,6 +31,6 @@ ROCKETMQCLIENT_API int AddMessage(CBatchMessage* batchMsg, CMessage* msg);
 ROCKETMQCLIENT_API int DestroyBatchMessage(CBatchMessage* batchMsg);
 
 #ifdef __cplusplus
-};
+}
 #endif
 #endif  //__C_BATCHMESSAGE_H__

include/CCommon.h

@@ -81,6 +81,6 @@ typedef enum _CLogLevel_ {
 typedef enum _CMessageModel_ { BROADCASTING, CLUSTERING } CMessageModel;
 
 #ifdef __cplusplus
-};
+}
 #endif
 #endif  //__C_COMMON_H__

include/CErrorMessage.h

@@ -27,6 +27,6 @@ extern "C" {
 ROCKETMQCLIENT_API const char* GetLatestErrorMessage();  // Return the last error message
 
 #ifdef __cplusplus
-};
+}
 #endif
 #endif  //__CERROR_MESSAGE_H__

include/CMQException.h

@@ -35,6 +35,6 @@ typedef struct _CMQException_ {
 } CMQException;
 
 #ifdef __cplusplus
-};
+}
 #endif
 #endif

include/CMessage.h

@@ -35,8 +35,14 @@ ROCKETMQCLIENT_API int SetMessageBody(CMessage* msg, const char* body);
 ROCKETMQCLIENT_API int SetByteMessageBody(CMessage* msg, const char* body, int len);
 ROCKETMQCLIENT_API int SetMessageProperty(CMessage* msg, const char* key, const char* value);
 ROCKETMQCLIENT_API int SetDelayTimeLevel(CMessage* msg, int level);
+ROCKETMQCLIENT_API const char* GetOriginMessageTopic(CMessage* msg);
+ROCKETMQCLIENT_API const char* GetOriginMessageTags(CMessage* msg);
+ROCKETMQCLIENT_API const char* GetOriginMessageKeys(CMessage* msg);
+ROCKETMQCLIENT_API const char* GetOriginMessageBody(CMessage* msg);
+ROCKETMQCLIENT_API const char* GetOriginMessageProperty(CMessage* msg, const char* key);
+ROCKETMQCLIENT_API int GetOriginDelayTimeLevel(CMessage* msg);
 
 #ifdef __cplusplus
-};
+}
 #endif
 #endif  //__C_MESSAGE_H__

include/CMessageExt.h

@@ -44,6 +44,6 @@ ROCKETMQCLIENT_API long long GetMessageCommitLogOffset(CMessageExt* msgExt);
 ROCKETMQCLIENT_API long long GetMessagePreparedTransactionOffset(CMessageExt* msgExt);
 
 #ifdef __cplusplus
-};
+}
 #endif
 #endif  //__C_MESSAGE_EXT_H__

include/CMessageQueue.h

@@ -31,6 +31,6 @@ typedef struct _CMessageQueue_ {
 } CMessageQueue;
 
 #ifdef __cplusplus
-};
+}
 #endif
 #endif  //__C_MESSAGE_H__

include/CProducer.h

@@ -21,7 +21,9 @@
 #include "CBatchMessage.h"
 #include "CMQException.h"
 #include "CMessage.h"
+#include "CMessageExt.h"
 #include "CSendResult.h"
+#include "CTransactionStatus.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -34,10 +36,14 @@ typedef void (*CSendSuccessCallback)(CSendResult result);
 typedef void (*CSendExceptionCallback)(CMQException e);
 typedef void (*COnSendSuccessCallback)(CSendResult result, CMessage* msg, void* userData);
 typedef void (*COnSendExceptionCallback)(CMQException e, CMessage* msg, void* userData);
+typedef CTransactionStatus (*CLocalTransactionCheckerCallback)(CProducer* producer, CMessageExt* msg, void* data);
+typedef CTransactionStatus (*CLocalTransactionExecutorCallback)(CProducer* producer, CMessage* msg, void* data);
 
 ROCKETMQCLIENT_API CProducer* CreateProducer(const char* groupId);
 ROCKETMQCLIENT_API CProducer* CreateOrderlyProducer(const char* groupId);
-ROCKETMQCLIENT_API CProducer* CreateTransactionProducer(const char* groupId);
+ROCKETMQCLIENT_API CProducer* CreateTransactionProducer(const char* groupId,
+                                                        CLocalTransactionCheckerCallback callback,
+                                                        void* userData);
 ROCKETMQCLIENT_API int DestroyProducer(CProducer* producer);
 ROCKETMQCLIENT_API int StartProducer(CProducer* producer);
 ROCKETMQCLIENT_API int ShutdownProducer(CProducer* producer);
@@ -90,8 +96,12 @@ ROCKETMQCLIENT_API int SendMessageOrderlyByShardingKey(CProducer* producer,
                                                        CMessage* msg,
                                                        const char* shardingKey,
                                                        CSendResult* result);
-
+ROCKETMQCLIENT_API int SendMessageTransaction(CProducer* producer,
+                                              CMessage* msg,
+                                              CLocalTransactionExecutorCallback callback,
+                                              void* userData,
+                                              CSendResult* result);
 #ifdef __cplusplus
-};
+}
 #endif
 #endif  //__C_PRODUCER_H__

include/CPullConsumer.h

@@ -56,6 +56,6 @@ Pull(CPullConsumer* consumer, const CMessageQueue* mq, const char* subExpression
 ROCKETMQCLIENT_API int ReleasePullResult(CPullResult pullResult);
 
 #ifdef __cplusplus
-};
+}
 #endif
 #endif  //__C_PUSH_CONSUMER_H__

include/CPullResult.h

@@ -43,6 +43,6 @@ typedef struct _CPullResult_ {
 } CPullResult;
 
 #ifdef __cplusplus
-};
+}
 #endif
 #endif  //__C_PULL_RESULT_H__

include/CPushConsumer.h

@@ -60,6 +60,6 @@ ROCKETMQCLIENT_API int SetPushConsumerMaxCacheMessageSize(CPushConsumer* consume
 ROCKETMQCLIENT_API int SetPushConsumerMaxCacheMessageSizeInMb(CPushConsumer* consumer, int maxCacheSizeInMb);
 
 #ifdef __cplusplus
-};
+}
 #endif
 #endif  //__C_PUSH_CONSUMER_H__

include/CSendResult.h

@@ -37,6 +37,6 @@ typedef struct _SendResult_ {
 } CSendResult;
 
 #ifdef __cplusplus
-};
+}
 #endif
 #endif  //__C_PRODUCER_H__

include/CTransactionStatus.h

@@ -0,0 +1,33 @@
+/*
+ * 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 __C_TRANSACTION_STATUS_H__
+#define __C_TRANSACTION_STATUS_H__
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+typedef enum E_CTransactionStatus {
+  E_COMMIT_TRANSACTION = 0,
+  E_ROLLBACK_TRANSACTION = 1,
+  E_UNKNOWN_TRANSACTION = 2,
+} CTransactionStatus;
+
+#ifdef __cplusplus
+}
+#endif
+#endif  //__C_TRANSACTION_STATUS_H__

include/ConsumeType.h

@@ -28,9 +28,9 @@ enum ConsumeType {
 //<!***************************************************************************
 enum ConsumeFromWhere {
   /**
-  *new consumer will consume from end offset of queue,
-  * and then consume from last consumed offset of queue follow-up
-  */
+   *new consumer will consume from end offset of queue,
+   * and then consume from last consumed offset of queue follow-up
+   */
   CONSUME_FROM_LAST_OFFSET,
 
   // @Deprecated
@@ -40,14 +40,14 @@ enum ConsumeFromWhere {
   // @Deprecated
   CONSUME_FROM_MAX_OFFSET,
   /**
-  *new consumer will consume from first offset of queue,
-  * and then consume from last consumed offset of queue follow-up
-  */
+   *new consumer will consume from first offset of queue,
+   * and then consume from last consumed offset of queue follow-up
+   */
   CONSUME_FROM_FIRST_OFFSET,
   /**
-  *new consumer will consume from the queue offset specified by timestamp,
-  * and then consume from last consumed offset of queue follow-up
-  */
+   *new consumer will consume from the queue offset specified by timestamp,
+   * and then consume from last consumed offset of queue follow-up
+   */
   CONSUME_FROM_TIMESTAMP,
 };
 
@@ -57,5 +57,5 @@ enum MessageModel {
   CLUSTERING,
 };
 //<!***************************************************************************
-}  //<!end namespace;
+}  // namespace rocketmq
 #endif

include/DefaultMQProducer.h

@@ -18,11 +18,11 @@
 #ifndef __DEFAULTMQPRODUCER_H__
 #define __DEFAULTMQPRODUCER_H__
 
+#include "BatchMessage.h"
 #include "MQMessageQueue.h"
 #include "MQProducer.h"
 #include "RocketMQClient.h"
 #include "SendResult.h"
-#include "BatchMessage.h"
 
 namespace rocketmq {
 //<!***************************************************************************
@@ -113,5 +113,5 @@ class ROCKETMQCLIENT_API DefaultMQProducer : public MQProducer {
   int m_retryTimes4Async;
 };
 //<!***************************************************************************
-}  //<!end namespace;
+}  // namespace rocketmq
 #endif

include/DefaultMQPullConsumer.h

@@ -60,21 +60,21 @@ class ROCKETMQCLIENT_API DefaultMQPullConsumer : public MQConsumer {
 
   void registerMessageQueueListener(const std::string& topic, MQueueListener* pListener);
   /**
-  * pull msg from specified queue, if no msg in queue, return directly
-  *
-  * @param mq
-  *            specify the pulled queue
-  * @param subExpression
-  *            set filter expression for pulled msg, broker will filter msg actively
-  *            Now only OR operation is supported, eg: "tag1 || tag2 || tag3"
-  *            if subExpression is setted to "null" or "*"��all msg will be subscribed
-  * @param offset
-  *            specify the started pull offset
-  * @param maxNums
-  *            specify max msg num by per pull
-  * @return
-  *            accroding to PullResult
-  */
+   * pull msg from specified queue, if no msg in queue, return directly
+   *
+   * @param mq
+   *            specify the pulled queue
+   * @param subExpression
+   *            set filter expression for pulled msg, broker will filter msg actively
+   *            Now only OR operation is supported, eg: "tag1 || tag2 || tag3"
+   *            if subExpression is setted to "null" or "*"��all msg will be subscribed
+   * @param offset
+   *            specify the started pull offset
+   * @param maxNums
+   *            specify max msg num by per pull
+   * @return
+   *            accroding to PullResult
+   */
   virtual PullResult pull(const MQMessageQueue& mq, const std::string& subExpression, int64 offset, int maxNums);
   virtual void pull(const MQMessageQueue& mq,
                     const std::string& subExpression,
@@ -83,21 +83,21 @@ class ROCKETMQCLIENT_API DefaultMQPullConsumer : public MQConsumer {
                     PullCallback* pPullCallback);
 
   /**
-  * pull msg from specified queue, if no msg, broker will suspend the pull request 20s
-  *
-  * @param mq
-  *            specify the pulled queue
-  * @param subExpression
-  *            set filter expression for pulled msg, broker will filter msg actively
-  *            Now only OR operation is supported, eg: "tag1 || tag2 || tag3"
-  *            if subExpression is setted to "null" or "*"��all msg will be subscribed
-  * @param offset
-  *            specify the started pull offset
-  * @param maxNums
-  *            specify max msg num by per pull
-  * @return
-  *            accroding to PullResult
-  */
+   * pull msg from specified queue, if no msg, broker will suspend the pull request 20s
+   *
+   * @param mq
+   *            specify the pulled queue
+   * @param subExpression
+   *            set filter expression for pulled msg, broker will filter msg actively
+   *            Now only OR operation is supported, eg: "tag1 || tag2 || tag3"
+   *            if subExpression is setted to "null" or "*"��all msg will be subscribed
+   * @param offset
+   *            specify the started pull offset
+   * @param maxNums
+   *            specify max msg num by per pull
+   * @return
+   *            accroding to PullResult
+   */
   PullResult pullBlockIfNotFound(const MQMessageQueue& mq, const std::string& subExpression, int64 offset, int maxNums);
   void pullBlockIfNotFound(const MQMessageQueue& mq,
                            const std::string& subExpression,
@@ -107,20 +107,20 @@ class ROCKETMQCLIENT_API DefaultMQPullConsumer : public MQConsumer {
 
   virtual ConsumerRunningInfo* getConsumerRunningInfo() { return NULL; }
   /**
-  * ��ȡ���ѽ��ȣ�����-1��ʾ����
-  *
-  * @param mq
-  * @param fromStore
-  * @return
-  */
+   * ��ȡ���ѽ��ȣ�����-1��ʾ����
+   *
+   * @param mq
+   * @param fromStore
+   * @return
+   */
   int64 fetchConsumeOffset(const MQMessageQueue& mq, bool fromStore);
   /**
-  * ����topic��ȡMessageQueue���Ծ��ⷽʽ�����ڶ����Ա֮�����
-  *
-  * @param topic
-  *            ��ϢTopic
-  * @return ���ض��м���
-  */
+   * ����topic��ȡMessageQueue���Ծ��ⷽʽ�����ڶ����Ա֮�����
+   *
+   * @param topic
+   *            ��ϢTopic
+   * @return ���ض��м���
+   */
   void fetchMessageQueuesInBalance(const std::string& topic, std::vector<MQMessageQueue> mqs);
 
   // temp persist consumer offset interface, only valid with
@@ -155,5 +155,5 @@ class ROCKETMQCLIENT_API DefaultMQPullConsumer : public MQConsumer {
   PullAPIWrapper* m_pPullAPIWrapper;
 };
 //<!***************************************************************************
-}  //<!end namespace;
+}  // namespace rocketmq
 #endif

include/DefaultMQPushConsumer.h

@@ -153,5 +153,5 @@ class ROCKETMQCLIENT_API DefaultMQPushConsumer : public MQConsumer {
   std::unique_ptr<boost::thread> m_pullmsgThread;
 };
 //<!***************************************************************************
-}  //<!end namespace;
+}  // namespace rocketmq
 #endif