More work for named queues

- Rename `WORKERS` setting to `CONCURRENT_JOBS` for more clarity. Added this
  setting to the command line options, the ini .conf & default settings conf.py
  files.
- Added support for JSON style arrays in INI config.
- Added support for weighted named queues. The style for the setting is
  [[weight, "name"], [weight, "name"]]. Configured in both the INI and command
  line for job manager. Added documentation.
- Updated the spec for the INFORM message. Weights are sent with the queue
  names. If there are no weights specified they will be given the default value
  of 0.
- updated Router.queues to a list from a deque so that it can be sorted
  by priority more easily.

15 files changed, 466 insertions, 246 deletions

diff --git a/bin/emq-jobmanager b/bin/emq-jobmanager
index 82c2e9b..4071ada 100755
--- a/bin/emq-jobmanager
+++ b/bin/emq-jobmanager
@@ -21,12 +21,19 @@ from eventmq.jobmanager import JobManager
 from eventmq import conf
 
 if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description='Listen for requests')
-    parser.add_argument('--config', '-c', type=str, nargs='?',
+    parser = argparse.ArgumentParser(description='Listen for job requests and '
+                                     'manage their execution')
+    parser.add_argument('--broker-addr', '-B', type=str, nargs='?',
+                        help='manually specify the broker address to connect '
+                        'to in order to receive jobs')
+    parser.add_argument('--config', '-C', type=str, nargs='?',
                         help='manually specify the location of eventmq.conf')
     parser.add_argument('--queues', '-Q', type=str, nargs='+',
                         help='space separated list of queue names to listen '
                              'on')
+    parser.add_argument('--jobs', '-J', type=int, nargs='?',
+                        help='the max number of concurrent jobs to manage at '
+                        'a time')
 
     args = parser.parse_args()
 
@@ -39,5 +46,8 @@ if __name__ == "__main__":
     if queues:
         queues = ','.join(queues)
 
-    j = JobManager(queues=queues)
-    j.jobmanager_main()
+    broker_addr = args.broker_addr
+    concurrent_jobs = args.jobs
+
+    j = JobManager(queues=queues, concurrent_jobs=concurrent_jobs)
+    j.jobmanager_main(broker_addr=broker_addr)
diff --git a/docs/conf.py b/docs/conf.py
index 864bbcf..aa23956 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -16,6 +16,7 @@ import sys
 import os
 import shlex
 
+import eventmq   # for __version__
 # If extensions (or modules to document with autodoc) are in another directory,
 # add these directories to sys.path here. If the directory is relative to the
 # documentation root, use os.path.abspath to make it absolute, like shown here.
@@ -52,15 +53,15 @@ master_doc = 'index'
 
 # General information about the project.
 project = u'EventMQ'
-copyright = u'2015, eventboard.io'
-author = u'eventboard.io'
+copyright = u'2016, eventboard.io'
+author = u'EventMQ Contributors'
 
 # The version info for the project you're documenting, acts as replacement for
 # |version| and |release|, also used in various other places throughout the
 # built documents.
 #
 # The short X.Y version.
-version = '0'
+version = eventmq.__version__
 # The full version, including alpha/beta/rc tags.
 release = '0'
 
diff --git a/docs/protocol.rst b/docs/protocol.rst
index 4c2aff0..c341403 100644
--- a/docs/protocol.rst
+++ b/docs/protocol.rst
@@ -71,7 +71,7 @@ FRAME  Value          Description
 1      eMQP/1.0       Protocol version
 2      REQUEST        command
 3      _MSGID_        A unique id for the msg
-4      _QUEUE_NAME_   the name of the queue the worker belongs to
+4      _QUEUE_NAME_   the name of the queue the request should be sent to
 5      _HEADERS_      dictionary of headers. can be an empty set
 6      _MSG_          The message to send
 ====== ============== ===========
@@ -85,7 +85,7 @@ FRAME  Value          Description
 1      eMQP/1.0       Protocol version
 2      PUBLISH        command
 3      _MSGID_        A unique id for the msg
-4      _TOPIC_NAME_   the name of the queue the worker belongs to
+4      _TOPIC_NAME_   the name of the topic this message should be published across
 5      _HEADERS_      csv list of headers
 6      _MSG_          The message to send
 ====== ============== ===========
@@ -99,7 +99,7 @@ FRAME   Value         Description
 1      eMQP/1.0       Protocol version
 2      SCHEDULE       command
 3      _MSGID_        A unique id for the msg
-4      _TOPIC_NAME_   name of queue that the job should run in
+4      _QUEUE_NAME_   name of queue that the job should run in
 5      _HEADERS_      csv list of headers for this message
 6      _MSG_          The message to send
 ====== ============== ===========
@@ -113,7 +113,7 @@ FRAME   Value         Description
 1      eMQP/1.0       Protocol version
 2      UNSCHEDULE     command
 3      _MSGID_        A unique id for the msg
-4      _TOPIC_NAME_   ignored for this command, broadcasted to all queues
+4      _QUEUE_NAME_   ignored for this command, broadcasted to all queues
 5      _HEADERS_      csv list of headers for this message
 6      _MSG_          The message to send
 ====== ============== ===========
@@ -129,7 +129,7 @@ FRAME   Value         Description
 1      eMQP/1.0       Protocol version
 2      INFORM         command
 3      _MSGID_        A unique id for the msg
-4      _QUEUE_NAME_   csv seperated names of queue the worker belongs to
+4                     Queues. Unused for scheduler
 5      scheduler      type of peer connecting
 ====== ============== ===========
 
@@ -144,7 +144,7 @@ FRAME   Value         Description
 1      eMQP/1.0       Protocol version
 2      INFORM         command
 3      _MSGID_        A unique id for the msg
-4      _QUEUE_NAME_   csv seperated names of queue the worker belongs to.
+4      _QUEUES_       csv seperated arrays containing an int and a string for weight and name. e.g. [40, 'email']
 5      worker         type of peer connecting
 ====== ============== ===========
 
@@ -203,7 +203,6 @@ Heartbeating
  * If the worker detects that the broker disconnected it SHOULD restart the conversation.
  * If the broker detects that a worker has disconnected it should stop sending it a message of any type.
  * If the scheduler detects that the broker disconnects it SHOULD restart the conversation.
- * If the broker detects that a scheduler has disconnected it should ??????????.
 
 REQUEST Headers
 ---------------
diff --git a/docs/settings_file.rst b/docs/settings_file.rst
index 07d2338..b0bc36c 100644
--- a/docs/settings_file.rst
+++ b/docs/settings_file.rst
@@ -15,16 +15,34 @@ Scheduler
 Job Manager
 ***********
 
+concurrent_jobs
+===============
+Default: 4
+
+This is the number of concurrent jobs the indiviudal job manager should execute
+at a time. If you are using the multiprocess or threading model this number
+becomes important as you will want to control the load on your server. If the
+load equals the number of cores on the server, processes will begin waiting for
+cpu cycles and things will begin to slow down.
+
+A safe number to choose if your jobs block a lot would be (2 * cores). If your
+jobs are cpu intensive you will want to set this number to the number of cores
+you have or (cores - 1) to leave cycles for the os and other processes. This is
+something that will have to be tuned based on the jobs that are
+running. Grouping similar jobs in named queues will help you tune this number.
+
 queues
 ======
-Default: default
-
-Comma seperated list of queues to process jobs for. Example:
-``queues=high,med,low,default``. The philosophy taken for this list is each job
-manager should have a single primary queue. This queue is the first in the list
-(in the case of the example ``high`` is the primary queue). Subsequent queues
-are queues that this job manager should help out with should jobs be backed up,
-and there are no primary queue jobs to take care of.
+Default: (10, default)
+
+Semi-colon seperated list of queues to process jobs for with thier
+weights. Example: ``queues=(10, data_process); (15, email)``.  With these
+weights and the ``CONCURRENT_JOBS`` setting, you should be able to tune managers
+running jobs locally pretty efficiently. If you have a larger box with a weight
+of 50 on q1 and 8 concurrent jobs and a smaller box with a weight 30 and 4
+concurrent jobs, the q1 jobs will be sent to the large box until it is no longer
+accepting jobs. At this point jobs will start to be sent to the next highest
+number until the large box is ready to accept another q1 job.
 
 .. note::
 
diff --git a/etc/eventmq.docker.conf b/etc/eventmq.docker.conf
index a61364c..55fee92 100644
--- a/etc/eventmq.docker.conf
+++ b/etc/eventmq.docker.conf
@@ -1,4 +1,10 @@
-[settings]
+[global]
+# Enable message output at different stages in the app.
+super_debug = true
+
+# Hide the heartbeat logs when super_debug is enabled. Showing them will generate a lot of messages.
+hide_heartbeat_logs = True
+
 frontend_addr=tcp://0.0.0.0:47291
 backend_addr=tcp://0.0.0.0:47290
 worker_addr=tcp://eventmq:47290
@@ -9,5 +15,6 @@ scheduler_addr=tcp://eventmq:47291
 [scheduler]
 
 [jobmanager]
-queues=one,two,three,default
-worker_addr=tcp://127.0.0.1:47290
\ No newline at end of file
+worker_addr=tcp://127.0.0.1:47290
+queues=[[50,"google"], [40,"pushes"], [10,"default"]]
+concurrent_jobs=2
\ No newline at end of file
diff --git a/eventmq/conf.py b/eventmq/conf.py
index 68fe7de..0c4b7cc 100644
--- a/eventmq/conf.py
+++ b/eventmq/conf.py
@@ -1,13 +1,40 @@
-# SUPER_DEBUG basically enables more debugging logs. specifically of messages
-# at different levels in the application
-SUPER_DEBUG = True
-# Don't show HEARTBEAT message when debug logging is enabled
+# This file is part of eventmq.
+#
+# eventmq is free software: you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# Software Foundation, either version 2.1 of the License, or (at your option)
+# any later version.
+#
+# eventmq is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with eventmq.  If not, see <http://www.gnu.org/licenses/>.
+"""
+:mod:`conf` -- Settings Definitions
+===================================
+"""
+
+#: SUPER_DEBUG basically enables more debugging logs. Specifically the messages
+#: at different levels in the application.
+#: Default: False
+SUPER_DEBUG = False
+
+#: Don't show HEARTBEAT message when debug logging is enabled
+#: Default: True
 HIDE_HEARTBEAT_LOGS = True
 
 # When a queue name isn't specified use this queue name for the default. It
 # would be a good idea to have a handful of workers listening on this queue
 # unless you're positive that everything specifies a queue with workers.
 DEFAULT_QUEUE_NAME = 'default'
+DEFAULT_QUEUE_WEIGHT = 10
+
+# Default queues for the Job Manager to listen on. The values here should match
+# the values defined on the router.
+QUEUES = [(DEFAULT_QUEUE_WEIGHT, DEFAULT_QUEUE_NAME), ]
 
 # {{{Job Manager
 # How long should we wait before retrying to connect to a broker?
@@ -30,13 +57,14 @@ BACKEND_ADDR = 'tcp://127.0.0.1:47290'
 SCHEDULER_ADDR = 'tcp://127.0.0.1:47291'
 WORKER_ADDR = 'tcp://127.0.0.1:47290'
 
+# How many jobs should the job manager concurrently handle?
+CONCURRENT_JOBS = 4
+HWM = 10000
+
 # Redis settings
 RQ_HOST = 'localhost'
 RQ_PORT = 6379
 RQ_DB = 0
 RQ_PASSWORD = ''
-WORKERS = 4
-HWM = 10000
 
-QUEUES = '{}'.format(DEFAULT_QUEUE_NAME)
 # }}}
diff --git a/eventmq/jobmanager.py b/eventmq/jobmanager.py
index 60501fb..332b178 100644
--- a/eventmq/jobmanager.py
+++ b/eventmq/jobmanager.py
@@ -66,10 +66,11 @@ class JobManager(HeartbeatMixin, EMQPService):
         logger.info('Initializing JobManager {}...'.format(self.name))
 
         #: keep track of workers
-        self.workers = Pool(processes=conf.WORKERS)
+        concurrent_jobs = kwargs.pop('concurrent_jobs', conf.CONCURRENT_JOBS)
+        self.workers = Pool(processes=concurrent_jobs)
 
         #: List of queues that this job manager is listening on
-        self.queues = kwargs.pop('queues', None)
+        self.queues = kwargs.pop('queues', conf.QUEUES)
 
         if not kwargs.pop('skip_signal', False):
             # handle any sighups by reloading config
@@ -91,7 +92,7 @@ class JobManager(HeartbeatMixin, EMQPService):
         """
         # Acknowledgment has come
         # Send a READY for each available worker
-        for i in range(0, conf.WORKERS):
+        for i in range(0, conf.CONCURRENT_JOBS):
             self.send_ready()
 
         while True:
@@ -173,19 +174,26 @@ class JobManager(HeartbeatMixin, EMQPService):
         import_settings(section='jobmanager')
         self.start(addr=conf.WORKER_ADDR)
 
-    def jobmanager_main(self):
+    def jobmanager_main(self, broker_addr=None):
         """
         Kick off jobmanager with logging and settings import
+
+        Args:
+            broker_addr (str): The address of the broker to connect to.
         """
         setup_logger('')
         import_settings()
         import_settings(section='jobmanager')
 
-        # If this manager was passed explicit queues, favor those.
+        # If this manager was passed explicit options, favor those
         if self.queues:
             conf.QUEUES = self.queues
 
-        self.start(addr=conf.WORKER_ADDR, queues=self.queues or conf.QUEUES)
+        if broker_addr:
+            conf.WORKER_ADDR = broker_addr
+
+        self.start(addr=conf.WORKER_ADDR,
+                   queues=conf.QUEUES)
 
 
 def jobmanager_main():
diff --git a/eventmq/router.py b/eventmq/router.py
index e91d74f..7aa0d9b 100644
--- a/eventmq/router.py
+++ b/eventmq/router.py
@@ -18,6 +18,7 @@
 Routes messages to workers (that are in named queues).
 """
 from copy import copy
+import json  # deserialize queues in on_inform. should be refactored
 import logging
 import signal
 
@@ -29,7 +30,7 @@ from .utils.messages import (
     fwd_emqp_router_message as fwdmsg,
     parse_router_message
 )
-from .utils import zero_index_cmp
+from .utils import tuplify, zero_index_cmp
 from .utils.settings import import_settings
 from .utils.devices import generate_device_name
 from .utils.timeutils import monotonic, timestamp
@@ -76,8 +77,8 @@ class Router(HeartbeatMixin):
         #: here.
         #:
         #: **Keys**
-        #:  * ``queues``: list() of queues the worker belongs to. The highest
-        #     priority queue should come first.
+        #:  * ``queues``: list() of queue names and prioritiess the worker
+        #:    belongs to. e.g. (10, 'default')
         #:  * ``hb``: monotonic timestamp of the last received message from
         #:    worker
         #:  * ``available_slots``: int count of jobs this manager can still
@@ -246,10 +247,10 @@ class Router(HeartbeatMixin):
         queue_names = msg[0]
         client_type = msg[1]
 
-        if not queue_names:
-            queues = ('default', )
+        if not queue_names:  # Ideally, this matches some workers
+            queues = [(conf.DEFAULT_QUEUE_WEIGHT, conf.DEFAULT_QUEUE_NAME), ]
         else:
-            queues = queue_names.split(',')
+            queues = list(map(tuplify, json.loads(queue_names)))
 
         logger.info('Received INFORM request from {} (type: {})'.format(
             sender, client_type))
@@ -274,16 +275,17 @@ class Router(HeartbeatMixin):
             msgid (str): Unique identifier for this message
             msg: The actual message that was sent
         """
+        queue_names = self.workers[sender]['queues']
+
         # if there are waiting messages for the queues this worker is a member
         # of, then reply back with the oldest waiting message, otherwise just
         # add the worker to the list of available workers.
         # Note: This is only taking into account the queue the worker is
         # returning from, and not other queue_names that might have had
         # messages waiting even longer.
-        queue_names = self.workers[sender]['queues']
-
         # Assumes the highest priority queue comes first
-        for queue_name in queue_names:
+        for queue in queue_names:
+            queue_name = queue[1]
             if queue_name in self.waiting_messages.keys():
                 logger.debug('Found waiting message in the %s waiting_messages'
                              ' queue' % queue_name)
@@ -292,12 +294,13 @@ class Router(HeartbeatMixin):
                 fwdmsg(self.outgoing, sender, msg)
 
                 # It is easier to check if a key exists rather than the len of
-                # a key if it exists elsewhere, so if that was the last message
-                # remove the queue
+                # a key's value if it exists elsewhere, so if that was the last
+                # message remove the queue
                 if len(self.waiting_messages[queue_name]) == 0:
                     logger.debug('No more messages in waiting_messages queue '
                                  '%s. Removing from list...' % queue_name)
                     del self.waiting_messages[queue_name]
+
                 # the message has been forwarded so short circuit that way the
                 # manager isn't reslotted
                 return
@@ -392,17 +395,9 @@ class Router(HeartbeatMixin):
                                 conf.HEARTBEAT_TIMEOUT))
 
                 # Remove the worker from the actual queues
-                for i in range(0, len(self.workers[worker_id]['queues'])):
-                    if i == 0:
-                        priority = 10
-                    else:
-                        priority = 0
-
-                    queue = self.workers[worker_id]['queues'][i]
-
+                for queue in self.workers[worker_id]['queues']:
                     try:
-                        self.queues[queue].remove((priority, worker_id))
-                        break
+                        self.queues[queue[1]].remove((queue[0], worker_id))
                     except KeyError:
                         # This queue disappeared for some reason
                         continue
@@ -426,6 +421,10 @@ class Router(HeartbeatMixin):
             raise TypeError('type of `queue` parameter not one of (list, '
                             'tuple). got {}'.format(type(queues)))
 
+        if worker_id in self.workers:
+            logger.warning('Worker id already found in `workers`. Overwriting '
+                           'data')
+
         # Add the worker to our worker dict
         self.workers[worker_id] = {}
         self.workers[worker_id]['queues'] = tuple(queues)
@@ -433,21 +432,15 @@ class Router(HeartbeatMixin):
         self.workers[worker_id]['available_slots'] = 0
 
         # Define priorities. First element is the highest priority
-        for i in range(len(queues)):
-            if i == 0:
-                priority = 10
-            else:
-                priority = 0
+        for q in queues:
+            if q[1] not in self.queues:
+                self.queues[q[1]] = list()
 
-            if queues[i] not in self.queues:
-                self.queues[queues[i]] = EMQdeque()
+            self.queues[q[1]].append((q[0], worker_id))
+            self.queues[q[1]] = self.prioritize_queue_list(self.queues[q[1]])
 
-            self.queues[queues[i]].append((priority, worker_id))
-
-            self.queues[queues[i]] = \
-                self.prioritize_queue_list(self.queues[queues[i]])
         logger.debug('Added worker {} to the queues {}'.format(
-                     worker_id, queues))
+                 worker_id, queues))
 
     def get_available_worker(self, queue_name=conf.DEFAULT_QUEUE_NAME):
         """
@@ -480,15 +473,19 @@ class Router(HeartbeatMixin):
                 # pop the next job manager id & check if it has a worker slot
                 # if it doesn't add it to popped_workers to be added back to
                 # self.queues after the loop
-                worker = self.queues[queue_name].popleft()
+                worker = self.queues[queue_name].pop(0)
+
                 # LRU when sorted later by appending
                 popped_workers.append(worker)
+
                 if self.workers[worker[1]]['available_slots'] > 0:
                     worker_addr = worker[1]
                     break
+
             except KeyError:
-                # This should only happen if worker[1] is missing:
-                #  - available slots is pre set to 0 self.add_worker
+                # This should only happen if worker[1] is missing 1 from
+                # self.workers because:
+                #  - available slots initialized to 0 self.add_worker()
                 #  - we already checked that self.queues[queue_name] exists
                 logger.error("Worker {} not found for queue {}".format(
                     worker, queue_name))
@@ -497,14 +494,16 @@ class Router(HeartbeatMixin):
                                  worker, queue_name
                              ))
                 continue
