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

Hi.

Your use case can be done with existing operators from reactive libraries
without introducing new protocols. Examples below use RxJava

For the multiple-producer part, you can merge() a set of known generators
implemented as Publishers:

    Publisher<T> p1 = ...
    Publisher<T> p2 = ...
    Publisher<T> p3 = ...

    Flowable<T> merged = Flowable.mergeArray(p1, p2, p3);

If the generators can appear dynamically, you can submit them to a Processor
and identity-flatMap the sequence:

    Processor<Publisher<T>> processor = PublishProcessor.create();

    Flowable<T> mergedDynamic = processor.flatMap(p -> p);

    processor.onNext(p1);
    processor.onNext(p2);
    // ...

For the multiple distinct consumer part, there is no standard RxJava
component to do it. However, I have a separate library for uncommon or
esoteric reactive components that has the DispatchWorkProcessor (
https://github.com/akarnokd/RxJavaExtensions#dispatchworkprocessor) whose
purpose is to dispatch each item to only one of the consumers ready to
receive.

    DispatchWorkProcessor<T> dispatch =
DispatchWorkProcessor.create(Schedulers.newThread());

    merged.subscribe(dispatch);

    dispatch.subscribe(consumer1);
    dispatch.subscribe(consumer2);
    dispatch.subscribe(consumer3);

For interoperating with JDK Flow types, you can use my other library for
Flow interop: https://github.com/akarnokd/RxJavaJdk9Interop#examples

Alexei Kaigorodov via Concurrency-interest <
[email protected]> ezt írta (időpont: 2019. dec. 19., Cs,
14:23):

> Consider the following structure of a multithreaded program:
>  - a single BlockingQueue
> - several Producer threads which push their results to the queue
> - several Consumer threads which take items from the queue
>
> Now I want to replace some of the Producers and Consumers with their
> asynchronous analogs: AsyncProducer and AsyncConsumer.
> First, I need to add asynchronous interfaces to the BlockingQueue, making
> it
> AsyncBlockingQueue. As a base, I can use my own implementation, not those
> from java.util.concurrent.
> For interaction between AsyncBlockingQueue and AsyncConsumer interfaces
> from
> java.util.concurrent.Flow are sutable: AsyncBlockingQueue implements
> Flow.Publisher and AsyncConsumer implements Flow.Subscriber.
> But for communication between AsyncProducer and AsyncBlockingQueue
> java.util.concurrent.Flow is not sutable.
> Let AsyncProducer implements  Flow.Publisher and AsyncBlockingQueue
> implements Flow.Subscriber.
> Then, when the queue has room to store items, it should call
> Flow.Subscription#request(n), where sum of all arguments 'n' shoud not
> exceed the amount of available memory.
> But as there are many subscriptions, which subscriptions to choose?
> The queue should know which AsyncProducers are ready to submit data,
> otherwise, the resource of its buffer memory is wasted.
> The evident solution to this problem is, Producers must play active role
> and
> so be like Subscribers, able to call Subscription#request() at their own
> discretion,
> Such a protocol I named ReverseFlow (native English speakers are invited to
> propose a better name).
> It is published  at
> https://gist.github.com/akaigoro/9506659d7d87a85a2a58a647405d85d6
>
> So I'd like to propose to include ReverseFlow interfaces in JDK, and
> provide
> a reference implementation of AsyncBlockingQueue interface, which extends
> BlockingQueue, ReverseFlow.Publisher, and Flow.Publisher.
>
>
>
>
>
>
> --
> Sent from: http://jsr166-concurrency.10961.n7.nabble.com/
> _______________________________________________
> Concurrency-interest mailing list
> [email protected]
> http://cs.oswego.edu/mailman/listinfo/concurrency-interest
>


-- 
Best regards,
David Karnok

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