Initial commit

Author: rockem

.gitignore

@@ -0,0 +1,3 @@
+venv
+.pytest_cache
+.idea

docker-compose.yml

@@ -0,0 +1,8 @@
+version: '3'
+
+services:
+  rabbit:
+    image: rabbitmq:3-management-alpine
+    ports:
+      - 8888:15672
+      - 5672:5672

eggplant/__init__.py

@@ -0,0 +1,19 @@
+class Eggplant:
+    def __init__(self, broker):
+        self._broker = broker
+        self._handlers = {}
+
+    def handler(self, message_name):
+        def decorator_handler(func):
+            self._handlers.update({message_name: func})
+
+        return decorator_handler
+
+    def start(self):
+        self._broker.consume(topics=self._handlers.keys(), callback=self._on_message)
+
+    def _on_message(self, message, delivery_info):
+        self._handlers[delivery_info['topic']](message)
+
+    def stop(self):
+        self._broker.stop()

eggplant/core.py

@@ -0,0 +1,45 @@
+from kombu import Connection, Exchange, Queue, binding
+from kombu.mixins import ConsumerMixin
+
+
+class Worker(ConsumerMixin):
+    def __init__(self, connection, queues, callback):
+        self._connection = connection
+        self._queues = queues
+        self._callback = callback
+
+    def get_consumers(self, consumer, channel):
+        return [consumer(queues=self._queues,
+                         callbacks=[self.on_message])]
+
+    def on_message(self, body, message):
+        self._callback(
+            body,
+            {'topic': message.delivery_info['routing_key']})
+        message.ack()
+
+
+class RabbitKombuConsumer:
+
+    def __init__(self, amqp_uri, exchange, queue):
+        self._amqp_uri = amqp_uri
+        self._exchange = exchange
+        self._queue = queue
+        self._worker = None
+
+    def consume(self, topics, callback):
+        queue = self._create_queue_for(topics)
+        with Connection(self._amqp_uri, heartbeat=4) as conn:
+            self._worker = Worker(conn, [queue], callback)
+            self._worker.run()
+
+    def _create_queue_for(self, topics):
+        exchange = Exchange(self._exchange, type="topic")
+        queue = Queue(
+            self._queue,
+            exchange,
+            bindings=[binding(exchange, routing_key=t) for t in topics])
+        return queue
+
+    def stop(self):
+        self._worker.should_stop = True

eggplant/kumbo.py

@@ -0,0 +1,3 @@
+pytest==5.3.5
+pika==1.1.0
+busypie==0.3.0

requirements.txt

@@ -0,0 +1,29 @@
+from threading import Thread
+from time import sleep
+
+from busypie import wait, FIVE_SECONDS
+from kombu import Connection, Exchange, Producer
+
+from eggplant.core import Eggplant
+from eggplant.kumbo import RabbitKombuConsumer
+
+
+def test_consume_using_function_handler():
+    app = Eggplant(RabbitKombuConsumer(amqp_uri='amqp://localhost', exchange='eggplant-exchange', queue='test_queue'))
+    received_messages = []
+
+    @app.handler('status_changed')
+    def status_changed_handler(message):
+        received_messages.append(message)
+
+    Thread(target=app.start).start()
+    sleep(1)
+    _send_message('status_changed', 'enabled')
+    wait().at_most(FIVE_SECONDS).until(lambda: 'enabled' in received_messages)
+    app.stop()
+
+
+def _send_message(topic, message):
+    with Connection('amqp://localhost') as conn:
+        exchange = Exchange("eggplant-exchange", type="topic")
+        Producer(exchange=exchange, channel=conn.channel(), routing_key=topic).publish(message)