You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by di...@apache.org on 2019/09/21 03:49:55 UTC
[rocketmq-client-cpp] 01/03: feat(producer): add method for orderly
message sending by shardingkey
This is an automated email from the ASF dual-hosted git repository.
dinglei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-client-cpp.git
commit 544eaea4bf2f12c2e6825f1964372ad9f3ab9e01
Author: ShannonDing <li...@163.com>
AuthorDate: Fri Sep 20 21:10:57 2019 +0800
feat(producer): add method for orderly message sending by shardingkey
---
include/CProducer.h | 3 +++
src/extern/CProducer.cpp | 37 ++++++++++++++++++++++++++++++++++++-
2 files changed, 39 insertions(+), 1 deletion(-)
diff --git a/include/CProducer.h b/include/CProducer.h
index 6b00575..7d533f9 100644
--- a/include/CProducer.h
+++ b/include/CProducer.h
@@ -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,8 @@ 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
};
diff --git a/src/extern/CProducer.cpp b/src/extern/CProducer.cpp
index 28b43d4..b2ebf5c 100644
--- a/src/extern/CProducer.cpp
+++ b/src/extern/CProducer.cpp
@@ -35,6 +35,15 @@ extern "C" {
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;
+ return mqs[index % mqs.size()];
+ }
+};
+
class SelectMessageQueue : public MessageQueueSelector {
public:
SelectMessageQueue(QueueSelectorCallback callback) { m_pCallback = callback; }
@@ -86,6 +95,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 +321,29 @@ 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;