+
             except IndexError:
                 # worker[1] should exist if it follows the (priority, id) fmt
-                logger.error("Invalid worker format in self.queues {}".format(
-                    worker
-                ))
+                logger.error("Invalid priority/worker format in self.queues "
+                             "{}".format(worker))
                 continue
+        else:
+            # No more queues to try
+            pass
 
-        # Should always evaluate to true
         if popped_workers:
             self.queues[queue_name].extend(popped_workers)
             self.queues[queue_name] = self.prioritize_queue_list(
@@ -553,12 +552,6 @@ class Router(HeartbeatMixin):
         """
         self.workers[worker_id]['available_slots'] += 1
 
-    def queue_message(self, msg):
-        """
-        Add a message to the queue for processing later
-        """
-        raise NotImplementedError()
-
     def process_client_message(self, original_msg, depth=0):
         """
         Args:
@@ -691,13 +684,9 @@ class Router(HeartbeatMixin):
             IndexError - There was no 0-index element.
 
         Returns:
-            sorted :class:`EMQdeque` with largest priorites being indexed
-            smaller. E.g. ((20,'a' ), (14, 'b'), ('12', c))
+            decsending order list. E.g. ((20, 'a'), (14, 'b'), (12, 'c'))
         """
-        return EMQdeque(
-            initial=sorted(unprioritized_iterable,
-                           cmp=zero_index_cmp,
-                           reverse=True))
+        return sorted(unprioritized_iterable, cmp=zero_index_cmp, reverse=True)
 
     def sighup_handler(self, signum, frame):
         """
diff --git a/eventmq/tests/test_jobmanager.py b/eventmq/tests/test_jobmanager.py
index 37014cb..6d3c663 100644
--- a/eventmq/tests/test_jobmanager.py
+++ b/eventmq/tests/test_jobmanager.py
@@ -23,8 +23,6 @@ ADDR = 'inproc://pour_the_rice_in_the_thing'
 
 
 class TestCase(unittest.TestCase):
-    jm = None
-
     def test__setup(self):
         jm = jobmanager.JobManager(name='RuckusBringer')
         self.assertEqual(jm.name, 'RuckusBringer')
@@ -62,8 +60,8 @@ class TestCase(unittest.TestCase):
 
         jm._start_event_loop()
 
-        # send conf.WORKERS ready messages
-        self.assertEqual(conf.WORKERS, send_ready_mock.call_count)
+        # send int(conf.CONCURRENT_JOBS) ready messages
+        self.assertEqual(conf.CONCURRENT_JOBS, send_ready_mock.call_count)
 
         process_msg_mock.assert_called_with(
             sender_mock.return_value)
@@ -86,6 +84,7 @@ class TestCase(unittest.TestCase):
             callback=jm.worker_done,
             func=run_mock)
 
