Fixed bug on rabbitmq backend module, removed useless code

Author: codepr

CHANGES.md

@@ -3,6 +3,7 @@
 
 - Refactored CLI commands and logging format
 - Added factory methods to supervisors
+- Fixed bug in rabbitmq backend module
 
 ### 1.1.7
 (Apr 26, 2019)

tasq/cli/main.py

@@ -65,7 +65,7 @@ def parse_arguments():
         nargs='?'
     )
     parser.add_argument(
-        '--shared-key', '-sk',
+        '--signkey',
         help='The shared key to use to sign byte streams between clients and '
         'supervisors',
         nargs='?'
@@ -99,7 +99,7 @@ def main():
     args = parse_arguments()
     conf = get_config(args.conf)
     logger.loglevel = args.log_level or conf['log_level']
-    sign_data = conf['sign_data']
+    signkey = args.signkey or conf['signkey']
     unix_socket = conf['unix_socket']
     num_workers = args.num_workers or conf['num_workers']
     worker_type = args.worker_type or 'actor'
@@ -108,16 +108,16 @@ def main():
         push_port = args.plport or conf['zmq']['push_port']
         pull_port = args.port or conf['zmq']['pull_port']
         start_worker('zmq', worker_type, addr, push_port=push_port,
-                     pull_port=pull_port, sign_data=sign_data,
-                     unix_socket=unix_socket)
+                     pull_port=pull_port, signkey=signkey,
+                     num_workers=num_workers, unix_socket=unix_socket)
     elif args.subcommand == 'redis-worker':
         port = args.port or conf['redis']['port']
         db = args.db or conf['redis']['db']
         name = args.name or conf['redis']['name']
-        start_worker('redis', worker_type, addr, port,
-                     db=db, name=name, sign_data=sign_data)
+        start_worker('redis', worker_type, addr, port=port, db=db,
+                     name=name, num_workers=num_workers, signkey=signkey)
     elif args.subcommand == 'rabbitmq-worker':
         port = args.port or conf['rabbitmq']['port']
         name = args.name or conf['rabbitmq']['name']
-        start_worker('rabbitmq', worker_type, addr,
-                     port, name=name, sign_data=sign_data)
+        start_worker('rabbitmq', worker_type, addr, port=port,
+                     name=name, num_workers=num_workers, signkey=signkey)

tasq/jobqueue.py

@@ -81,7 +81,7 @@ def get_job(self, timeout=None):
 
         :return: A `tasq.Job` object
         """
-        return self.get(timeout)
+        return self.get(timeout=timeout)
 
     def get_result(self, block=True, timeout=None):
         return self._completed_jobs.get(block, timeout)

tasq/queue.py

@@ -45,20 +45,20 @@ class TasqQueue:
     :type store: str or None
     :param store: An URL to connect to for the results store service
 
-    :type sign_data: bool or False
-    :param sign_data: A boolean flag, sign data with a shared key
+    :type signkey: bool or False
+    :param signkey: A boolean flag, sign data with a shared key
 
     """
 
     def __init__(self, backend=u'zmq://localhost:9000',
-                 store=None, sign_data=False):
+                 store=None, signkey=None):
 
         if isinstance(backend, str):
             url = urlparse(backend)
             scheme = url.scheme or 'zmq'
             assert url.scheme in {'redis', 'zmq', 'amqp', 'unix', 'tcp'}, \
                 f"Unsupported {url.scheme}"
-            self._backend = backends[scheme].from_url(backend, sign_data)
+            self._backend = backends[scheme].from_url(backend, signkey)
         elif isinstance(backend,
                         (ZMQTasqClient, RabbitMQTasqClient, RedisTasqClient)):
             self._backend = backend

tasq/remote/backends/rabbitmq.py

@@ -51,9 +51,11 @@ def _start(self):
         channel = self._get_channel()
         channel.basic_qos(prefetch_count=1)
         if self._role == 'receiver':
+            channel.queue_declare(queue=self._queue_name, durable=True)
             channel.basic_consume(queue=self._queue_name,
                                   on_message_callback=self._get_job)
         else:
