Support for publishing messages with a routing key

severb · Oct 12, 2016 · 0838d7d · 0838d7d
1 parent a9408bb
commit 0838d7d
Showing 1 changed file with 7 additions and 7 deletions.
diff --git a/graypy/rabbitmq.py b/graypy/rabbitmq.py
@@ -35,8 +35,8 @@ class GELFRabbitHandler(SocketHandler):
     """
 
     def __init__(self, url, exchange='logging.gelf', debugging_fields=True,
-            extra_fields=True, fqdn=False, exchange_type='fanout', localname=None,
-            facility=None, virtual_host='/'):
+                 extra_fields=True, fqdn=False, exchange_type='fanout', localname=None,
+                 facility=None, virtual_host='/', routing_key=''):
         self.url = url
         parsed = urlparse(url)
         if parsed.scheme != 'amqp':
@@ -59,12 +59,13 @@ def __init__(self, url, exchange='logging.gelf', debugging_fields=True,
         self.localname = localname
         self.facility = facility
         self.virtual_host = virtual_host
+        self.routing_key = routing_key
         SocketHandler.__init__(self, host, port)
         self.addFilter(ExcludeFilter('amqplib'))
 
     def makeSocket(self, timeout=1):
         return RabbitSocket(self.cn_args, timeout, self.exchange,
-            self.exchange_type)
+                            self.exchange_type, self.routing_key)
 
     def makePickle(self, record):
         message_dict = make_message_dict(
@@ -74,12 +75,12 @@ def makePickle(self, record):
 
 
 class RabbitSocket(object):
-
-    def __init__(self, cn_args, timeout, exchange, exchange_type):
+    def __init__(self, cn_args, timeout, exchange, exchange_type, routing_key):
         self.cn_args = cn_args
         self.timeout = timeout
         self.exchange = exchange
         self.exchange_type = exchange_type
+        self.routing_key = routing_key
         self.connection = amqp.Connection(
             connection_timeout=timeout, **self.cn_args)
         self.channel = self.connection.channel()
@@ -92,7 +93,7 @@ def __init__(self, cn_args, timeout, exchange, exchange_type):
 
     def sendall(self, data):
         msg = amqp.Message(data, delivery_mode=2)
-        self.channel.basic_publish(msg, exchange=self.exchange)
+        self.channel.basic_publish(msg, exchange=self.exchange, routing_key=self.routing_key)
 
     def close(self):
         try:
@@ -102,7 +103,6 @@ def close(self):
 
 
 class ExcludeFilter(Filter):
-
     def __init__(self, name):
         """Initialize filter.