feat(producer): add method for orderly message sending by shardingkey

include/CProducer.h

@@ -19,9 +19,9 @@
 #define __C_PRODUCER_H__
 
 #include "CBatchMessage.h"
+#include "CMQException.h"
 #include "CMessage.h"
 #include "CSendResult.h"
-#include "CMQException.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -34,6 +34,7 @@ typedef void (*CSendSuccessCallback)(CSendResult result);
 typedef void (*CSendExceptionCallback)(CMQException e);
 
 ROCKETMQCLIENT_API CProducer* CreateProducer(const char* groupId);
+ROCKETMQCLIENT_API CProducer* CreateOrderlyProducer(const char* groupId);
 ROCKETMQCLIENT_API int DestroyProducer(CProducer* producer);
 ROCKETMQCLIENT_API int StartProducer(CProducer* producer);
 ROCKETMQCLIENT_API int ShutdownProducer(CProducer* producer);
@@ -77,6 +78,10 @@ ROCKETMQCLIENT_API int SendMessageOrderlyAsync(CProducer* producer,
                                                void* arg,
                                                CSendSuccessCallback cSendSuccessCallback,
                                                CSendExceptionCallback cSendExceptionCallback);
+ROCKETMQCLIENT_API int SendMessageOrderlyByShardingKey(CProducer* producer,
+                                                       CMessage* msg,
+                                                       const char* shardingKey,
+                                                       CSendResult* result);
 
 #ifdef __cplusplus
 };

src/extern/CProducer.cpp

@@ -17,8 +17,8 @@
 
 #include "CProducer.h"
 #include <string.h>
+#include <functional>
 #include <typeindex>
-#include <string.h>
 #include <typeinfo>
 #include "AsyncCallback.h"
 #include "CBatchMessage.h"
@@ -28,13 +28,25 @@
 #include "CSendResult.h"
 #include "DefaultMQProducer.h"
 #include "MQClientErrorContainer.h"
+#include "UtilAll.h"
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 using namespace rocketmq;
 using namespace std;
 
+class SelectMessageQueueInner : public MessageQueueSelector {
+ public:
+  MQMessageQueue select(const std::vector<MQMessageQueue>& mqs, const MQMessage& msg, void* arg) {
+    int index = 0;
+    std::string shardingKey = rocketmq::UtilAll::to_string((char*)arg);
+
+    index = std::hash<std::string>{}(shardingKey) % mqs.size();
+    return mqs[index % mqs.size()];
+  }
+};
+
 class SelectMessageQueue : public MessageQueueSelector {
  public:
   SelectMessageQueue(QueueSelectorCallback callback) { m_pCallback = callback; }
@@ -86,6 +98,10 @@ CProducer* CreateProducer(const char* groupId) {
   DefaultMQProducer* defaultMQProducer = new DefaultMQProducer(groupId);
   return (CProducer*)defaultMQProducer;
 }
+
+CProducer* CreateOrderlyProducer(const char* groupId) {
+  return CreateProducer(groupId);
+}
 int DestroyProducer(CProducer* pProducer) {
   if (pProducer == NULL) {
     return NULL_POINTER;
@@ -308,7 +324,28 @@ int SendMessageOrderly(CProducer* producer,
   }
   return OK;
 }
-
+int SendMessageOrderlyByShardingKey(CProducer* producer, CMessage* msg, const char* shardingKey, CSendResult* result) {
+  if (producer == NULL || msg == NULL || shardingKey == NULL || result == NULL) {
+    return NULL_POINTER;
+  }
+  DefaultMQProducer* defaultMQProducer = (DefaultMQProducer*)producer;
+  MQMessage* message = (MQMessage*)msg;
+  try {
+    // Constructing SelectMessageQueue objects through function pointer callback
+    int retryTimes = 3;
+    SelectMessageQueueInner selectMessageQueue;
+    SendResult sendResult = defaultMQProducer->send(*message, &selectMessageQueue, (void*)shardingKey, retryTimes);
+    // Convert SendStatus to CSendStatus
+    result->sendStatus = CSendStatus((int)sendResult.getSendStatus());
+    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) {
+    MQClientErrorContainer::setErr(string(e.what()));
+    return PRODUCER_SEND_ORDERLY_FAILED;
+  }
+  return OK;
+}
 int SetProducerGroupName(CProducer* producer, const char* groupName) {
   if (producer == NULL) {
     return NULL_POINTER;