+            channel.queue_declare(queue=self._result_name, durable=True)
             channel.basic_consume(queue=self._result_name,
                                   on_message_callback=self._get_res)
         channel.start_consuming()
@@ -68,13 +70,13 @@ def put_result(self, result):
 
     def get_next_job(self, timeout=None):
         try:
-            return self._jobs.get(timeout)
+            return self._jobs.get(timeout=timeout)
         except queue.Empty:
             return None
 
     def get_available_result(self, timeout=None):
         try:
-            return self._results.get(timeout)
+            return self._results.get(timeout=timeout)
         except queue.Empty:
             return None
 

tasq/remote/client.py

@@ -55,19 +55,19 @@ class BaseTasqClient(metaclass=ABCMeta):
     :type port: int
     :param port: The port associated with the host param
 
-    :type sign_data: bool or False
-    :param sign_data: Boolean flag, sign bytes passing around through sockets
+    :type signkey: bool or False
+    :param signkey: Boolean flag, sign bytes passing around through sockets
                       if True
 
     """
 
-    def __init__(self, host, port, sign_data=False):
+    def __init__(self, host, port, signkey=None):
         # Host address of a remote supervisor to connect to
         self._host = host
         # Port for push side (outgoing) of the communication channel
         self._port = port
         # Send digital signed data
-        self._sign_data = sign_data
+        self._signkey = signkey
         # Client reference, set up the communication with a Supervisor
         self._client = self._make_client()
         # Connection flag
@@ -226,8 +226,8 @@ class ZMQTasqClient(BaseTasqClient):
     :type plport: int or None
     :param plport: The pull port to retrieve bytes from
 
-    :type sign_data: bool or False
-    :param sign_data: Boolean flag, sign bytes passing around through sockets
+    :type signkey: bool or False
+    :param signkey: Boolean flag, sign bytes passing around through sockets
                       if True
 
     :type unix_socket: bool or False
@@ -238,13 +238,13 @@ class ZMQTasqClient(BaseTasqClient):
 
     __extraparams__ = {'plport'}
 
-    def __init__(self, host, port, plport=None, sign_data=False, unix_socket=False):
+    def __init__(self, host, port, plport=None, signkey=None, unix_socket=False):
         self._plport = plport or port + 1
         # Unix socket flag, if set to true, unix sockets for interprocess
         # communication will be used and ports will be used to differentiate
         # push and pull channel
         self._unix_socket = unix_socket
-        super().__init__(host, port, sign_data)
+        super().__init__(host, port, signkey)
 
     @property
     def plport(self):
@@ -259,7 +259,7 @@ def __repr__(self):
     def _make_client(self):
         return ConnectionFactory \
             .make_client(self.host, self.port, self.plport,
-                         self._sign_data, self._unix_socket)
+                         self._signkey, self._unix_socket)
 
     def _gather_results(self):
         """Gathering subroutine, must be run in another thread to concurrently
@@ -279,17 +279,17 @@ def _gather_results(self):
                 self._log.error("Can't update result: key not found")
 
     @classmethod
-    def from_url(cls, url, sign_data=False):
+    def from_url(cls, url, signkey=None):
         u = urlparse(url)
         scheme = u.scheme or 'zmq'
         assert scheme in ('zmq', 'unix', 'tcp'), f"Unsupported {scheme}"
-        extras = {t.split('=')[0]: t.split('=')[1] for t in u.query.split('?')}
+        extras = {t.split('=')[0]: t.split('=')[1] for t in u.query.split('?') if t}
         extras = {k: v for k, v in extras.items() if k in cls.__extraparams__}
         conn_args = {
             'host': u.hostname or '127.0.0.1',
             'port': u.port or 9000,
             'plport': int(extras.get('plport', 0)),
-            'sign_data': sign_data,
+            'signkey': signkey,
             'unix_socket': scheme == 'unix'
         }
         return cls(**conn_args)
@@ -314,19 +314,19 @@ class RedisTasqClient(BaseTasqClient):
     :type name: str or redis-queue
     :param name: The name of the redis queue
 
