feat(tasks): change nsq reader to async

Author: duanhongyi

rootfs/tasks/task.py

@@ -1,22 +1,29 @@
 import os
 import json
+import logging
 import threading
 import nsq
 from functools import wraps
 
 
+logger = logging.getLogger(__name__)
+
+
 def _message_handler(message):
-    data = json.loads(message.body)
-    method = TASKS[data["target_id"]]
-    threading.Thread(
-        target=method, args=data["args"], kwargs=data["kwargs"]).start()
-    return True
+
+    def _method():
+        data = json.loads(message.body)
+        fun = TASKS[data["target_id"]]
+        fun(*data["args"], **data["kwargs"])
+        message.finish()
+    message.enable_async()
+    threading.Thread(target=_method).start()
 
 
 TASKS = {}
 NSQD_ADDRS = os.environ.get('DRYCC_NSQD_ADDRS', '127.0.0.1:4150').split(",")
-NSQ_TOPIC = os.environ.get('DRYCC_NSQ_TASKS_TOPIC', 'tasks:topic')
-NSQ_CHANNEL = os.environ.get('DRYCC_NSQ_TASKS_CHANNEL', 'tasks:channel')
+NSQ_TOPIC = os.environ.get('DRYCC_NSQ_TASKS_TOPIC', 'drycc-tasks-topic')
+NSQ_CHANNEL = os.environ.get('DRYCC_NSQ_TASKS_CHANNEL', 'drycc-tasks-channel')
 NSQD_WRITER = nsq.Writer(NSQD_ADDRS)
 NSQD_READER = nsq.Reader(
     message_handler=_message_handler,

rootfs/tasks/tests/test_task.py

@@ -26,10 +26,13 @@ def t1(name, value):
         def t2(t):
             self.assertEqual(time.time() - t > 3, True)
 
+        def callback(addr, msg):
+            self.assertEqual(msg==b'OK', True)
+
         threading.Thread(
             target=tornado.ioloop.IOLoop.current().start).start()
         time.sleep(3)
-        apply_async(t1, name="hello", value="word")
-        apply_async(t2, delay=3000, t=time.time())
-        time.sleep(6)
+        apply_async(t1, callback=callback, name="hello", value="word")
+        apply_async(t2, callback=callback, delay=3000, t=time.time())
+        time.sleep(12)
         tornado.ioloop.IOLoop.current().stop()