+# Other Tests
     @mock.patch('eventmq.jobmanager.JobManager.start')
     @mock.patch('eventmq.jobmanager.import_settings')
     @mock.patch('eventmq.jobmanager.Sender.rebuild')
@@ -111,7 +110,7 @@ class TestCase(unittest.TestCase):
     @mock.patch('eventmq.jobmanager.import_settings')
     @mock.patch('eventmq.jobmanager.setup_logger')
     def test_jobmanager_main(self, setup_logger_mock, import_settings_mock,
-                        start_mock):
+                             start_mock):
         jm = jobmanager.JobManager()
 
         jm.jobmanager_main()
@@ -125,7 +124,7 @@ class TestCase(unittest.TestCase):
         start_mock.assert_called_with(addr=conf.WORKER_ADDR,
                                       queues=conf.QUEUES)
 
-        jm.queues = ('derp', 'blurp')
+        jm.queues = ((10, 'derp'), (0, 'blurp'))
         jm.jobmanager_main()
 
         start_mock.assert_called_with(addr=conf.WORKER_ADDR,
diff --git a/eventmq/tests/test_router.py b/eventmq/tests/test_router.py
index 6349159..82fe0b5 100644
--- a/eventmq/tests/test_router.py
+++ b/eventmq/tests/test_router.py
@@ -44,33 +44,77 @@ class TestCase(unittest.TestCase):
 
     @mock.patch('eventmq.router.Router.send_ack')
     @mock.patch('eventmq.router.Router.add_worker')
-    def test_on_inform_worker_defaut_queue(self, add_worker_mock,
-                                           send_ack_mock):
+    def test_on_inform_worker(self, add_worker_mock, send_ack_mock):
+        sender_id = 'omgsenderid19'
+        queues = '[[32, "top"], [23, "drop"], [12, "shop"]]'
+        inform_msgid = 'msg31'
+
+        self.router.on_inform(
+            sender_id, inform_msgid, [queues, 'worker'])
+
+        self.router.send_ack.assert_called_with(
+            self.router.outgoing, sender_id, inform_msgid)
+
+        self.router.add_worker.assert_called_with(
+            sender_id, [(32, 'top'), (23, 'drop'), (12, 'shop')])
+
+    @mock.patch('eventmq.router.Router.send_ack')
+    @mock.patch('eventmq.router.Router.add_worker')
+    def test_on_inform_worker_default_queue(self, add_worker_mock,
+                                            send_ack_mock):
+        # Test on_inform when no queue is specified
         sender_id = 'omgsender18'
         queues = ''
         inform_msgid = 'msg29'
 
-        self.router.on_inform(sender_id, inform_msgid,
-                              [queues, 'worker'])
+        self.router.on_inform(
+            sender_id, inform_msgid, [queues, constants.CLIENT_TYPE.worker])
 
-        self.router.send_ack.assert_called_with(self.router.outgoing,
-                                                sender_id, inform_msgid)
-        self.router.add_worker.assert_called_with(sender_id, ('default',))
+        self.router.send_ack.assert_called_with(
+            self.router.outgoing, sender_id, inform_msgid)
+        self.router.add_worker.assert_called_with(
+            sender_id, [(10, 'default'), ])
 
-    @mock.patch('eventmq.router.Router.send_ack')
-    @mock.patch('eventmq.router.Router.add_worker')
-    def test_on_inform_worker(self, add_worker_mock, send_ack_mock):
-        sender_id = 'omgsenderid19'
-        queues = 'top,drop,shop'
-        inform_msgid = 'msg31'
+    # @mock.patch('eventmq.router.Router.prioritize_queue_list')
+    def test_add_worker(self):
+        worker1_id = 'w1'
+        worker2_id = 'w2'
 
-        self.router.on_inform(sender_id, inform_msgid,
-                              [queues, 'worker'])
+        queues1 = [(10, 'top'), (9, 'drop'), (8, 'shop')]
+        queues2 = [(10, 'default'), (9, 'shop'), (8, 'top')]
 
-        self.router.send_ack.assert_called_with(self.router.outgoing,
-                                                sender_id, inform_msgid)
-        self.router.add_worker.assert_called_with(sender_id,
-                                                  queues.split(','))
+        self.router.add_worker(worker1_id, queues=queues1)
+        self.router.add_worker(worker2_id, queues=queues2)
+        # added to the list of workers
+        self.assertIn(worker1_id, self.router.workers)
+        self.assertIn(worker2_id, self.router.workers)
+        self.assertGreater(self.router.workers[worker1_id]['hb'], 0)
+        # no slots yet
+        self.assertEqual(self.router.workers[worker1_id]['available_slots'], 0)
+
+        # aware of the queues
+        self.assertEqual(3, len(self.router.workers[worker1_id]['queues']))
+        self.assertIn((10, 'top'), self.router.workers[worker1_id]['queues'])
+        self.assertIn((9, 'drop'), self.router.workers[worker1_id]['queues'])
+        self.assertIn((8, 'shop'), self.router.workers[worker1_id]['queues'])
+
+        # Worker2
+        self.assertIn((10, 'default'),
+                      self.router.workers[worker2_id]['queues'])
+        self.assertIn((9, 'shop'), self.router.workers[worker2_id]['queues'])
+        self.assertIn((8, 'top'), self.router.workers[worker2_id]['queues'])
+
+        self.assertIn((10, worker1_id), self.router.queues['top'])
+        self.assertIn((9, worker1_id), self.router.queues['drop'])
+        self.assertIn((8, worker1_id), self.router.queues['shop'])
+
+        self.assertIn((10, worker2_id), self.router.queues['default'])
+        self.assertIn((9, worker2_id), self.router.queues['shop'])
+        self.assertIn((8, worker2_id), self.router.queues['top'])
+
+    def test_add_worker_invalid_queues(self):
+        with self.assertRaises(TypeError):
+            self.router.add_worker('83902', 8902)
 
     @mock.patch('eventmq.utils.messages.generate_msgid')
     def test_send_ack(self, generate_msgid_mock):
@@ -115,12 +159,12 @@ class TestCase(unittest.TestCase):
         self.assertEqual(self.router._meta['last_sent_heartbeat'], 0)
         self.router.workers = {
             'w1': {
-                'queues': ['default', ],
+                'queues': [(10, 'default'), ],
                 'hb': 123.2,
                 'available_slots': 3,
             },
             'w2': {
-                'queues': ['default', ],
+                'queues': [(10, 'not-default'), ],
                 'hb': 123.2,
                 'available_slots': 2,
             }
@@ -131,10 +175,27 @@ class TestCase(unittest.TestCase):
         # is very hard to mock)
         self.assertGreater(self.router._meta['last_sent_heartbeat'], 0)
 
-        self.router.send_heartbeat.assert_has_calls(
+        send_heartbeat_mock.assert_has_calls(
             [mock.call(self.router.outgoing, 'w1'),
              mock.call(self.router.outgoing, 'w2')], any_order=True)
 
+    @mock.patch('eventmq.router.Router.send_heartbeat')
+    def test_send_schedulers_heartbeats(self, send_hb_mock):
+        scheduler_id = 's39'
+        self.assertEqual(self.router._meta['last_sent_scheduler_heartbeat'], 0)
+
+        self.router.schedulers = {
+            scheduler_id: {
+                'hb': 0,
+            }
+        }
+
+        self.router.send_schedulers_heartbeats()
+
+        self.assertGreater(
+            self.router._meta['last_sent_scheduler_heartbeat'], 0)
+        send_hb_mock.assert_called_with(self.router.incoming, scheduler_id)
+
     def test_on_disconnect(self):
         self.assertFalse(self.router.received_disconnect)
         self.router.on_disconnect('msg1', 'derp')
@@ -152,14 +213,14 @@ class TestCase(unittest.TestCase):
 
         self.router.workers = {
             worker_id: {
-                'queues': ['default', ],
+                'queues': [(10, 'default'), ],
                 'hb': 123.2,
                 'available_slots': 3,
             },
         }
 
         self.router.waiting_messages['default'] = EMQdeque(
-            initial=[waiting_msg])
+            initial=[waiting_msg, ])
 
         self.router.on_ready(worker_id, msgid, msg)
 
@@ -171,8 +232,10 @@ class TestCase(unittest.TestCase):
 
     @mock.patch('eventmq.router.fwdmsg')
     @mock.patch('eventmq.router.Router.requeue_worker')
-    def test_on_ready_prioritized_queue(self, requeue_worker_mock,
-                                        fwdmsg_mock):
+    def test_on_ready_multpile_queues(self, requeue_worker_mock,
+                                      fwdmsg_mock):
+        # Test that if messages are waiting on multiple queues, they are
+        # dispatched immediatly after a READY message.
         worker1_id = 'w1'
         worker2_id = 'w2'
 
@@ -188,53 +251,61 @@ class TestCase(unittest.TestCase):
 
         self.router.workers = {
             worker1_id: {
-                'queues': ['kun', 'blu'],
+                'queues': [(10, 'kun'), (0, 'blu')],
                 'hb': 123.2,
                 'available_slots': 0,
             },
             worker2_id: {
-                'queues': ['blu', 'kun'],
+                'queues': [(10, 'blu'), (0, 'kun')],
                 'hb': 123.2,
                 'available_slots': 0
             }
         }
 
         self.router.queues = {
-            'kun': EMQdeque(initial=[(10, worker1_id), (0, worker2_id)]),
-            'blu': EMQdeque(initial=[(10, worker2_id), (0, worker1_id)])
-
+            'kun': [(10, worker1_id), (0, worker2_id)],
+            'blu': [(10, worker2_id), (0, worker1_id)],
         }
 
         self.router.waiting_messages = {
             'kun': EMQdeque(initial=[waiting_msg1, waiting_msg2]),
-            'blu': EMQdeque(initial=[waiting_msg3])
+            'blu': EMQdeque(initial=[waiting_msg3, ]),
         }
 
+        # Forward waiting_msg1
         ready_msgid1 = 'ready23'
         self.router.on_ready(worker1_id, ready_msgid1, ['READY', ready_msgid1])
         fwdmsg_mock.assert_called_with(self.router.outgoing, worker1_id,
                                        waiting_msg1)
 
-        ready_msgid2 = 'ready19'
-        self.router.on_ready(worker2_id, ready_msgid2, ['READY', ready_msgid2])
+        # Forward waiting_msg3 -- blu is a higher priority for worker2
+        ready_msgid3 = 'ready19'
+        self.router.on_ready(worker2_id, ready_msgid3, ['READY', ready_msgid3])
         fwdmsg_mock.assert_called_with(self.router.outgoing, worker2_id,
                                        waiting_msg3)
 
-        ready_msgid3 = 'ready5'
-        self.router.on_ready(worker2_id, ready_msgid3, ['READY', ready_msgid3])
+        # Forward waiting_msg2
+        ready_msgid2 = 'ready5'
+        self.router.on_ready(worker2_id, ready_msgid2, ['READY', ready_msgid2])
         fwdmsg_mock.assert_called_with(self.router.outgoing, worker2_id,
                                        waiting_msg2)
 
+        # There should be no keys because the code checks for their existence
+        # to know if there is a waiting message
+        self.assertEqual(0, len(self.router.waiting_messages.keys()))
+
+        # No waiting messages
         self.router.on_ready(worker1_id, ready_msgid1, ['READY', ready_msgid1])
         requeue_worker_mock.assert_called_with(worker1_id)
         self.router.on_ready(worker2_id, ready_msgid2, ['READY', ready_msgid2])
         requeue_worker_mock.assert_called_with(worker2_id)
 
+    @mock.patch('eventmq.router.Router.clean_up_dead_workers')
     @mock.patch('eventmq.router.Router.process_client_message')
     @mock.patch('eventmq.router.Router.get_available_worker')
     @mock.patch('eventmq.router.fwdmsg')
     def test_on_request(self, fwdmsg_mock, get_worker_mock,
-                        process_client_msg_mock):
+                        process_client_msg_mock, cleanupworkers_mock):
         client_id = 'c1'
         msgid = 'msg18'
         queue = 'default'
@@ -245,13 +316,13 @@ class TestCase(unittest.TestCase):
 
         self.router.workers = {
             worker_id: {
-                'queues': EMQdeque(initial=(queue,)),
+                'queues': [(10, queue)],
                 'hb': 2903.34,
                 'available_slots': 1,
             }
         }
         self.router.queues = {
-            'default': EMQdeque(initial=((10, worker_id)))
+            queue: [(10, worker_id),]
         }
 
         # Router accepts job for 1 available slot
@@ -268,7 +339,7 @@ class TestCase(unittest.TestCase):
         get_worker_mock.side_effect = raise_no_workers
         self.router.on_request(client_id, msgid, msg)
 
-        self.assertIn(msg[0], self.router.waiting_messages)
+        self.assertIn(queue, self.router.waiting_messages)
         self.assertEqual(list(self.router.waiting_messages[queue])[0],
                          ['', constants.PROTOCOL_VERSION, 'REQUEST',
                           msgid] + msg)
@@ -286,14 +357,85 @@ class TestCase(unittest.TestCase):
             [client_id, '', constants.PROTOCOL_VERSION, 'REQUEST', msgid]+msg,
             depth=2)
 
