feat: Better example code

Author: Alan Christie

README.md

@@ -78,6 +78,24 @@ To list (**GET**) all the existing event streams, run the following:
 
     http localhost:8081/event-stream/ -b
 
+The docker-compose file will start the web socket service on port 8080 and
+the internal API on port 8081. ThIt will also run a RabbitMQ server, which can be found
+on port 5672.
+
+As well as creating and deleting sockets via the internal API some very simple Python
+modules have also been provided to inject messages onto the RabbitMQ bus and to
+read messages from the corresponding web socket: -
+
+If you have a websocket you can start a simple listener with the following command,
+which will print each message received: -
+
+    ./simple_es_subscriber.py <location>
+
+You can then 'inject' a very simple **MerchantCharge** message that will be picked up
+by the client using the command: -
+
+    ./simple_es_publisher.py <routing_key>
+
 ---
 
 [black]: https://black.readthedocs.io/en/stable

poetry.lock

@@ -24,6 +24,7 @@ httpie = "^3.2.4"
 isort = "^6.0.0"
 black = "^25.1.0"
 mypy = "^1.15.0"
+im-protobuf = "^7.1.0"
 
 [build-system]
 requires = ["poetry-core"]

pyproject.toml

@@ -0,0 +1,55 @@
+#!/usr/bin/env python
+"""A simple RabbitMQ message publisher for local testing.
+Takes a 'routing key' and sends a message to the 'expected' AS exchange
+on a localhost RabbitMQ server.
+
+Usage: simple_es_publisher.py <routing_key>
+"""
+import asyncio
+import sys
+from datetime import datetime, timezone
+
+import aio_pika
+from google.protobuf import text_format
+from informaticsmatters.protobuf.accountserver.merchant_charge_message_pb2 import (
+    MerchantChargeMessage,
+    OperationEnum,
+)
+
+_ROUTING_KEY: str = sys.argv[1]
+
+_AMPQ_EXCHANGE: str = "event-streams"
+_AMPQ_URL: str = "amqp://es:cheddar1963@localhost:5672"
+
+# A demonstration message
+_MESSAGE_CLASS: str = "accountserver.MerchantCharge"
+_MESSAGE: MerchantChargeMessage = MerchantChargeMessage()
+_MESSAGE.timestamp = datetime.now(timezone.utc).isoformat(timespec="milliseconds")
+_MESSAGE.merchant_kind = "DATA_MANAGER"
+_MESSAGE.merchant_name = "squonk"
+_MESSAGE.merchant_id = 1
+_MESSAGE.operation = OperationEnum.OPERATION_ENUM_PROCESSING
+_MESSAGE.auth_code = 456782
+_MESSAGE.value = "0.50"
+_MESSAGE.sqn = 1
+_MESSAGE_STRING: str = text_format.MessageToString(_MESSAGE, as_one_line=True)
+
+
+async def main() -> None:
+    """Publish a message."""
+    connection = await aio_pika.connect_robust(_AMPQ_URL)
+    async with connection:
+        channel = await connection.channel()
+
+        es_exchange = await channel.declare_exchange(
+            _AMPQ_EXCHANGE,
+            aio_pika.ExchangeType.DIRECT,
+        )
+        await es_exchange.publish(
+            aio_pika.Message(body=f"{_MESSAGE_CLASS}|{_MESSAGE_STRING}".encode()),
+            routing_key=_ROUTING_KEY,
+        )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

simple_es_publisher.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 """A simple WebSocket client for local testing.
 
-Usage: simple_ws_client.py <location>
+Usage: simple_es_subscriber.py <location>
 """
 import asyncio
 import sys
@@ -17,7 +17,7 @@ async def main():
     try:
         while True:
             data = await ws.receive()
-            print(f"< {data}")
+            print(str(data))
     except (KeyboardInterrupt, EOFError, ConnectionClosed):
         await ws.close()