Actor: Sink can now be: actor, topic or callable/async callable

faust/actors.py

@@ -2,7 +2,7 @@
 import typing
 from collections import defaultdict
 from typing import (
-    Any, AsyncIterable, AsyncIterator, Awaitable,
+    Any, AsyncIterable, AsyncIterator, Awaitable, Callable,
     Iterable, List, MutableMapping, MutableSequence, Union, cast,
 )
 from uuid import uuid4
@@ -319,7 +319,12 @@ async def _slurp(self, res: ActorInstanceT, it: AsyncIterator):
 
     async def _delegate_to_sinks(self, value: Any) -> None:
         for sink in self._sinks:
-            await maybe_async(sink(value))
+            if isinstance(sink, ActorT):
+                await cast(ActorT, sink).send(value=value)
+            elif isinstance(sink, TopicT):
+                await cast(TopicT, sink).send(value=value)
+            else:
+                await maybe_async(cast(Callable, sink)(value))
 
     async def _reply(self, key: Any, value: Any, req: ReqRepRequest) -> None:
         assert req.reply_to

faust/types/actors.py

@@ -22,7 +22,10 @@ class AppT: ...  # noqa
 
 ActorErrorHandler = Callable[['ActorT', Exception], Awaitable]
 ActorFun = Callable[[AsyncIterator], Union[Awaitable, AsyncIterable]]
-SinkT = Callable[[Any], Union[Awaitable, None]]
+
+#: A sink can be: Actor, Topic,
+#: or callable/async callable taking value as argument.
+SinkT = Union['ActorT', TopicT, Callable[[Any], Union[Awaitable, None]]]
 
 
 class ActorT(ServiceT):