Improve tracing output, update for newer runnables features,

This commit is contained in:
Nuno Campos
2023-08-24 17:21:53 +02:00
parent d1a518fdfa
commit 67ee1d87c7
5 changed files with 120 additions and 123 deletions
+2 -2
View File
@@ -94,14 +94,14 @@ class RunnableSubscriber(RunnableBinding[T, Any]):
raise NotImplementedError()
class RunnablePublisher(RunnablePassthrough[T]):
class RunnablePublisher(Runnable[T, T]):
topic: Topic[T]
def invoke(self, input: T, config: Optional[RunnableConfigForPubSub] = None) -> T:
send = config.get(CONFIG_SEND_KEY, None)
if send is not None:
send(self.topic.name, input)
return super().invoke(input, config)
return input
class RunnablePublisherEach(RunnablePublisher[Sequence[T]]):