-    def test_cleanup_dead_workers(self):
+    def test_get_available_worker(self):
+        worker2_id = 'w2'
+        worker3_id = 'w3'
+
+        queue1_id = 'default'
+        queue2_id = 'jimjam'
+
+        self.router.queues = {
+            queue1_id: [(10, worker3_id), (0, worker2_id)],
+            queue2_id: [(10, worker2_id)],
+        }
+
+        self.router.workers = {
+            worker2_id: {
+                'queues': [(10, queue2_id), (0, queue1_id)],
+                'available_slots': 1,
+            },
+            worker3_id: {
+                'queues': [(10, queue1_id), ],
+                'available_slots': 1,
+            },
+        }
+
+        # Get the next available worker for queue2
+        check1 = self.router.get_available_worker(queue_name=queue2_id)
+        self.assertEqual(worker2_id, check1)
+
+        # Get the next available worker for queue1
+        check2 = self.router.get_available_worker(queue_name=queue1_id)
+        self.assertEqual(worker3_id, check2)
+
+        # Pretend worker 3 is doing something
+        self.router.workers[worker3_id]['available_slots'] = 0
+
+        # Get the next available worker for queue1
+        check3 = self.router.get_available_worker(queue_name=queue1_id)
+        self.assertEqual(worker2_id, check3)
+
+    def test_get_available_worker_dont_decrement_slots(self):
+        # Once upon a time get_available_worker() decremented the available
+        # slots counter and the townsfolk greived
+        queue1_id = 'q1'
+        worker1_id = 'w1'
+
+        self.router.queues = {
+            queue1_id: [(10, worker1_id, ), ]
+        }
+
+        self.router.workers = {
+            worker1_id: {
+                'queues': [(10, queue1_id), ],
+                'available_slots': 1,
+            }
+        }
+
+        self.router.get_available_worker(queue_name=queue1_id)
+
+        self.assertEqual(self.router.workers[worker1_id]['available_slots'], 1)
+
+    def test_requeue_worker(self):
+        worker_id = 'w1'
+
+        self.router.workers = {
+            worker_id: {
+                'available_slots': 1
+            }
+        }
+
+        self.router.requeue_worker(worker_id)
+        self.assertEqual(self.router.workers[worker_id]['available_slots'], 2)
+
+    def test_clean_up_dead_workers(self):
         worker1_id = 'w1'
         worker2_id = 'w2'
         worker3_id = 'w3'
 
         queue1_id = 'default'
         queue2_id = 'jimjam'
