Sample pipethrough (#156)

New service to consume a message from a queue, and forward it to
another.

The sample producer and sample consumer services allow a message to be
sent between them. The sample pipethrough service produces a test
service which can be used to both receive and send messages in a recipe.

This is useful to test features that need some basic functionality over
multiple services, such as distributed tracing.

Author: dan-fernandes

pyproject.toml

@@ -39,6 +39,7 @@ workflows = "workflows"
 
 [project.entry-points."workflows.services"]
 SampleConsumer = "workflows.services.sample_consumer:SampleConsumer"
+SamplePipethrough = "workflows.services.sample_pipethrough:SamplePipethrough"
 SampleProducer = "workflows.services.sample_producer:SampleProducer"
 SampleTxn = "workflows.services.sample_transaction:SampleTxn"
 SampleTxnProducer = "workflows.services.sample_transaction:SampleTxnProducer"

src/workflows/services/sample_pipethrough.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import json
+import time
+
+import workflows.recipe
+from workflows.services.common_service import CommonService
+
+
+class SamplePipethrough(CommonService):
+    """An example services building on top of the workflow.services architecture,
+    demonstrating how this architecture can be used.
+    This services consumes messages off a queue, and forwards to another."""
+
+    # Human readable service name
+    _service_name = "Message Pipethrough"
+
+    # Logger name
+    _logger_name = "workflows.service.sample_pipethrough"
+
+    def initializing(self):
+        """Subscribe to a channel."""
+        workflows.recipe.wrap_subscribe(
+            self._transport,
+            "transient.destination",
+            self.process,
+        )
+
+    def process(self, rw, header, message):
+        """Consume message and send to output pipe."""
+        t = (time.time() % 1000) * 1000
+
+        if header:
+            header_str = json.dumps(header, indent=2) + "\n" + "----------------" + "\n"
+        else:
+            header_str = ""
+        if isinstance(message, dict):
+            message = json.dumps(message, indent=2) + "\n" + "----------------" + "\n"
+
+        self.log.info(
+            f"=== Consume ====\n{header_str}{message}\nReceived message @{t:10.3f} ms"
+        )
+
+        rw.send(0)