Proposed Change to `DuplexConnection`

[nebhale]

Currently, the DuplexConnection looks like following:

Mono<Void> send(Publisher<Frame> frames);

Flux<Frame> receive();

As discussed in #481, this interface leaks stream ids into the main implementation of reactor-core. This is required because the coordination of each logical stream (tying an outbound Publisher<Frame> to in an inbound Publisher<Frame>) is required to be above the DuplexConnection since the sending and receiving are split into two different method invocations. There is a desire to push the existence of a stream id down into transport layer abstractly because the specification treats them as optional, but more concretely because a forthcoming HTTP/2 transport would not need stream ids and this would lead to some serious work to start treating them as optional in the main reactor-core implementation.

To that end, I propose that the DuplexConnection interface be changed to:

/**
 * Initiate an exchange as the requester.
 *
 * @param outbound the outbound {@link Frame}s
 * @return the inbound {@link Frame}s
 * @throws NullPointerException if {@code outbound} is {@code null}
 */
Flux<Frame> request(Publisher<Frame> outbound);

/**
 * Respond to an exchange as the responder.
 *
 * @param responder a {@link Function} that transforms a {@link Flux} of inbound frames to a
 *     {@link Publisher} of outbound frames
 * @throws NullPointerException if {@code responder} is {@code null}
 */
void respond(Function<Flux<Frame>, Publisher<Frame>> responder);

This interface is quite similar to the original, but with the critical change that each method invocation of request() and each invocation of the Function passed into respond() constitute a “logical” stream (no stream id necessary).

An added benefit is that the Publishers passed around in this design have the proper Reactive Streams lifecycles. For example, an client implementation of RSocket#requestResponse() could make the call request(Mono.just(frame)).single() and ensure that only a single request frame was sent and a single response frame was received. The CANCEL signal sent up the Flux from the single() would be used to signal the transport implementation to cleanup any stored state for that logical stream.

On the server side, an implementation delegating to RSocket#requestResponse() could do so with respond(incoming -> Mono.from(incoming).flatMap(rsocket::requestResponse)) (this elides dispatching based on the first incoming Frame type, so more complex, but you get the idea). Again, the proper Reactive Streams lifecycle means that CANCEL and onComplete() can be used to trigger proper cleanup in the transport at the appropriate times.

Finally this change doesn’t make decorators more difficult and in some ways makes them less complex by isolating stream-specific state. For example, the FragmentingDuplexConnection is reduced to:

@Override
public Flux<Frame> request(Publisher<Frame> requests) {
  Objects.requireNonNull(requests, "requests must not be null");

  Flux<Frame> fragmented = Flux.from(requests).concatMap(frameFragmenter);
  return getReassembled(delegate.request(fragmented));
}

@Override
public void respond(Function<Flux<Frame>, Publisher<Frame>> responder) {
  Objects.requireNonNull(responder, "responder must not be null");

  delegate.respond(
      inbound -> {
        Flux<Frame> reassembled = getReassembled(inbound);
        return Flux.from(responder.apply(reassembled)).concatMap(frameFragmenter);
      });
}

private Flux<Frame> getReassembled(Flux<Frame> flux) {
  FrameReassembler frameReassembler = createFrameReassembler(byteBufAllocator);
  return flux.handle(frameReassembler).doOnTerminate(frameReassembler::dispose);
}

Note that the reassembler state is now isolated in the logical stream and the proper Reactive Streams lifecycle provides an opportunity to cleanup when the stream either ends organically or experiences an error. An example of a decorator that doesn’t need to maintain individual stream state can be found in the MicrometerDuplexConnection:

@Override
public Flux<Frame> request(Publisher<Frame> requests) {
  Objects.requireNonNull(requests, "requests must not be null");

  Flux<Frame> metricsOutbound = Flux.from(requests).doOnNext(outboundFrameCounters);
  return delegate.request(metricsOutbound).doOnNext(inboundFrameCounters);
}

@Override
public void respond(Function<Flux<Frame>, Publisher<Frame>> responder) {
  delegate.respond(
      inbound ->
          Flux.from(responder.apply(inbound.doOnNext(inboundFrameCounters)))
              .doOnNext(outboundFrameCounters));
}

In this example, each element is counted using doOnNext() invocations that decorate each inbound and outbound flow. They use a single set of counters for all streams, and all frames that cross those streams so no stream-scoped state.

With that, let’s being discussion.

[robertroeser]

How does this deal with side-effect frames if you don't the stream id? How are you going to maintain frame order? That was really tricky, and why everything goes into a processor and then fed into a duplex connection.

The requester and responder are the RSocketClient, and RSocketServer - they can be renamed to RSocketRequest and RSocketResponder - there's a comment in a pull request about renaming somewhere, and it hasn't been done yet.

