feature(producer) add async send message for producer
diff --git a/src/PythonWrapper.cpp b/src/PythonWrapper.cpp
index 2c5e8c2..b45fc46 100644
--- a/src/PythonWrapper.cpp
+++ b/src/PythonWrapper.cpp
@@ -21,6 +21,7 @@
#include "CProducer.h"
#include "CPushConsumer.h"
#include "PythonWrapper.h"
+#include "CMQException.h"
#include <boost/python.hpp>
#include <map>
@@ -31,6 +32,9 @@
"PYTHON_CLIENT_VERSION: " PYTHON_CLIENT_VERSION ", BUILD DATE: " PYCLI_BUILD_DATE " ";
map<CPushConsumer *, pair<PyObject *, object>> g_CallBackMap;
+map<CPushConsumer *, pair<PyObject *, object>> g_CallBackMap;
+map<void *, PyObject *> g_SendSuccessCallbackMap;
+map<void *, PyObject *> g_SendExceptionCallbackMap;
class PyThreadStateLock {
public:
@@ -153,6 +157,29 @@
return SendMessageOneway((CProducer *) producer, (CMessage *) msg);
}
+void PySendSuccessCallback(CSendResult result){
+ map<void *, PyObject *>::iterator iter = g_SendSuccessCallbackMap.begin();
+ while(iter != pCallbackMap.end()) {
+ boost::python::call<void>(iter->second, iter->first);
+ }
+}
+
+void PySendExceptionCallback(CMQException e){
+ map<void *, PyObject *>::iterator iter = g_SendExceptionCallbackMap.begin();
+ while(iter != pCallbackMap.end()) {
+ boost::python::call<void>(iter->second, iter->first);
+ }
+}
+
+int PySendMessageAsync(void *producer, void *msg, PyObject *sendSuccessCallback , PyObject *sendExceptionCallback){
+ g_SendSuccessCallbackMap[msg] = sendSuccessCallback;
+ g_SendExceptionCallbackMap[msg] = sendExceptionCallback;
+
+ return SendMessageAsync((CProducer *) producer, (CMessage *) msg, &PySendSuccessCallback, &PySendExceptionCallback);
+}
+
+
+
PySendResult PySendMessageOrderly(void *producer, void *msg, int autoRetryTimes, void *args, PyObject *queueSelector) {
PySendResult ret;
CSendResult result;
@@ -323,6 +350,11 @@
def("SetProducerInstanceName", PySetProducerInstanceName);
def("SetProducerSessionCredentials", PySetProducerSessionCredentials);
def("SendMessageSync", PySendMessageSync);
+
+ def("SendSuccessCallback", PySendSuccessCallback);
+ def("SendExceptionCallback", PySendExceptionCallback);
+ def("SendMessageAsync", PySendMessageAsync);
+
def("SendMessageOneway", PySendMessageOneway);
def("SendMessageOrderly", PySendMessageOrderly);
def("SendMessageOrderlyByShardingKey", PySendMessageOrderlyByShardingKey);
diff --git a/src/PythonWrapper.h b/src/PythonWrapper.h
index 28bfc1e..5fd46af 100644
--- a/src/PythonWrapper.h
+++ b/src/PythonWrapper.h
@@ -22,6 +22,7 @@
#include "CProducer.h"
#include "CPushConsumer.h"
#include "CPullConsumer.h"
+#include "CMQException.h"
#include <boost/python.hpp>
using namespace boost::python;
@@ -82,7 +83,12 @@
int PySetProducerSessionCredentials(void *producer, const char *accessKey, const char *secretKey, const char *channel);
PySendResult PySendMessageSync(void *producer, void *msg);
int PySendMessageOneway(void *producer, void *msg);
-// PySendResult PySendMessageOrderly(void *producer, void *msg , int autoRetryTimes, PyObject *args, PyObject *callback);
+
+void PySendSuccessCallback(CSendResult result);
+void PySendExceptionCallback(CMQException ex);
+int PySendMessageAsync(void *producer, void *msg, PyObject *sendSuccessCallback , PyObject *sendExceptionCallback);
+
+
PySendResult PySendMessageOrderly(void *producer, void *msg, int autoRetryTimes, void *args, PyObject *queueSelector);
PySendResult PySendMessageOrderlyByShardingKey(void *producer, void *msg, const char *shardingKey);
diff --git a/test/TestSendMessages.py b/test/TestSendMessages.py
index 7258d70..dc80784 100644
--- a/test/TestSendMessages.py
+++ b/test/TestSendMessages.py
@@ -214,6 +214,21 @@
DestroyMessage(msg)
print 'msg id =' + result.GetMsgId()
+def send_message_orderly_with_shardingkey(count):
+ key = 'rmq-key'
+ print 'start sending sharding key order-ly message'
+ tag = 'test'
+ for n in range(count):
+ body = 'hi rmq sharding orderly-message, now is' + str(n)
+ msg = CreateMessage(topic_orderly)
+ SetMessageBody(msg, body)
+ SetMessageKeys(msg, key)
+ SetMessageTags(msg, tag)
+
+ result = SendMessageOrderlyByShardingKey(producer, msg, 'orderId')
+ DestroyMessage(msg)
+ print 'msg id =' + result.GetMsgId()
+
def calc_which_queue_to_send(size, msg, arg): ## it is index start with 0....
return 0
