architecture - Python streamz: integrate stream into stream?

Question

Please note that I am searching for the "most pythonic" way in terms of software architecture / design to tackle this kind of problem. I am aware that streamz itself already induces micro-second delay and that a class-based (or similar pythonic, explicit, clean and easy solution) will probably add even more delay.

---

Let's say we have different streams s1, s2, ... where we'd like to integrate a common method for calculating some metric and using this information afterwards, like:

s1 = streamz.Stream()
s1.map(foo).map(some_metric).sink(print)

s2 = streamz.Stream()
s2.map(bar).map(some_metric).sink(print)

...

We defined some_metric() as a function but some_metric() itself is quite long, needs some branching & joining, ... so there is a strong motivation to represent some_metric() as a Stream itself instead of one (or many) functions.

However, if we would simply approach this by creating just another stream:

some_metric = streamz.Stream()
some_metric_pipeline = some_metric.map(...)...

...we'd ran into the problem that we cannot directly pipe those together: we cannot directly integrate this into s1 and the other streams since some_metric_pipeline awaits some_metric stream to emit data instead of, e.g., s1 and returns the last object of the chain, so, for example, map(). Thus, if we'd simply try to do s1.map(foo).map(some_metric_pipeline).sink(print) we'd receive a map object is not callable exception. Furthermore, even if this would work in theory, some_metric_pipeline's input would have to change from awaiting some_metric's Stream emitting items to the items that s1 stream routes through its pipeline.

Question: Is there a nice, pythonic way to design this kind of "sub-streams" and integrate them into main streams s1, s2, ...? Or would the most pythonic way for abstracting complex calculations to write a class, combine all some_metric functionality there and then simply overwrite its __call__ method so that we could insert an instance of this class into some map() call of the s1, s2, ... streams?

Answer 1

等待大神答复