add args field for python callback

ifplusor · ifplusor · commit c31733c8a095 · 2018-12-28T11:16:44.000+08:00
diff --git a/.gitignore b/.gitignore
@@ -1,2 +1,8 @@
 .idea/
-cmake-build-debug/
+cmake-build-*/
+
+*.pyc
+*.so
+
+bin/
+
diff --git a/doc/Introduction.md b/doc/Introduction.md
@@ -73,8 +73,10 @@
 ----------
 ## How to use
 - set LD_LIBRARY_PATH
-  ``````
+  ```
   export LD_LIBRARY_PATH=/usr/local/lib
+  ```
+  
 - import module
   ```
   from librocketmqclientpython import *
@@ -90,28 +92,28 @@
 - producer must invoke following interface:
   ```
   - producer = CreateProducer("please_rename_unique_group_name");
-  - SetProducerNameServerAddress(producer,"please_rename_unique_name_server")
+  - SetProducerNameServerAddress(producer, "please_rename_unique_name_server")
   - StartProducer(producer)
-  - SendMessageSync(producer,msg)
+  - SendMessageSync(producer, msg)
   - ShutdownProducer(producer)
   - DestroyProducer(producer)
   ```
 - how to consumer messages
   ```
-  - def consumerMessage(msg):
-  - topic = GetMessageTopic(msg)
-  - body = GetMessageBody(msg)
-  - tags = GetMessageTags(msg)
-  - msgid = GetMessageId(msg)
-  - handle message
-  - return 0
+  - def consumerMessage(msg, args):
+  -     topic = GetMessageTopic(msg)
+  -     body = GetMessageBody(msg)
+  -     tags = GetMessageTags(msg)
+  -     msgid = GetMessageId(msg)
+  -     # handle message...
+  -     return 0
   ```
 - pushconsumer must invoke following interface:
   ```
   - consumer = CreatePushConsumer("please_rename_unique_group_name_1");
-  - SetPushConsumerNameServerAddress(consumer,"please_rename_unique_name_server")
+  - SetPushConsumerNameServerAddress(consumer, "please_rename_unique_name_server")
   - Subscribe(consumer, "your_topic", "*")
-  - RegisterMessageCallback(consumer,consumerMessage)
+  - RegisterMessageCallback(consumer, consumerMessage, args)
   - StartPushConsumer(consumer)
   - ShutdownPushConsumer(consumer)
   - DestroyPushConsumer(consumer)
@@ -122,3 +124,4 @@
   - python testProducer.py
 - push consumer
   - python testConsumer.py
+
diff --git a/doc/api-doc/consumer-push.md b/doc/api-doc/consumer-push.md
@@ -31,13 +31,14 @@
     topic: topic name
     tag: topic tag
 
-* RegisterMessageCallback(consumer, pyCallBack) <br />
+* RegisterMessageCallback(consumer, pyCallBack, pyArgs) <br />
   - function description<br />
     set callback for push consumer instance <br />
 
   - input <br />
     consumer: consumer intance<br />
-    pyCallBack: py callback method. when message pulled, they would be send to a pyCallback method
+    pyCallBack: py callback method. when message pulled, they would be send to a pyCallback method<br />
+    pyArgs: the arguments will be passed to pyCallBack
 
 * SetPushConsumerThreadCount(consumer, threadCount)
   - function description<br />
diff --git a/sample/testConsumer.py b/sample/testConsumer.py
@@ -18,8 +18,10 @@
 import base
 import time
 from librocketmqclientpython import *
+
 totalMsg = 0
-def consumerMessage(msg):
+
+def consumerMessage(msg, args):
      global totalMsg
      totalMsg += 1
      print(">>ConsumerMessage Called:",totalMsg)
@@ -33,11 +35,12 @@ def consumerMessage(msg):
 
 consumer = CreatePushConsumer("awtTest_Producer_Python_Test")
 print(consumer)
-SetPushConsumerNameServerAddress(consumer,"172.17.0.2:9876")
-SetPushConsumerThreadCount(consumer,1)
+SetPushConsumerNameServerAddress(consumer, "172.17.0.2:9876")
+SetPushConsumerThreadCount(consumer, 1)
 Subscribe(consumer, "T_TestTopic", "*")
-RegisterMessageCallback(consumer,consumerMessage)
+RegisterMessageCallback(consumer, consumerMessage, None)
 StartPushConsumer(consumer)
+
 i = 1
 while i <= 60:
     print(i)
diff --git a/src/PythonWrapper.cpp b/src/PythonWrapper.cpp
@@ -30,7 +30,7 @@ using namespace std;
 const char *VERSION =
         "PYTHON_CLIENT_VERSION: " PYTHON_CLIENT_VERSION ", BUILD DATE: " PYCLI_BUILD_DATE " ";
 
