Add message broker module

MANIFEST.in

@@ -1,10 +1,11 @@
 exclude tests
 recursive-exclude tests *
+exclude examples
+recursive-exclude examples *
 
 exclude .pre-commit-config.yaml
 exclude .gitlab-ci.yml
 
 include LICENSE
 include README.md
-include example.py
 recursive-include logos *.png

examples/rpc_server.py

@@ -0,0 +1,33 @@
+import json
+
+from marketplace_standard_app_api.models.message_broker import (
+    MessageBrokerRequestModel,
+    MessageBrokerResponseModel,
+)
+
+from marketplace.message_broker.rpc_server import RpcServer
+
+
+def my_message_handler(
+    request_message: MessageBrokerRequestModel,
+) -> MessageBrokerResponseModel:
+    print("Routing to endpoint %r..." % request_message.endpoint)
+    payload = json.loads(request_message.body) if request_message.body else {}
+    result = len(payload)
+    response = {"numberOfKeysInPayload": str(result)}
+    print("Done!")
+    response_message = MessageBrokerResponseModel(
+        status_code=200,
+        body=json.dumps(response),
+        headers={"Content-Type": "application/json"},
+    )
+    return response_message
+
+
+rpc_server = RpcServer(
+    host="www.materials-marketplace.eu",
+    application_id="<application-id>",
+    application_secret="<application-secret>",
+    message_handler=my_message_handler,
+)
+rpc_server.consume_messages()

marketplace/message_broker/rpc_server.py

@@ -0,0 +1,43 @@
+import json
+import logging
+
+import pika
+from marketplace_standard_app_api.models.message_broker import MessageBrokerRequestModel
+
+from .utils import calc_queue_name
+
+logger = logging.getLogger(__name__)
+
+
+class RpcServer:
+    def __init__(self, host, application_id, application_secret, message_handler):
+        self.queue_name = calc_queue_name(application_id, application_secret)
+        self.message_handler = message_handler
+        connection = pika.BlockingConnection(pika.ConnectionParameters(host=host))
+        self.channel = connection.channel()
+
+        self.channel.queue_declare(queue=self.queue_name)
+
+    def consume_messages(self):
+        def callback(ch, method, properties, body):
+            request_message = MessageBrokerRequestModel.parse_obj(
+                json.loads(body.decode())
+            )
+
+            response_message = self.message_handler(request_message)
+
+            ch.basic_publish(
+                exchange="",
+                routing_key=properties.reply_to,
+                properties=pika.BasicProperties(
+                    correlation_id=properties.correlation_id,
+                ),
+                body=response_message.json(),
+            )
+            ch.basic_ack(delivery_tag=method.delivery_tag)
+
+        self.channel.basic_qos(prefetch_count=1)
+        self.channel.basic_consume(queue=self.queue_name, on_message_callback=callback)
+
+        logger.info("Waiting for messages. To exit press CTRL+C")
+        self.channel.start_consuming()

marketplace/message_broker/utils.py

@@ -0,0 +1,11 @@
+import hashlib
+
+
+def calc_queue_name(application_id: str, application_secret: str) -> str:
+    h = int(
+        hashlib.sha256(
+            (application_id + application_secret).encode("utf-8")
+        ).hexdigest(),
+        16,
+    ) % (10**16)
+    return f"{application_id}:{h}"

setup.cfg

@@ -22,6 +22,9 @@ classifiers =
 [options]
 packages = find:
 install_requires =
+    marketplace-standard-app-api==0.1.0
+    pika==1.2.1
+    pydantic==1.9.0
     requests~=2.26.0
 python_requires = >=3.8