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):

class RequesterResponderSupport {

  private final int mtu;
  private final int maxFrameLength;
  private final int maxInboundPayloadSize;
  private final PayloadDecoder payloadDecoder;
  private final ByteBufAllocator allocator;

  @Nullable
  final StreamIdSupplier streamIdSupplier;
  final IntObjectMap<FrameHandler> activeStreams;

  private final UnboundedProcessor<ByteBuf> sendProcessor;

  public RequesterResponderSupport(
    int mtu,
    int maxFrameLength,
    int maxInboundPayloadSize,
    PayloadDecoder payloadDecoder,
    ByteBufAllocator allocator,
    @Nullable StreamIdSupplier streamIdSupplier,
    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 = new UnboundedProcessor<>();
  }

  public int getMtu() {
    return mtu;
  }

  public int getMaxFrameLength() {
    return maxFrameLength;
  }

  public int getMaxInboundPayloadSize() {
    return maxInboundPayloadSize;
  }

  public PayloadDecoder getPayloadDecoder() {
    return payloadDecoder;
  }

  public ByteBufAllocator getAllocator() {
    return allocator;
  }

  public UnboundedProcessor<ByteBuf> getSendProcessor() {
    return sendProcessor;
  }

  public synchronized int getNextId() {
    if (this.streamIdSupplier != null) {
      return this.streamIdSupplier.nextStreamId(this.activeStreams);
    }
    else {
      throw new UnsupportedOperationException("Responder can not issue id");
    }
  }

  public synchronized int addAndGetNextId(FrameHandler frameHandler) {
    if (this.streamIdSupplier != null) {
      final IntObjectMap<FrameHandler> activeStreams = this.activeStreams;
      final int streamId = this.streamIdSupplier.nextStreamId(activeStreams);

      activeStreams.put(streamId, frameHandler);

      return streamId;
    }
    else {
      throw new UnsupportedOperationException("Responder can not issue id");
    }
  }

  public FrameHandler get(int streamId) {
    return this.activeStreams.get(streamId);
  }

  public boolean remove(int streamId, FrameHandler frameHandler) {
    return this.activeStreams.remove(streamId, frameHandler);
  }
}

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

  @Override
  public Flux<Payload> requestStream(Payload payload) {
    return new RequestStreamFlux(payload, this);
  }

Likewise for responder Subscriber's:

RequestResponseSubscriber subscriber = new RequestResponseSubscriber(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):

class RequesterResponderSupport {

  private final int mtu;
  private final int maxFrameLength;
  private final int maxInboundPayloadSize;
  private final PayloadDecoder payloadDecoder;
  private final ByteBufAllocator allocator;

  @Nullable
  final StreamIdSupplier streamIdSupplier;
  final IntObjectMap<FrameHandler> activeStreams;

  private final UnboundedProcessor<ByteBuf> sendProcessor;

  public RequesterResponderSupport(
    int mtu,
    int maxFrameLength,
    int maxInboundPayloadSize,
    PayloadDecoder payloadDecoder,
    ByteBufAllocator allocator,
    @Nullable StreamIdSupplier streamIdSupplier,
    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 = new UnboundedProcessor<>();
  }

  public int getMtu() {
    return mtu;
  }

  public int getMaxFrameLength() {
    return maxFrameLength;
  }

  public int getMaxInboundPayloadSize() {
    return maxInboundPayloadSize;
  }

  public PayloadDecoder getPayloadDecoder() {
    return payloadDecoder;
  }

  public ByteBufAllocator getAllocator() {
    return allocator;
  }

  public UnboundedProcessor<ByteBuf> getSendProcessor() {
    return sendProcessor;
  }

  public synchronized int getNextId() {
    if (this.streamIdSupplier != null) {
      return this.streamIdSupplier.nextStreamId(this.activeStreams);
    }
    else {
      throw new UnsupportedOperationException("Responder can not issue id");
    }
  }

  public synchronized int addAndGetNextId(FrameHandler frameHandler) {
    if (this.streamIdSupplier != null) {
      final IntObjectMap<FrameHandler> activeStreams = this.activeStreams;
      final int streamId = this.streamIdSupplier.nextStreamId(activeStreams);

      activeStreams.put(streamId, frameHandler);

      return streamId;
    }
    else {
      throw new UnsupportedOperationException("Responder can not issue id");
    }
  }

  public FrameHandler get(int streamId) {
    return this.activeStreams.get(streamId);
  }

  public boolean remove(int streamId, FrameHandler frameHandler) {
    return this.activeStreams.remove(streamId, frameHandler);
  }
}

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

  @Override
  public Flux<Payload> requestStream(Payload payload) {
    return new RequestStreamFlux(payload, this);
  }

Likewise for responder Subscriber's:

RequestResponseSubscriber subscriber = new RequestResponseSubscriber(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