How to unify sender- and receiver-based coroutines?

Question

As far as I understood the coroutine concept in Python, you can basically have two different modes of passing data (sorry, I couldn't come up or find better terms for these):

1. Sender-based: Each coroutine consumes data from "outside" and sends it to a consumer, e.g.

   def coro(consumer):
       while True:
           item = yield
           consumer.send(process(item))
   

   To build pipelines, one would produce from the outer coroutine to the inner:

   producer(filter(sink()))
   

2. Receiver-based: Each coroutine consumes data from its argument and yields it to a consumer, e.g.

   def coro(producer):
       while True:
           item = next(producer)
           yield process(item)
   

   To build pipelines, one would produce from the inner to the outer coroutine, which in the end looks more what people expect from functions:

   sink(filter(producer()))
   

Both approaches have their own advantages. With sender-based coroutines I can broadcast to many consumers

    def broadcast(consumers):
        while True:
            item = yield
            for consumer in consumers:
                consumer.send(item)

However, sender-based coroutines are always limited to one "input" coroutine because they cannot distinguish who sent them what (well, actually yes, but that would be nasty). This on the other hand is trivial with receiver-based coroutines:

    def adder(producer1, producer2):
        while True:
            x = next(producer1)
            y = next(producer2)
            yield x + y

---

Now my question is: Is there any sane and simple way to unify both approaches? For example broadcasting the result of the adder?

Answer 1

My guess is that it is only possible to do something like this:

def adder(producer1, producer2, consumers):
    while True:
        x = next(producer1)
        y = next(producer2)
        for consumer in consumers:
            consumer.send(x+y)

and then just call adder(x_producer, y_producer, consumers). As you can see from the diagram:

the adder can't be placed in the middle of the pipeline and the reason for that is that it has to have all references to consumers and producers, which is possible if we call the adder at the highest level.

UPDATE: here is another approach, which makes adder a generator:

class AdderWithBroadcast(object):

    consumers = []

    def __init__(self, x_prod, y_prod):
        self.x_prod = x_prod
        self.y_prod = y_prod

    def __iter__(self):
        return self

    def next(self):
        x = next(self.x_prod)
        y = next(self.y_prod)

        for consumer in self.consumers:
            consumer.send(x+y)

        return x+y

    def consumer():
        while True:
            a = (yield)
            print a, ' in consumer'


k = iter(range(10))
adder = AdderWithBroadcast(k, k)
cons = consumer()
cons.send(None)
adder.consumers.append(cons)
for i in adder: 
    #  I won't include the actual result here, you can try in no your own
    print i

Decorator approach:

class Broadcaster(object):

    consumers = []

    def __init__(self, gen):
        self.gen = gen

    def __iter__(self):
        return self

    def __call__(self, *args, **kwargs):
        self.gen = self.gen(*args, **kwargs)

    def next(self):
        yielded = next(self.gen)

        for consumer in self.consumers:
            consumer.send(yielded)

        return yielded


@Broadcaster
def adder(producer1, producer2):
        while True:
            x = next(producer1)
            y = next(producer2)
            yield x + y
#  result is the same as in previous solution