As of now DuplexConnecton has a pretty straightforward mechanism for sending frames, however, it may be useful in many cases to follow the connection level backpressure so the sender will be protected by overwhelming itself.
Such a capability should be configurable by each specific transport and should be the details of implementation. However, to enable such a capability, the DuplexConnection should expose a particular API enabling a Requester/Responder to sends frames only when the connection is ready to receive them.
The API changes should look like the following:
class DuplexConnection extends Availability, Closeable {
void startStream(FrameHandler frameHandler);
...
}where now to initiate streaming the Responder/Requester should provide a Request operator to the duplex connection, so once the connection is ready to receive frames it will invoke an appropriate method called requestFrames(long n).
Note, in this context, FrameHandler represents a REQUEST operator implementation. With these changes, this interface should get a new method requestFrames which lets the startStream method to use FrameHandler as a Subscription. The main reason to avoid mixing the actual Reactive-Streams subscriptions with the FrameHandler interface is to avoid extra object allocations in the final implementation.
Subtask for this issue is #752
As of now
DuplexConnectonhas a pretty straightforward mechanism for sending frames, however, it may be useful in many cases to follow the connection level backpressure so the sender will be protected by overwhelming itself.Such a capability should be configurable by each specific transport and should be the details of implementation. However, to enable such a capability, the DuplexConnection should expose a particular API enabling a Requester/Responder to sends frames only when the connection is ready to receive them.
The API changes should look like the following:
where now to initiate streaming the Responder/Requester should provide a Request operator to the duplex connection, so once the connection is ready to receive frames it will invoke an appropriate method called
requestFrames(long n).Note, in this context,
FrameHandlerrepresents a REQUEST operator implementation. With these changes, this interface should get a new methodrequestFrameswhich lets thestartStreammethod to useFrameHandleras aSubscription. The main reason to avoid mixing the actual Reactive-Streams subscriptions with theFrameHandlerinterface is to avoid extra object allocations in the final implementation.Subtask for this issue is #752