Implements dedicated Publisher/Subscriber for each request type #761

[OlegDokuka]

This PR provides fully reworked internals for all possible interactions for Rsocket requester and responder

fixes#742#641#613#641#760

[OlegDokuka]

Benches

Base Line RC7-SNAPSHOT

Benchmark Mode Cnt Score Error Units
RSocketPerf.fireAndForget thrpt 10 1919644.508 ± 55411.004 ops/s
RSocketPerf.requestChannelWithRequestAllStrategy thrpt 10 7.325 ± 0.031 ops/s
RSocketPerf.requestChannelWithRequestByOneStrategy thrpt 10 4.909 ± 0.051 ops/s
RSocketPerf.requestResponse thrpt 10 855283.194 ± 6451.949 ops/s
RSocketPerf.requestStreamWithRequestAllStrategy thrpt 10 18.015 ± 0.599 ops/s
RSocketPerf.requestStreamWithRequestByOneStrategy thrpt 10 11.161 ± 0.114 ops/s

Current Branch

Benchmark Mode Cnt Score Error Units
RSocketPerf.fireAndForget thrpt 10 2125869.835 ± 46295.155 ops/s
RSocketPerf.requestChannelWithRequestAllStrategy thrpt 10 9.901 ± 0.089 ops/s
RSocketPerf.requestChannelWithRequestByOneStrategy thrpt 10 5.727 ± 0.025 ops/s
RSocketPerf.requestResponse thrpt 10 937050.386 ± 22547.203 ops/s
RSocketPerf.requestStreamWithRequestAllStrategy thrpt 10 21.247 ± 0.560 ops/s
RSocketPerf.requestStreamWithRequestByOneStrategy thrpt 10 12.417 ± 0.184 ops/s

[OlegDokuka]

Testing Matrix Checklist

Generic

Interaction\Cases	State Machine Transition	First Frame Sent Condition	Initial RequestN	Frame RequestN	Frame Cancel	Request/Cancel Frames Serialisation	Reassembly Support	Fragmentation Support	RefCnt Validation (during subscribe)	Payload Size Validation (during subscribe)	Terminates onComplete/onError/cancel
FireAndForgetMono
RequestResponseMono
RequestStreamFlux
RequestChannelFlux

Racing Cases

[rstoyanchev]

StreamManager provides a way for requester Mono's and responder Subscriber's to access something back from RSocketRequester and RSocketResponder. This can be taken further to avoid passing other fields as well.

For example RSocketRequester and RSocketResponder could extend this (replacing StreamManager and AbstractStreamManager):

classRequesterResponderSupport {
privatefinalintmtu;
privatefinalintmaxFrameLength;
privatefinalintmaxInboundPayloadSize;
privatefinalPayloadDecoderpayloadDecoder;
privatefinalByteBufAllocatorallocator;
@NullablefinalStreamIdSupplierstreamIdSupplier;
finalIntObjectMap<FrameHandler> activeStreams;
privatefinalUnboundedProcessor<ByteBuf> sendProcessor;
publicRequesterResponderSupport(
intmtu,
intmaxFrameLength,
intmaxInboundPayloadSize,
PayloadDecoderpayloadDecoder,
ByteBufAllocatorallocator,
@NullableStreamIdSupplierstreamIdSupplier,
IntObjectMap<FrameHandler> activeStreams) {
this.activeStreams = activeStreams;
this.mtu = mtu;
this.maxFrameLength = maxFrameLength;
this.maxInboundPayloadSize = maxInboundPayloadSize;
this.payloadDecoder = payloadDecoder;
this.allocator = allocator;
this.streamIdSupplier = streamIdSupplier;
this.sendProcessor = newUnboundedProcessor<>();
}
publicintgetMtu() {
returnmtu;
}
publicintgetMaxFrameLength() {
returnmaxFrameLength;
}
publicintgetMaxInboundPayloadSize() {
returnmaxInboundPayloadSize;
}
publicPayloadDecodergetPayloadDecoder() {
returnpayloadDecoder;
}
publicByteBufAllocatorgetAllocator() {
returnallocator;
}
publicUnboundedProcessor<ByteBuf> getSendProcessor() {
returnsendProcessor;
}
publicsynchronizedintgetNextId() {
if (this.streamIdSupplier != null) {
returnthis.streamIdSupplier.nextStreamId(this.activeStreams);
}
else {
thrownewUnsupportedOperationException("Responder can not issue id");
}
}
publicsynchronizedintaddAndGetNextId(FrameHandlerframeHandler) {
if (this.streamIdSupplier != null) {
finalIntObjectMap<FrameHandler> activeStreams = this.activeStreams;
finalintstreamId = this.streamIdSupplier.nextStreamId(activeStreams);
activeStreams.put(streamId, frameHandler);
returnstreamId;
}
else {
thrownewUnsupportedOperationException("Responder can not issue id");
}
}
publicFrameHandlerget(intstreamId) {
returnthis.activeStreams.get(streamId);
}
publicbooleanremove(intstreamId, FrameHandlerframeHandler) {
returnthis.activeStreams.remove(streamId, frameHandler);
}
}