-        nonexistent_queue1 = 'pig'
+        nonexistent_queue1 = 'nonexistent'
 
         # To ensure the value was changed later because monotonic() is hard to
         # mock
@@ -307,21 +449,21 @@ class TestCase(unittest.TestCase):
         }
 
         self.router.workers = {
-            # 1 second away from timeout
+            # 3 in the future
             worker1_id: {
-                'queues': (queue1_id,),
-                'hb': monotonic() - conf.HEARTBEAT_TIMEOUT + 1,
+                'queues': [(10, queue1_id), ],
+                'hb': monotonic() + 3,
                 'available_slots': 0,
             },
             # below the timeout
             worker2_id: {
-                'queues': (queue2_id, queue1_id),
+                'queues': [(10, queue2_id), (0, queue1_id)],
                 'hb': 0,
                 'available_slots': 2,
             },
             # below the timeout and a queue missing from self.router.queues
             worker3_id: {
-                'queues': (queue2_id, nonexistent_queue1),
+                'queues': [(10, queue2_id), (3, nonexistent_queue1)],
                 'hb': 0,
                 'available_slots': 0,
             },
@@ -337,74 +479,6 @@ class TestCase(unittest.TestCase):
         self.assertNotIn(queue2_id, self.router.queues)
         self.assertNotIn(nonexistent_queue1, self.router.queues)
 
-    # @mock.patch('eventmq.router.Router.prioritize_queue_list')
-    def test_add_worker(self):
-        worker1_id = 'w1'
-        queues = ('top', 'drop', 'shop')
-
-        self.router.add_worker(worker1_id, queues=queues)
-        # added to the list of workers
-        self.assertIn(worker1_id, self.router.workers)
-        # got an inital heartbeat
-        self.assertGreater(self.router.workers[worker1_id]['hb'], 0)
-        # no slots yet
-        self.assertEqual(self.router.workers[worker1_id]['available_slots'], 0)
-        # aware of the queues
-        self.assertEqual(3, len(self.router.workers[worker1_id]['queues']))
-        self.assertIn((10, worker1_id), list(self.router.queues['top']))
-        self.assertIn((0, worker1_id), list(self.router.queues['drop']))
-        self.assertIn((0, worker1_id), list(self.router.queues['shop']))
-
-    def test_get_available_worker(self):
-        worker2_id = 'w2'
-        worker3_id = 'w3'
-
-        queue1_id = 'default'
-        queue2_id = 'jimjam'
-
-        self.router.queues = {
-            queue1_id: EMQdeque(initial=[(10, worker3_id), (0, worker2_id)]),
-            queue2_id: EMQdeque(initial=[(10, worker2_id)]),
-        }
-
-        self.router.workers = {
-            worker2_id: {
-                'queues': (queue2_id, queue1_id),
-                'available_slots': 1,
-            },
-            worker3_id: {
-                'queues': (queue1_id,),
-                'available_slots': 1,
-            },
-        }
-
-        # worker1 has no available slots.
-        check1 = self.router.get_available_worker(queue_name=queue2_id)
-        self.assertEqual(worker2_id, check1)
-        self.assertEqual(self.router.workers[worker2_id]['available_slots'], 1)
-
-        check2 = self.router.get_available_worker(queue_name=queue1_id)
-        self.assertEqual(worker3_id, check2)
-        self.assertEqual(self.router.workers[worker3_id]['available_slots'], 1)
-
-        self.router.workers[worker3_id]['available_slots'] = 0
-
-        check3 = self.router.get_available_worker(queue_name=queue1_id)
-        self.assertEqual(worker2_id, check3)
-        self.assertEqual(self.router.workers[worker2_id]['available_slots'], 1)
-
-    def test_requeue_worker(self):
-        worker_id = 'w1'
-
-        self.router.workers = {
-            worker_id: {
-                'available_slots': 1
-            }
-        }
-
-        self.router.requeue_worker(worker_id)
-        self.assertEqual(self.router.workers[worker_id]['available_slots'], 2)
-
     @mock.patch('eventmq.router.Router.on_inform')
     @mock.patch('eventmq.router.Router.on_request')
     @mock.patch('eventmq.router.parse_router_message')
@@ -457,37 +531,35 @@ class TestCase(unittest.TestCase):
         on_inform_mock.assert_called_with(sender_id, msgid, msg)
 
     def test_prioritize_queue_list(self):
-        queue = EMQdeque(initial=[(0, 'd'), (10, 'b'), (0, 'e'), (10, 'a'),
-                                  (0, 'c')])
+        queue = [(0, 'd'), (10, 'b'), (0, 'e'), (10, 'a'), (0, 'c')]
 
         sorted1 = self.router.prioritize_queue_list(queue)
         self.assertEqual([(10, 'b'), (10, 'a'), (0, 'd'), (0, 'e'),
-                          (0, 'c')], list(sorted1))
+                          (0, 'c')], sorted1)
 