-    :type sign_data: bool or False
-    :param sign_data: Boolean flag, sign bytes passing around through sockets
+    :type signkey: bool or False
+    :param signkey: Boolean flag, sign bytes passing around through sockets
                       if True
 
     """
 
     __extraparams__ = {'db', 'name'}
 
     def __init__(self, host='localhost', port=6379,
-                 db=0, name='redis-queue', sign_data=False):
+                 db=0, name='redis-queue', signkey=None):
         self._db = db
         self._name = name
-        super().__init__(host, port, sign_data)
+        super().__init__(host, port, signkey)
 
     @property
     def name(self):
@@ -340,7 +340,7 @@ def __repr__(self):
     def _make_client(self):
         return ConnectionFactory \
             .make_redis_client(self.host, self.port, self._db,
-                               self._name, secure=self._sign_data)
+                               self._name, signkey=self._signkey)
 
     def _gather_results(self):
         """Gathering subroutine, must be run in another thread to concurrently
@@ -378,18 +378,18 @@ def disconnect(self):
             self._is_connected = False
 
     @classmethod
-    def from_url(cls, url, sign_data=False):
+    def from_url(cls, url, signkey=None):
         u = urlparse(url)
         scheme = u.scheme or 'redis'
         assert scheme == 'redis', f"Unsupported {scheme}"
-        extras = {t.split('=')[0]: t.split('=')[1] for t in u.query.split('?')}
+        extras = {t.split('=')[0]: t.split('=')[1] for t in u.query.split('?') if t}
         extras = {k: v for k, v in extras.items() if k in cls.__extraparams__}
         conn_args = {
             'host': u.hostname or 'localhost',
             'port': u.port or 6379,
             'db': int(extras.get('db', 0)),
             'name': extras.get('name', 'redis-queue'),
-            'sign_data': sign_data
+            'signkey': signkey
         }
         return cls(**conn_args)
 
@@ -410,18 +410,18 @@ class RabbitMQTasqClient(BaseTasqClient):
     :type name: str or amqp-queue
     :param name: The name of the RabbitMQ queue
 
-    :type sign_data: bool or False
-    :param sign_data: Boolean flag, sign bytes passing around through sockets
+    :type signkey: bool or False
+    :param signkey: Boolean flag, sign bytes passing around through sockets
                       if True
 
     """
 
     __extraparams__ = {'name'}
 
     def __init__(self, host='localhost', port=5672,
-                 name='amqp-queue', sign_data=False):
+                 name='amqp-queue', signkey=None):
         self._name = name
-        super().__init__(host, port, sign_data)
+        super().__init__(host, port, signkey)
 
     @property
     def name(self):
@@ -435,14 +435,16 @@ def __repr__(self):
     def _make_client(self):
         return ConnectionFactory \
             .make_rabbitmq_client(self.host, self.port, 'sender',
-                                  self._name, secure=self._sign_data)
+                                  self._name, signkey=self._signkey)
 
     def _gather_results(self):
         """Gathering subroutine, must be run in another thread to concurrently
         listen for results and store them into a dedicated dictionary
         """
         while True:
             job_result = self._client.recv_result()
+            if not job_result:
+                continue
             self._log.debug("Gathered result: %s", job_result)
             try:
                 self._results[job_result.name].set_result(job_result)
@@ -468,17 +470,17 @@ def disconnect(self):
             self._is_connected = False
 
     @classmethod
-    def from_url(cls, url, sign_data=False):
+    def from_url(cls, url, signkey=None):
         u = urlparse(url)
         scheme = u.scheme or 'amqp'
         assert scheme == 'amqp', f"Unsupported {scheme}"
-        extras = {t.split('=')[0]: t.split('=')[1] for t in u.query.split('?')}
+        extras = {t.split('=')[0]: t.split('=')[1] for t in u.query.split('?') if t}
         extras = {k: v for k, v in extras.items() if k in cls.__extraparams__}
         conn_args = {
             'host': u.hostname or 'localhost',
             'port': u.port or 5672,
             'name': extras.get('name', 'amqp-queue'),
-            'sign_data': sign_data
+            'signkey': signkey
         }
         return cls(**conn_args)