{{ message }}
Repository navigation
Provide FlatMappingProcessor as an alternative to UnboundedProcessor #752
Description
The Problem
Right now in RSocket we use UnboundedProcessor which simply is an MultiProducerSingleConsumerArrayLinkedQueue.
It means that all frames from all possible interactions are enqueued into that queue with NO backpressure control and then underlying connection drains elements with the speed it can:
Such a design may lead to many problems including memory leaks and general application instability
Proposal
In order to have fine-grained control on all the levels, I propose to provide a customizable processor and introduce FlatMappingProcessor instead.
In a few words, FlatMappingProcessor should behave identically to FlatMap operator in ProjectReactor / RxJava
Main Characteristic
- Prefetch Mode for Flux
- Concurrency Control (RSocket user can specify the number of active streams)
- Small SpScBoundedArrayQueue for every Flux (
requestChannel/requestStreamcases) and general scalarMpScUnboundedLinkedArrayQueueforMonocases
Activity
[-]Replace UnboundedProcessor with FlatMap Like Processor[/-][+]Customizable SendingProcessor [UnboundedProcessor vs FlatMappingProcessor][/+][-]Customizable SendingProcessor [UnboundedProcessor vs FlatMappingProcessor][/-][+]Provide `FlatMappingProcessor` as an alternative to `UnboundedProcessor`[/+]