-        pop1 = sorted1.popleft()
+        pop1 = sorted1.pop(0)
         self.assertEqual(pop1, (10, 'b'))
         sorted1.append(pop1)
         # a, b, d, e, c
         sorted2 = self.router.prioritize_queue_list(sorted1)
         self.assertEqual([(10, 'a'), (10, 'b'), (0, 'd'), (0, 'e'), (0, 'c')],
-                         list(sorted2))
-        pop2 = sorted2.popleft()  # a
-        pop3 = sorted2.popleft()  # b
-        pop4 = sorted2.popleft()  # d
+                         sorted2)
+        pop2 = sorted2.pop(0)  # a
+        pop3 = sorted2.pop(0)  # b
+        pop4 = sorted2.pop(0)  # d
         self.assertEqual(pop4, (0, 'd'))
         self.assertEqual(pop2, (10, 'a'))
         self.assertEqual(pop3, (10, 'b'))
         self.assertEqual([(0, 'e'), (0, 'c')], list(sorted2))
 
-        sorted2.appendleft(pop2)
-        sorted2.appendleft(pop4)
-        sorted2.appendleft(pop3)
+        sorted2.append(pop2)
+        sorted2.append(pop4)
+        sorted2.append(pop3)
 
-        # a, b, d, e, c
         sorted3 = self.router.prioritize_queue_list(sorted2)
 
