Merge pull request #3846 from dannon/mq_routing_fix

[17.01] Fix for direct message routing using a 'real' exchange like rabbitmq

lib/galaxy/queue_worker.py

@@ -24,7 +24,7 @@ def send_local_control_task(app, task, kwargs={}):
     This sends a message to the process-local control worker, which is useful
     for one-time asynchronous tasks like recalculating user disk usage.
     """
-    log.info("Queuing async task %s." % task)
+    log.info("Queuing async task %s for %s." % (task, app.config.server_name))
     payload = {'task': task,
                'kwargs': kwargs}
     try:
@@ -33,7 +33,7 @@ def send_local_control_task(app, task, kwargs={}):
             producer.publish(payload,
                              exchange=galaxy.queues.galaxy_exchange,
                              declare=[galaxy.queues.galaxy_exchange] + [galaxy.queues.control_queue_from_config(app.config)],
-                             routing_key='control')
+                             routing_key='control.%s' % app.config.server_name)
     except Exception:
         log.exception("Error queueing async task: %s." % payload)
 

lib/galaxy/queues.py

@@ -31,7 +31,9 @@ def control_queue_from_config(config):
     Returns a Queue instance with the correct name and routing key for this
     galaxy process's config
     """
-    return Queue("control.%s" % config.server_name, galaxy_exchange, routing_key='control')
+    return Queue("control.%s" % config.server_name,
+                 galaxy_exchange,
+                 routing_key='control.%s' % config.server_name)
 
 
 def connection_from_config(config):