Would something likeMono<Void> send(Publisher<Payload> frames, BiFunction<Integer, Payload, Frame> handle); work? The duplex connection can supply the stream id and then we can close over it.

I need to see how this would play out with requestStream or something to have a better opinion - as I'm probably missing something.

[nebhale]

I’m not completely sure what you mean by side-effect frames, but any frame generated on a Publisher is tied to the stream id (or really to the multiplex stream, possibly implemented with a stream id) that it is generated on. So if an error occurs, the error heads down the Publisher to the transport and the transport wraps the ERROR frame with the id of the stream that it was generated on. The same thing happens in the other direction; when an ERROR frame arrives, its stream id is extracted, the proper Publisher is retrieved and the error is propagated with it.

The net result is that within a given multiplexed stream, each frame is sunk in order (you might chose a UnicastProcessor as your outbound Publisher implementation and use the SynchronousSink that it provides to guarantee order from multiple threads). The threads can interleave but when a frame goes on, but wherever it comes from its order is held, relative to other frames on that stream.

I’m guessing that I didn’t quite get the “side-effect frame” idea, so maybe some more explanation there.

As far as naming, I think the RSocketServer and RSocketClient might be OK at the highest level (the spec refers to them this way when initiating a connection), but I chose request/respond at the DuplexConnection to better match the requester/responder terminology of exchanges once the connection had been established. I’m by no means wedded to the names; they’re just send/receive with names that match the spec a little more directly Heck @smaldini suggested map and uMap to resemble some work that he and David did previously.

I don’t think the BiFunction<Integer, Payload, Frame> signature works because every implementation of it would have to deal with the possibility that Integer is null for some transports, leaking transport-specific compensation into the core, a place that shouldn’t care about it because transports wrap it post-facto.

As far as examples, I’ll spend today doing some more prototyping of a transport implementation, and some more of the end-to-end lifecycle of requestStream.

[robertroeser]

The frames that come from the doOnRequest, doOnCanels, doOnErrors side-effects,etc. - and then get sent over the network - everything right now goes into a single processor, and consumed in order.

BiFunction<Integer, Payload, Frame> - I don't think the integer would ever be null - just have it produce a stream id in the duplexconnection - the TCP one could use the counter, and the HTTP/2 could us the one in the http/2 frame.

There were lots of really weird, and time consuming bugs dealing frames over the network, streams of streams across the network, and bugs that showed up after running 20 minutes, bugs that appeared after 300,000 requests - horrible things like that, that got fixed. Like I said, I'm not discounting the change - I'm worried about any regressions now that it is working reliably. I would really need to see a rough implementation to get sense a of how this would play out.

Current naming looks like this:

Client Side                         Server Side
RSocketClient->DuplexConnection->DuplexConnection->RSocketServer
RSocketServer<-DuplexConnection<-DuplexConnection<-RSocketClient

Renaming would look like this:

Client Side                                       Server Side
RSocketRequestor->DuplexConnection->DuplexConnection->RSocketResponder
RSocketResponder<-DuplexConnection<-DuplexConnection<-RSocketRequestor

[nebhale]

I've done some straw man implementations of the RSocketRequestor and RSocketResponder. Granted they're woefully incomplete, but I think the illustrate the simplification on that side of DuplexConnection:

final class RSocketRequester implements RSocket {

  private final DuplexConnection duplexConnection;

  RSocketRequester(DuplexConnection duplexConnection) {
    this.duplexConnection = duplexConnection;
  }

  @Override
  public Flux<PayloadFrame> requestChannel(Publisher<PayloadFrame> payloadFrames) {
    return duplexConnection
        .request(payloadFrames)
        .handle(RSocketRequester::handleError)
        .cast(PayloadFrame.class);
  }

  @Override
  public Mono<Void> requestFireAndForget(RequestFireAndForgetFrame requestFrame) {
    return duplexConnection
        .request(Mono.just(requestFrame))
        .handle(RSocketRequester::handleError)
        .take(0)
        .then();
  }

  @Override
  public Mono<PayloadFrame> requestResponse(RequestResponseFrame requestFrame) {
    return duplexConnection
        .request(Mono.just(requestFrame))
        .handle(RSocketRequester::handleError)
        .cast(PayloadFrame.class)
        .single();
  }

  @Override
  public Flux<PayloadFrame> requestStream(RequestStreamFrame requestFrame) {
    return duplexConnection
        .request(Mono.just(requestFrame))
        .handle(RSocketRequester::handleError)
        .cast(PayloadFrame.class);
  }