-        self.assertEqual(sorted3.popleft(), (10, 'b'))
-        self.assertEqual(sorted3.popleft(), (10, 'a'))
-        self.assertEqual(sorted3.popleft(), (0, 'd'))
-        self.assertEqual(sorted3.popleft(), (0, 'e'))
-        self.assertEqual(sorted3.popleft(), (0, 'c'))
+        self.assertEqual(sorted3.pop(0), (10, 'a'))
+        self.assertEqual(sorted3.pop(0), (10, 'b'))
+        self.assertEqual(sorted3.pop(0), (0, 'e'))
+        self.assertEqual(sorted3.pop(0), (0, 'c'))
+        self.assertEqual(sorted3.pop(0), (0, 'd'))
diff --git a/eventmq/tests/test_utils.py b/eventmq/tests/test_utils.py
index ff3c238..f4669ea 100644
--- a/eventmq/tests/test_utils.py
+++ b/eventmq/tests/test_utils.py
@@ -18,6 +18,7 @@ import unittest
 
 import mock
 
+from .. import constants
 from .. import exceptions
 from ..utils import messages
 from ..utils import classes
@@ -95,6 +96,75 @@ class SettingsTestCase(unittest.TestCase):
         self.assertEqual(conf.WORKER_ADDR, 'tcp://160.254.23.88:47290')
 
 
