Re: java.util.concurrent.Flow with opposite direction of item flow

This symmetry is superficial. The methods block for different reasons.
Which means they don't necessarily have the same representation in an
asynchronous world.

If you were to try and add symmetry, first you need a primitive to replace
the queue. Let's call it HalfQueue. It has just one method, Y put(X). Then
you "entangle" two such half-queues so the argument of put() on one end is
the return value of the put() of the other end. Then both the producers and
the consumers call put() on different ends of the "entangled" half-queues,
and block until a corresponding put() is called on the other end. The
consumers "produce" Unit objects, and heed the result of such put(), the
producers put nontrivial objects and discard the Unit "produced" by
consumer. The asymmetry should become obvious: for this to represent a
BlockingQueue with some sized buffer, the consumer end must "produce" some
Unit objects ahead of time. This difference in where they are placed in
time is the representation that the consumers drive the flow by requesting
a number of items. This is the key difference between the Producers and the
Consumers, whether it is a synchronous world with blocking or asynchronous
world.

Alex

On Sat, 21 Dec 2019, 03:40 Alexei Kaigorodov via Concurrency-interest, <
[email protected]> wrote:

> /I don't see why one would expect symmetry between consumers and
> producers./
> This symmetry stems from the original synchronous interface: when
> communicating with  BlockingQueue, both consumers and producers are callers
> of the methods of the queue (take() and put(), respectively). When a
> synchronous interface is converted to asynchronous, callers are transformed
> to subscribers, and callees are transformed to publishers. So producers and
> consumers become subscribers, and the queue becomes Publisher of 2
> different
> kinds: Flow.Publisher and ReverseFlow.Publisher. Since the signatures of
> the
> methods put() and take() are different, the resulting asynchronous
> interfaces are different: when all the preparations like subscribe() and
> request() are done and actual transmission of information can be performed,
> it is always done by the publisher side, and in our case the publisher side
> is the queue on both ends. So to move information from producer to the
> queue, the queue calls ReverseFlow.Subscriber.remove(), and to move
> information from the queue to consumer, the queue calls
> Flow.Subscriber.onNext(item).
>
>
>
> --
> Sent from: http://jsr166-concurrency.10961.n7.nabble.com/
> _______________________________________________
> Concurrency-interest mailing list
> [email protected]
> http://cs.oswego.edu/mailman/listinfo/concurrency-interest
>

_______________________________________________
Concurrency-interest mailing list
[email protected]
http://cs.oswego.edu/mailman/listinfo/concurrency-interest