pre-commit and remove RmqCoordinator to tests/util only

src/plumpy/__init__.py

@@ -18,9 +18,9 @@
 from .process_listener import *
 from .process_states import *
 from .processes import *
+from .rmq import *
 from .utils import *
 from .workchains import *
-from .rmq import *
 
 __all__ = (
     events.__all__

src/plumpy/futures.py

@@ -88,6 +88,4 @@ def create_task(coro: Callable[[], Awaitable[Any]], loop: Optional[asyncio.Abstr
     # asyncio.run_coroutine_threadsafe(run_task(), loop)
     # return future
 
-    return asyncio.wrap_future(
-        asyncio.run_coroutine_threadsafe(coro(), loop)
-    )
+    return asyncio.wrap_future(asyncio.run_coroutine_threadsafe(coro(), loop))

src/plumpy/message.py

@@ -10,8 +10,6 @@
 from plumpy.coordinator import Coordinator
 from plumpy.exceptions import PersistenceError, TaskRejectedError
 
-from plumpy.exceptions import PersistenceError, TaskRejectedError
-
 from . import loaders, persistence
 from .utils import PID_TYPE
 

src/plumpy/processes.py

@@ -1,6 +1,8 @@
 # -*- coding: utf-8 -*-
 """The main Process module"""
 
+from __future__ import annotations
+
 import abc
 import asyncio
 import concurrent.futures

src/plumpy/rmq/communications.py

@@ -78,11 +78,11 @@ def converted(communicator: kiwipy.Communicator, *args: Any, **kwargs: Any) -> k
 
     return converted
 
+
 T = TypeVar('T', bound=kiwipy.Communicator)
 
-def wrap_communicator(
-    communicator: T, loop: Optional[asyncio.AbstractEventLoop] = None
-) -> 'LoopCommunicator[T]':
+
+def wrap_communicator(communicator: T, loop: Optional[asyncio.AbstractEventLoop] = None) -> 'LoopCommunicator[T]':
     """
     Wrap a communicator such that all callbacks made to any subscribers are scheduled on the
     given event loop.

src/plumpy/rmq/futures.py

@@ -10,7 +10,7 @@
 
 import kiwipy
 
-__all__ = ['wrap_to_concurrent_future', 'unwrap_kiwi_future']
+__all__ = ['unwrap_kiwi_future', 'wrap_to_concurrent_future']
 
 
 def _convert_future_exc(exc):
@@ -112,6 +112,7 @@ def wrap_to_concurrent_future(future: asyncio.Future[Any]) -> kiwipy.Future:
     _chain_future(future, new_future)
     return new_future
 
+
 # XXX: this required in aiida-core, see if really need this unwrap.
 def unwrap_kiwi_future(future: kiwipy.Future) -> kiwipy.Future:
     """

tests/rmq/__init__.py

@@ -1,14 +1,23 @@
 # -*- coding: utf-8 -*-
+from typing import Generic, TypeVar, final
 import kiwipy
 import concurrent.futures
 
 from plumpy.exceptions import CoordinatorConnectionError
 
 
-class RmqCoordinator:
-    def __init__(self, comm: kiwipy.Communicator):
+U = TypeVar('U', bound=kiwipy.Communicator)
+
+@final
+class RmqCoordinator(Generic[U]):
+    def __init__(self, comm: U):
         self._comm = comm
 
+    @property
+    def communicator(self) -> U:
+        """The inner communicator."""
+        return self._comm
+
     # XXX: naming - `add_receiver_rpc`
     def add_rpc_subscriber(self, subscriber, identifier=None):
         return self._comm.add_rpc_subscriber(subscriber, identifier)
@@ -18,9 +27,19 @@ def add_broadcast_subscriber(
         self,
         subscriber,
         subject_filters=None,
+        sender_filters=None,
         identifier=None,
     ):
-        subscriber = kiwipy.BroadcastFilter(subscriber, subject=subject_filters)
+        subscriber = kiwipy.BroadcastFilter(subscriber)
+
+        subject_filters = subject_filters or []
+        sender_filters = sender_filters or []
+
+        for filter in subject_filters:
+            subscriber.add_subject_filter(filter)
+        for filter in sender_filters:
+            subscriber.add_sender_filter(filter)
+
         return self._comm.add_broadcast_subscriber(subscriber, identifier)
 
     # XXX: naming - `add_reciver_task` (can be combined with two above maybe??)