+class EMQPServiceTestCase(unittest.TestCase):
+
+    # pretend to be an emq socket
+    outgoing = 'some-outgoing-socket'
+
+    def get_worker(self):
+        """return an EMQPService mimicking a worker"""
+        obj = classes.EMQPService()
+        obj.SERVICE_TYPE = constants.CLIENT_TYPE.worker
+        obj.outgoing = self.outgoing
+        obj._meta = {
+            'last_sent_heartbeat': 0
+        }
+
+        return obj
+
+    @mock.patch('eventmq.utils.classes.sendmsg')
+    def test_send_inform_return_msgid(self, sendmsg_mock):
+        obj = self.get_worker()
+        sendmsg_mock.return_value = 'some-msgid'
+
+        retval = obj.send_inform(queues=[(10, 'default'), ])
+
+        self.assertEqual(retval, sendmsg_mock.return_value)
+
+    @mock.patch('eventmq.utils.classes.sendmsg')
+    def test_send_inform_single_weightless_queue(self, sendmsg_mock):
+        # Test that the inform message is backward compatible with a change
+        # in v0.2.0
+        obj = self.get_worker()
+
+        obj.send_inform(queues='derpfault')
+
+        sendmsg_mock.assert_called_with(
+            'some-outgoing-socket', 'INFORM',
+            ['derpfault', constants.CLIENT_TYPE.worker]
+        )
+
+    @mock.patch('eventmq.utils.classes.sendmsg')
+    def test_send_inform_empty_queue_name(self, sendmsg_mock):
+        obj = self.get_worker()
+
+        obj.send_inform()
+
+        sendmsg_mock.assert_called_with(
+            'some-outgoing-socket', 'INFORM',
+            ['', constants.CLIENT_TYPE.worker])
+
+    @mock.patch('eventmq.utils.classes.sendmsg')
+    def test_send_inform_specified_valid_queues(self, sendmsg_mock):
+        obj = self.get_worker()
+
+        obj.send_inform(queues=([10, 'push'], [7, 'email'],
+                                [3, 'default']))
+        sendmsg_mock.asert_called_with(
+            'some-outgoing-socket', 'INFORM',
+            ["[10, 'push'],[7, 'email'],[3, 'default]",
+             constants.CLIENT_TYPE.worker]
+        )
+
+    @mock.patch('eventmq.utils.classes.sendmsg')
+    def test_send_inform_update_last_sent_heartbeat(self, sendmsg_mock):
+        obj = self.get_worker()
+
+        obj.send_inform(queues=(['', constants.CLIENT_TYPE.worker]))
+
+        self.assertGreater(obj._meta['last_sent_heartbeat'], 0)
+
+
 class TestCase(unittest.TestCase):
     def test_generate_msgid(self):
         msgid = messages.generate_msgid()
diff --git a/eventmq/utils/__init__.py b/eventmq/utils/__init__.py
index 5973c93..2a047a5 100644
--- a/eventmq/utils/__init__.py
+++ b/eventmq/utils/__init__.py
@@ -46,3 +46,15 @@ def zero_index_cmp(a, b):
     when sorting the values in :attr:`router.Router.queues`.
     """
     return cmp(a[0], b[0])
+
+
+def tuplify(v):
+    """
+    Recursively convert lists to tuples.
+
+    Args:
+        v (object): any value of interest
+    """
+    if isinstance(v, list):
+        return tuple(map(tuplify, v))
+    return v
diff --git a/eventmq/utils/classes.py b/eventmq/utils/classes.py
index a0f9740..0e54f43 100644
--- a/eventmq/utils/classes.py
+++ b/eventmq/utils/classes.py
@@ -53,14 +53,20 @@ class EMQPService(object):
 
     See the code for :class:`Scheduler` and :class:`JobManager` for examples.
     """
-    def send_inform(self, queue=None):
+    def send_inform(self, queues=()):
         """
-        Queues an INFORM command to `self.outgoing`.
+        Notify the router that this job manager is online and and ready for
+        work. This includes a list of queues the router should forward messages
+        for.
 
         Args:
             type_ (str): Either 'worker' or 'scheduler'
-            queue (list):
-                - For 'worker' type, the queues the worker is listening on
+            queues (list):
+                - For 'worker' type, the queues the worker is listening on and
+                  their weights.
+
+                  Example:
+                    ([10, 'default'], [15, 'push_notifications'])
                 - Ignored for 'scheduler' type
 
         Raises:
@@ -68,6 +74,11 @@ class EMQPService(object):
 
         Returns:
             str: ID of the message
+
+        .. note::
+
+           Passing a single string for queues is supported for backward
+           compatibility and not recommended for new apps.
         """
         valid_types = (constants.CLIENT_TYPE.worker,
                        constants.CLIENT_TYPE.scheduler)
@@ -75,8 +86,15 @@ class EMQPService(object):
         if self.SERVICE_TYPE not in valid_types:
             raise ValueError('{} not one of {}'.format(self.SERVICE_TYPE,
                                                        valid_types))
+
+        if isinstance(queues, (list, tuple)):
+            stringified_queues = ''
+            for pair in queues:
+                stringified_queues += '{},'.format(str(pair))
+            queues = stringified_queues[:-1]  # strip off the last comma
+
         msgid = sendmsg(self.outgoing, 'INFORM', [
-            queue or conf.DEFAULT_QUEUE_NAME,
+            queues,
             self.SERVICE_TYPE
         ])
 
diff --git a/eventmq/utils/settings.py b/eventmq/utils/settings.py
index 0621aaf..1c2671e 100644
--- a/eventmq/utils/settings.py
+++ b/eventmq/utils/settings.py
@@ -21,6 +21,7 @@ import json
 import logging
 import os
 
+from . import tuplify
 from .. import conf
 
 
@@ -71,15 +72,3 @@ def import_settings(section='global'):
     else:
         logger.warning('Config file at {} not found. Continuing with '
                        'defaults.'.format(conf.CONFIG_FILE))
-
-
-def tuplify(v):
-    """
-    Recursively convert lists to tuples.
-
-    Args:
-        v (object): any value of interest
-    """
-    if isinstance(v, list):
-        return tuple(map(tuplify, v))
-    return v
diff --git a/eventmq/worker.py b/eventmq/worker.py
index 0e931ff..23c9dfd 100644
--- a/eventmq/worker.py
+++ b/eventmq/worker.py
@@ -20,7 +20,7 @@ Defines different short-lived workers that execute jobs
 from importlib import import_module
 import logging
 
-logger = logging.getLogger('*')
+logger = logging.getLogger(__name__)
 
 
 def run(payload):