  private static void handleError(Frame frame, SynchronousSink<Frame> sink) {
    if (frame instanceof ErrorFrame) {
      sink.error(Exceptions.from((ErrorFrame) frame));
    } else {
      sink.next(frame);
    }
  }
}

final class RSocketResponder {

  private final ByteBufAllocator byteBufAllocator;

  private final RSocket delegate;

  RSocketResponder(
      ByteBufAllocator byteBufAllocator, RSocket delegate, DuplexConnection duplexConnection) {

    this.byteBufAllocator = byteBufAllocator;
    this.delegate = delegate;

    duplexConnection.respond(this::respond);
  }

  private Publisher<Frame> respond(Flux<Frame> inbounds) {
    return inbounds
        .<Frame>flatMap(
            frame -> {
              if (frame instanceof RequestChannelFrame) {
                // TODO
              } else if (frame instanceof RequestFireAndForgetFrame) {
                return delegate
                    .requestFireAndForget((RequestFireAndForgetFrame) frame)
                    .then(Mono.empty());
              } else if (frame instanceof RequestResponseFrame) {
                return delegate.requestResponse((RequestResponseFrame) frame);
              } else if (frame instanceof RequestStreamFrame) {
                return delegate.requestStream((RequestStreamFrame) frame);
              }

              return Mono.empty();
            })
        .onErrorResume(
            t -> {
              if (t instanceof RSocketException) {
                return Mono.just(Exceptions.from(byteBufAllocator, (RSocketException) t));
              }

              return Mono.just(
                  createErrorFrame(byteBufAllocator, ErrorType.APPLICATION_ERROR, t.getMessage()));
            });
  }
}

I'd expect things like resume counters, watching for cancel frames, leasing and keepalive to be added has operators on the chains at they go in and out; decorators on a smaller scale. I think I've finally got a grip on what you mean by side-effect frames. Give me a bit to do a straw man implementation of these.

Some example usages (eliding some of the connective tissue) might look like:

final class RSocketExamples {

  private ByteBufAllocator byteBufAllocator;

  private SpringDataRepository repository;

  private RSocketRequester requester;

  void requestChannel() {
    RSocket responder =
        new RSocket() {

          @Override
          public Flux<PayloadFrame> requestChannel(Publisher<PayloadFrame> payloadFrames) {
            return Flux.from(payloadFrames)
                .flatMap(
                    payloadFrame -> {
                      int i = Integer.parseInt(payloadFrame.getDataAsUtf8());
                      return repository.findWhere(i);
                    })
                .map(s -> createPayloadFrame(byteBufAllocator, false, true, null, s));
          }
        };

    EmitterProcessor<PayloadFrame> outbound = EmitterProcessor.create();

    requester.requestChannel(outbound).subscribe();

    outbound.sink().next(createPayloadFrame(byteBufAllocator, false, true, null, "1"));
    outbound.sink().next(createPayloadFrame(byteBufAllocator, false, true, null, "2"));
    outbound.sink().next(createPayloadFrame(byteBufAllocator, false, true, null, "3"));
  }

  void requestFireAndForget() {
    RSocket responder =
        new RSocket() {

          @Override
          public Mono<Void> requestFireAndForget(RequestFireAndForgetFrame requestFrame) {
            int i = Integer.parseInt(requestFrame.getDataAsUtf8());
            return repository.insert(i);
          }
        };

    requester
        .requestFireAndForget(createRequestFireAndForgetFrame(byteBufAllocator, false, null, "1"))
        .block();

    requester
        .requestFireAndForget(createRequestFireAndForgetFrame(byteBufAllocator, false, null, "2"))
        .block();

    requester
        .requestFireAndForget(createRequestFireAndForgetFrame(byteBufAllocator, false, null, "3"))
        .block();
  }

  void requestResponse() {
    RSocket responder =
        new RSocket() {

          @Override
          public Mono<PayloadFrame> requestResponse(RequestResponseFrame requestFrame) {
            int i = Integer.parseInt(requestFrame.getDataAsUtf8());
            return repository
                .get(i)
                .map(s -> createPayloadFrame(byteBufAllocator, false, true, null, s));
          }
        };

    requester
        .requestResponse(createRequestResponseFrame(byteBufAllocator, false, null, "1"))
        .block();

    requester
        .requestResponse(createRequestResponseFrame(byteBufAllocator, false, null, "2"))
        .block();

    requester
        .requestResponse(createRequestResponseFrame(byteBufAllocator, false, null, "3"))
        .block();
  }