-map<CPushConsumer *, PyObject *> g_CallBackMap;
+map<CPushConsumer *, pair<PyObject *, object>> g_CallBackMap;
 
 class PyThreadStateLock {
 public:
@@ -163,7 +163,14 @@ void *PyCreatePushConsumer(const char *groupId) {
     return (void *) CreatePushConsumer(groupId);
 }
 int PyDestroyPushConsumer(void *consumer) {
-    return DestroyPushConsumer((CPushConsumer *) consumer);
+    CPushConsumer *consumerInner = (CPushConsumer *) consumer;
+    map<CPushConsumer *, pair<PyObject *, object>>::iterator iter;
+    iter = g_CallBackMap.find(consumerInner);
+    if (iter != g_CallBackMap.end()) {
+        UnregisterMessageCallback(consumerInner);
+        g_CallBackMap.erase(iter);
+    }
+    return DestroyPushConsumer(consumerInner);
 }
 int PyStartPushConsumer(void *consumer) {
     return StartPushConsumer((CPushConsumer *) consumer);
@@ -181,29 +188,27 @@ int PySetPushConsumerNameServerDomain(void *consumer, const char *domain){
 int PySubscribe(void *consumer, const char *topic, const char *expression) {
     return Subscribe((CPushConsumer *) consumer, topic, expression);
 }
-int PyRegisterMessageCallback(void *consumer, PyObject *pCallback) {
+int PyRegisterMessageCallback(void *consumer, PyObject *pCallback, object args) {
     CPushConsumer *consumerInner = (CPushConsumer *) consumer;
-    g_CallBackMap[consumerInner] = pCallback;
+    g_CallBackMap[consumerInner] = make_pair(pCallback, std::move(args));
     return RegisterMessageCallback(consumerInner, &PythonMessageCallBackInner);
 }
 
 int PythonMessageCallBackInner(CPushConsumer *consumer, CMessageExt *msg) {
-
-    class PyThreadStateLock PyThreadLock;
-    PyMessageExt message;
-    message.pMessageExt = msg;
-    map<CPushConsumer *, PyObject *>::iterator iter;
+    PyThreadStateLock PyThreadLock;  // ensure hold GIL, before call python callback
+    PyMessageExt message = { .pMessageExt = msg };
+    map<CPushConsumer *, pair<PyObject *, object>>::iterator iter;
     iter = g_CallBackMap.find(consumer);
     if (iter != g_CallBackMap.end()) {
-        PyObject * pCallback = iter->second;
+        pair<PyObject *, object> callback = iter->second;
+        PyObject * pCallback = callback.first;
+        object& args = callback.second;
         if (pCallback != NULL) {
-            int status =
-                    boost::python::call<int>(pCallback, message);
+            int status = boost::python::call<int>(pCallback, message, args);
             return status;
         }
     }
     return 1;
-
 }
 
 int PySetPushConsumerThreadCount(void *consumer, int threadCount) {
diff --git a/src/PythonWrapper.h b/src/PythonWrapper.h
@@ -89,7 +89,7 @@ int PyShutdownPushConsumer(void *consumer);
 int PySetPushConsumerNameServerAddress(void *consumer, const char *namesrv);
 int PySetPushConsumerNameServerDomain(void *consumer, const char *domain);
 int PySubscribe(void *consumer, const char *topic, const char *expression);
-int PyRegisterMessageCallback(void *consumer, PyObject *pCallback);
+int PyRegisterMessageCallback(void *consumer, PyObject *pCallback, object args);
 int PythonMessageCallBackInner(CPushConsumer *consumer, CMessageExt *msg);
 int PySetPushConsumerThreadCount(void *consumer, int threadCount);
 int PySetPushConsumerMessageBatchMaxSize(void *consumer, int batchSize);
diff --git a/test/TestConsumeMessages.py b/test/TestConsumeMessages.py
@@ -34,7 +34,7 @@ def sigint_handler(signum, frame):
     sys.exit(0)
 
 
-def consumer_message(msg):
+def consumer_message(msg, args):
     global totalMsg
     totalMsg += 1
     print 'total count %d' % totalMsg
@@ -55,7 +55,7 @@ def init_producer(_group, _topic, _tag):
     SetPushConsumerNameServerAddress(consumer, name_srv)
     SetPushConsumerThreadCount(consumer, 1)
     Subscribe(consumer, _topic, _tag)
-    RegisterMessageCallback(consumer, consumerMessage)
+    RegisterMessageCallback(consumer, consumer_message, None)
     StartPushConsumer(consumer)
     print 'consumer is ready...'
     return consumer

-Original file line number
+Diff line change
@@ @@ -1,2 +1,8 @@ @@
 .idea/
 -cmake-build-debug/
 +cmake-build-*/
++
 +*.pyc
 +*.so
++
 +bin/
++