hive.messaging: Factor Notifier out of hive-tell-user

libs/messaging/hive/messaging/__init__.py

@@ -1,10 +1,11 @@
-from .channel import Channel
+from .channel import Channel, Notifier
 from .connection import Connection
 from .message_bus import MessageBus, Producer
 
 DEFAULT_MESSAGE_BUS = MessageBus()
 
 blocking_connection = DEFAULT_MESSAGE_BUS.blocking_connection
 send_to_queue = DEFAULT_MESSAGE_BUS.send_to_queue
+tell_user = DEFAULT_MESSAGE_BUS.tell_user
 
 producer_connection = DEFAULT_MESSAGE_BUS.producer_connection

libs/messaging/hive/messaging/channel.py

@@ -1,13 +1,23 @@
 import json
 
-from typing import Optional
+from functools import cached_property
+from typing import Callable, Optional
 
 from pika import BasicProperties, DeliveryMode
 
 from .wrapper import WrappedPikaThing
+from .channel_services import Notifier
 
 
 class Channel(WrappedPikaThing):
+    @cached_property
+    def notifier(self) -> Notifier:
+        return Notifier(self)
+
+    @cached_property
+    def tell_user(self) -> Callable:
+        return self.notifier.tell_user
+
     def send_to_queue(
             self,
             queue: str,

libs/messaging/hive/messaging/channel_services/__init__.py

@@ -0,0 +1 @@
+from .notifier import Notifier

libs/messaging/hive/messaging/channel_services/channel_service.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+import weakref
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from ..channel import Channel
+
+
+class ChannelService:
+    def __init__(self, channel: Channel):
+        self.channel = weakref.proxy(channel)
+
+    def _declared_queue(self, queue, **kwargs) -> str:
+        """Declare a queue, returning its name.
+
+        Intended to be wrapped in a cached_property.
+        """
+        self.channel.queue_declare(queue, **kwargs)
+        return queue

libs/messaging/hive/messaging/channel_services/notifier.py

@@ -0,0 +1,18 @@
+from functools import cached_property
+
+from .channel_service import ChannelService
+
+
+class Notifier(ChannelService):
+    @cached_property
+    def outbound_queue(self) -> str:
+        return self._declared_queue(
+            "matrix.messages.outgoing",
+            durable=True,
+        )
+
+    def tell_user(self, message: str, format: str = "text"):
+        self.channel.send_to_queue(self.outbound_queue, {
+            "format": format,
+            "messages": [message],
+        })

libs/messaging/hive/messaging/commands/tell_user.py

@@ -1,6 +1,6 @@
 from hive.common import ArgumentParser
 
-from .. import send_to_queue
+from .. import tell_user
 
 
 class TellUserArgumentParser(ArgumentParser):
@@ -28,7 +28,4 @@ def main():
         parser.add_format_argument(format)
     args = parser.parse_args()
 
-    send_to_queue("matrix.messages.outgoing", {
-        "format": args.format,
-        "messages": [args.message],
-    })
+    tell_user(args.message, format=args.format)

libs/messaging/hive/messaging/message_bus.py

@@ -91,6 +91,11 @@ def send_to_queue(self, queue: str, *args, **kwargs):
             )
             return channel.send_to_queue(queue, *args, **kwargs)
 
+    def tell_user(self, *args, **kwargs):
+        with self.blocking_connection(connection_attempts=1) as conn:
+            channel = conn.channel()
+            return channel.tell_user(*args, **kwargs)
+
     # Streams
 
     def stream_connection_parameters(self, *, port: int = 5552, **kwargs):