[BLOCKED] Server properties

kiwipy/communications.py

@@ -120,7 +120,7 @@ def broadcast_send(self, body, sender=None, subject=None, correlation_id=None) -
 
 class CommunicatorHelper(Communicator):
     # Have to disable this linter because this class remains abstract and it is
-    # just used by calsses that will themselves be concrete
+    # just used by classes that will themselves be concrete
     # pylint: disable=abstract-method
 
     def __init__(self):

kiwipy/rmq/communicator.py

@@ -4,7 +4,7 @@
 import copy
 import logging
 import typing
-from typing import Union, Optional
+from typing import Union, Optional, Dict
 
 import shortuuid
 import aio_pika
@@ -345,6 +345,13 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
     def __str__(self):
         return f'RMQCommunicator({self._connection})'
 
+    @property
+    def server_properties(self) -> Dict:
+        if self._connection is None:
+            return {}
+
+        return self._connection.connection.server_properties
+
     @property
     def loop(self):
         """Get the event loop instance driving this communicator connection."""

kiwipy/rmq/threadcomms.py

@@ -5,7 +5,7 @@
 from concurrent.futures import Future as ThreadFuture
 import functools
 import logging
-from typing import Union
+from typing import Union, Dict
 
 import aio_pika
 import deprecation
@@ -148,6 +148,10 @@ def start(self):
     def stop(self):
         self.close()
 
+    @property
+    def server_properties(self) -> Dict:
+        return self._communicator.server_properties
+
     def __enter__(self):
         return self
 

test/rmq/test_coroutine_communicator.py

@@ -197,3 +197,12 @@ def broadcast_subscriber(_comm, _body, _sender=None, _subject=None, _correlation
 
 
 # endregion
+
+
+@pytest.mark.asyncio
+async def test_server_properties(communicator: kiwipy.rmq.RmqCommunicator):
+    props = communicator.server_properties
+    assert isinstance(props, dict)
+
+    assert props['product'] == b'RabbitMQ'
+    assert props['platform'].startswith(b'Erlang')