Export send batch messages api in c style

Author: jonnxu

example/CBatchProducer.c

@@ -0,0 +1,62 @@
+/*
+ * 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 <stdio.h>
+#include "CBatchMessage.h"
+#include "CCommon.h"
+#include "CMessage.h"
+#include "CProducer.h"
+#include "CSendResult.h"
+
+void StartSendMessage(CProducer* producer) {
+  int i = 0;
+  int ret_code = 0;
+  char body[128];
+  CBatchMessage* batchMessage = CreateBatchMessage("T_TestTopic");
+
+  for (i = 0; i < 10; i++) {
+    CMessage* msg = CreateMessage("T_TestTopic");
+    SetMessageTags(msg, "Test_Tag");
+    SetMessageKeys(msg, "Test_Keys");
+    memset(body, 0, sizeof(body));
+    snprintf(body, sizeof(body), "new message body, index %d", i);
+    SetMessageBody(msg, body);
+    AddMessage(batchMessage, msg);
+  }
+  CSendResult result;
+  int ok = SendBatchMessage(producer, batchMessage, &result);
+  printf("SendBatchMessage is %s .....\n", ok == 0 ? "Success" : ok == 11 ? "FAILED" : " It is null value");
+  DestroyBatchMessage(batchMessage);
+}
+
+void CreateProducerAndStartSendMessage() {
+  printf("Producer Initializing.....\n");
+  CProducer* producer = CreateProducer("Group_producer");
+  SetProducerNameServerAddress(producer, "127.0.0.1:9876");
+  StartProducer(producer);
+  printf("Producer start.....\n");
+  StartSendMessage(producer);
+  ShutdownProducer(producer);
+  DestroyProducer(producer);
+  printf("Producer Shutdown!\n");
+}
+
+int main(int argc, char* argv[]) {
+  printf("Send Batch.....\n");
+  CreateProducerAndStartSendMessage();
+  return 0;
+}

include/CBatchMessage.h

@@ -0,0 +1,36 @@
+/*
+ * 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_BATCHMESSAGE_H__
+#define __C_BATCHMESSAGE_H__
+#include "CCommon.h"
+#include "CMessage.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct CBatchMessage CBatchMessage;
+
+ROCKETMQCLIENT_API CBatchMessage* CreateBatchMessage();
+ROCKETMQCLIENT_API int AddMessage(CBatchMessage* batchMsg, CMessage* msg);
+ROCKETMQCLIENT_API int DestroyBatchMessage(CBatchMessage* batchMsg);
+
+#ifdef __cplusplus
+};
+#endif
+#endif  //__C_BATCHMESSAGE_H__

include/CProducer.h

@@ -18,6 +18,7 @@
 #ifndef __C_PRODUCER_H__
 #define __C_PRODUCER_H__
 
+#include "CBatchMessage.h"
 #include "CMessage.h"
 #include "CSendResult.h"
 #include "CMQException.h"
@@ -53,6 +54,7 @@ ROCKETMQCLIENT_API int SetProducerCompressLevel(CProducer* producer, int level);
 ROCKETMQCLIENT_API int SetProducerMaxMessageSize(CProducer* producer, int size);
 
 ROCKETMQCLIENT_API int SendMessageSync(CProducer* producer, CMessage* msg, CSendResult* result);
+ROCKETMQCLIENT_API int SendBatchMessage(CProducer* producer, CBatchMessage* msg, CSendResult* result);
 ROCKETMQCLIENT_API int SendMessageAsync(CProducer* producer,
                                         CMessage* msg,
                                         CSendSuccessCallback cSendSuccessCallback,
@@ -79,4 +81,4 @@ ROCKETMQCLIENT_API int SendMessageOrderlyAsync(CProducer* producer,
 #ifdef __cplusplus
 };
 #endif
-#endif  //__C_PRODUCER_H__
+#endif  //__C_PRODUCER_H__

src/extern/CBatchMessage.cpp

@@ -0,0 +1,59 @@
+/*
+ * 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 <vector>
+
+#include "CBatchMessage.h"
+#include "CCommon.h"
+#include "CMessage.h"
+#include "MQMessage.h"
+
+using std::vector;
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+using namespace rocketmq;
+
+CBatchMessage* CreateBatchMessage() {
+  vector<MQMessage>* msgs = new vector<MQMessage>();
+  return (CBatchMessage*)msgs;
+}
+
+int AddMessage(CBatchMessage* batchMsg, CMessage* msg) {
+  if (msg == NULL) {
+    return NULL_POINTER;
+  }
+  if (batchMsg == NULL) {
+    return NULL_POINTER;
+  }
+  MQMessage* message = (MQMessage*)msg;
+  ((vector<MQMessage>*)batchMsg)->push_back(*message);
+  return OK;
+}
+int DestroyBatchMessage(CBatchMessage* batchMsg) {
+  if (batchMsg == NULL) {
+    return NULL_POINTER;
+  }
+  delete (vector<MQMessage>*)batchMsg;
+  return OK;
+}
+
+#ifdef __cplusplus
+};
+#endif

src/extern/CProducer.cpp

@@ -17,6 +17,7 @@
 
 #include "DefaultMQProducer.h"
 #include "AsyncCallback.h"
+#include "CBatchMessage.h"
 #include "CProducer.h"
 #include "CCommon.h"
 #include "CSendResult.h"
@@ -156,6 +157,41 @@ int SendMessageSync(CProducer* producer, CMessage* msg, CSendResult* result) {
   return OK;
 }
 
+int SendBatchMessage(CProducer* producer, CBatchMessage* batcMsg, CSendResult* result) {
+  // CSendResult sendResult;
+  if (producer == NULL || batcMsg == NULL || result == NULL) {
+    return NULL_POINTER;
+  }
+  try {
+    DefaultMQProducer* defaultMQProducer = (DefaultMQProducer*)producer;
+    vector<MQMessage>* message = (vector<MQMessage>*)batcMsg;
+    SendResult sendResult = defaultMQProducer->send(*message);
+    switch (sendResult.getSendStatus()) {
+      case SEND_OK:
+        result->sendStatus = E_SEND_OK;
+        break;
+      case SEND_FLUSH_DISK_TIMEOUT:
+        result->sendStatus = E_SEND_FLUSH_DISK_TIMEOUT;
+        break;
+      case SEND_FLUSH_SLAVE_TIMEOUT:
+        result->sendStatus = E_SEND_FLUSH_SLAVE_TIMEOUT;
+        break;
+      case SEND_SLAVE_NOT_AVAILABLE:
+        result->sendStatus = E_SEND_SLAVE_NOT_AVAILABLE;
+        break;
+      default:
+        result->sendStatus = E_SEND_OK;
+        break;
+    }
+    result->offset = sendResult.getQueueOffset();
+    strncpy(result->msgId, sendResult.getMsgId().c_str(), MAX_MESSAGE_ID_LENGTH - 1);
+    result->msgId[MAX_MESSAGE_ID_LENGTH - 1] = 0;
+  } catch (exception& e) {
+    return PRODUCER_SEND_SYNC_FAILED;
+  }
+  return OK;
+}
+
 int SendMessageAsync(CProducer* producer,
                      CMessage* msg,
                      CSendSuccessCallback cSendSuccessCallback,
@@ -346,4 +382,4 @@ int SetProducerMaxMessageSize(CProducer* producer, int size) {
 }
 #ifdef __cplusplus
 };
-#endif
+#endif