  void requestStream() {
    RSocket responder =
        new RSocket() {

          @Override
          public Flux<PayloadFrame> requestStream(RequestStreamFrame requestFrame) {
            int i = Integer.parseInt(requestFrame.getDataAsUtf8());
            return repository
                .findWhere(i)
                .map(s -> createPayloadFrame(byteBufAllocator, false, true, null, s));
          }
        };

    requester
        .requestStream(createRequestStreamFrame(byteBufAllocator, false, 10, null, "1"))
        .blockLast();

    requester
        .requestStream(createRequestStreamFrame(byteBufAllocator, false, 10, null, "2"))
        .blockLast();

    requester
        .requestStream(createRequestStreamFrame(byteBufAllocator, false, 10, null, "3"))
        .blockLast();
  }

  private interface SpringDataRepository {

    Flux<String> findWhere(int i);

    Mono<String> get(int i);

    Mono<Void> insert(int i);
  }
}

[nebhale]

On Friday, I really dug deep into the issue of side-effect frames and guaranteeing the ordering and behavior of them with the streams they affect. After a couple of fits and starts, I enlisted @smaldini who's expert advice was that this couldn't be done reliably or easily using reactive operators and instead should be implemented as a Reactor Operator. We had a quick go working on errors to see if it was reasonable, and once we were happy that it could work, we stopped and went on to identify all of the behaviors that this operator would need to implement.

There is no doubt that the complexity of an operator is high, but the advantages are plain to see. If you take a look at our first attempt you'll see the following (cleaned up slightly from the commit) that dictates behavior when an onError() is signaled for outbound frames (#3 below).

public void onError(Throwable t) {
  if (done) {
    Operators.onErrorDropped(t, currentContext());
    return;
  }

  done = true;

  ioActual.onNext(Exceptions.from(UnpooledByteBufAllocator.DEFAULT, t));    // 1
  ioActual.onComplete();    // 2

  Subscription s = inboundSubscriber.ioSubscription;    // 3
  if (s != null) {
    s.cancel();  
  }

  inboundSubscriber.onComplete();    // 4
}

You can explicitly see:

1. onError() received on outbound
2. Outbound Subscriber onNext(ErrorFrame)
3. Outbound Subscriber onComplete() (no more frames being sent)
4. Inbound Subscription cancel() (no more frames arriving)
5. Inbound Subscriber onComplete() (no more frames being sent)

This level of explicitness, and even more importantly, ordered orchestration is, I believe, exactly what you were looking for @robertroeser. This is readable, explicitly testable, and works with the (slightly enlarged) DuplexConnection interface that pushes stream id into the transport.

---

I'd like to continue progressing along this experiment (I'm calling it the RSocket Bow Tie since it affects four streams) but wanted to get some feedback about what you think all of the side-effect behaviors were and how you think they should behave. I've identified seven of them, but if you can think of others, or you think that these aren't actually how the system should work, please let me know.

Errors

1. ErrorFrame incoming from Responder

1. ErrorFrame received on inbound
2. Outbound Subscription cancel() (no more frames generated)
3. Outbound Subscriber onComplete() (no more frames being sent)
4. Inbound Subscription cancel() (no more frames arriving)
5. Inbound Subscriber onError() (propagate translated error frame)

2. onError() from DuplexConnection

1. onError() received on inbound
2. Outbound Subscription cancel() (no more frames generated)
3. Outbound Subscriber onComplete() (no more frames being sent)
4. Inbound Subscriber onError() (propagate error)

3. onError() from Requester

1. onError() received on outbound
2. Outbound Subscriber onNext(ErrorFrame)
3. Outbound Subscriber onComplete() (no more frames being sent)
4. Inbound Subscription cancel() (no more frames arriving)
5. Inbound Subscriber onComplete() (no more frames being sent)

Cancels

4. CancelFrame incoming from Responder

1. CancelFrame received on inbound
2. Outbound Subscription cancel() (no more frames generated)
3. Outbound Subscriber onComplete() (no more frames being sent)
4. Inbound Subscription cancel() (no more frames arriving)
5. Inbound Subscriber onComplete() (no more frames being sent)

5. cancel() from Requester

1. cancel() received on inbound
2. Outbound Subscription cancel() (no more frames generated)
3. Outbound Subscriber onNext(CancelFrame)
4. Outbound Subscriber onComplete() (no more frames being sent)
5. Inbound Subscription cancel() (no more frames arriving)

6. cancel() from DuplexConnection

1. cancel() received on outbound
2. Inbound Subscription cancel() (no more frames arriving)
3. Inbound Subscriber onComplete() (no more frames being sent)
4. Outbound Subscription cancel() (propagate cancellation)

Demand

7. request() from Requester

1. request() received on inbound
2. Outbound Subscriber onNext(RequestNFrame)

---

Again, if you can think of any more, or that I've gotten some of these wrong, I'd love to hear feedback.

[OlegDokuka]

closing this as superseded by #878