Now the requester Mono's and Flux's can be created more easily:

@OverridepublicFlux<Payload> requestStream(Payloadpayload) {
returnnewRequestStreamFlux(payload, this);
}

Likewise for responder Subscriber's:

RequestResponseSubscribersubscriber = newRequestResponseSubscriber(streamId, frame, this);

There might be other opportunities for re-use as well through such a common base class.

[rstoyanchev]

The subscription is the RequesterFrameHandler instance. Why don't we add cancel() to RequesterFrameHandler or have it extend Subscription so that only RequesterFrameHandler is passed in, making it easier to understand where cancel() is handled?

[OlegDokuka]

The idea is to avoid mixing RSocket spec frame handling with the reactive streams. First of all, it makes it impossible to understand from where method was invoked, especially when it comes to requestChannel case. Initially, it was reactive streams interfaces, but after lots of issues with understanding where and how should I handle every call (depends on the inbound vs outbound) I decided to get rid of that idea and make things fully separate

[rstoyanchev]

Couldn't the payload be validated before subscription, basically as soon as it is provided? Maybe RSocketRequester could hold this logic and make the check before creating a requester Mono/Flux, thus also re-using the logic.

[OlegDokuka]

In any case, it is going to be code duplication. The only different - that code would be in a different place. Thus, once something needed to be adjusted - we would need to look at more places and not only to Requester / Responder operators

[simonbasle]

I didn't dig far deeper after yesterday's live review. it seems @rstoyanchev has already spotted quite a few elements to improve. overall I'd try to add more comments and documentations, especially on utils classes (like what's been done in StateUtils 👍)

[OlegDokuka]

StreamManager provides a way for requester Mono's and responder Subscriber's to access something back from RSocketRequester and RSocketResponder. This can be taken further to avoid passing other fields as well.

For example RSocketRequester and RSocketResponder could extend this (replacing StreamManager and AbstractStreamManager):

classRequesterResponderSupport {
privatefinalintmtu;
privatefinalintmaxFrameLength;
privatefinalintmaxInboundPayloadSize;
privatefinalPayloadDecoderpayloadDecoder;
privatefinalByteBufAllocatorallocator;
@NullablefinalStreamIdSupplierstreamIdSupplier;
finalIntObjectMap<FrameHandler> activeStreams;
privatefinalUnboundedProcessor<ByteBuf> sendProcessor;
publicRequesterResponderSupport(
intmtu,
intmaxFrameLength,
intmaxInboundPayloadSize,
PayloadDecoderpayloadDecoder,
ByteBufAllocatorallocator,
@NullableStreamIdSupplierstreamIdSupplier,
IntObjectMap<FrameHandler> activeStreams) {
this.activeStreams = activeStreams;
this.mtu = mtu;
this.maxFrameLength = maxFrameLength;
this.maxInboundPayloadSize = maxInboundPayloadSize;
this.payloadDecoder = payloadDecoder;
this.allocator = allocator;
this.streamIdSupplier = streamIdSupplier;
this.sendProcessor = newUnboundedProcessor<>();
}
publicintgetMtu() {
returnmtu;
}
publicintgetMaxFrameLength() {
returnmaxFrameLength;
}
publicintgetMaxInboundPayloadSize() {
returnmaxInboundPayloadSize;
}
publicPayloadDecodergetPayloadDecoder() {
returnpayloadDecoder;
}
publicByteBufAllocatorgetAllocator() {
returnallocator;
}
publicUnboundedProcessor<ByteBuf> getSendProcessor() {
returnsendProcessor;
}
publicsynchronizedintgetNextId() {
if (this.streamIdSupplier != null) {
returnthis.streamIdSupplier.nextStreamId(this.activeStreams);
}
else {
thrownewUnsupportedOperationException("Responder can not issue id");
}
}
publicsynchronizedintaddAndGetNextId(FrameHandlerframeHandler) {
if (this.streamIdSupplier != null) {
finalIntObjectMap<FrameHandler> activeStreams = this.activeStreams;
finalintstreamId = this.streamIdSupplier.nextStreamId(activeStreams);
activeStreams.put(streamId, frameHandler);
returnstreamId;
}
else {
thrownewUnsupportedOperationException("Responder can not issue id");
}
}
publicFrameHandlerget(intstreamId) {
returnthis.activeStreams.get(streamId);
}
publicbooleanremove(intstreamId, FrameHandlerframeHandler) {
returnthis.activeStreams.remove(streamId, frameHandler);
}
}

Now the requester Mono's and Flux's can be created more easily:

@OverridepublicFlux<Payload> requestStream(Payloadpayload) {
returnnewRequestStreamFlux(payload, this);
}

Likewise for responder Subscriber's:

RequestResponseSubscribersubscriber = newRequestResponseSubscriber(streamId, frame, this);

There might be other opportunities for re-use as well through such a common base class.

Done. Ready for another round of review

[OlegDokuka]

Applied most of requested changes. Will be merging this one. More polishing can be done in the followups to this PR