Inconsistent behaviour of `requestChannel(Payload, Publisher<Payload>)` in Java and Cpp

[OlegDokuka]

Actual Behaviour

RSocket-CPP RequestChannel(Payload, Publisher)

rsocketRequester.requestChannel(Payload('a'), Flowable.justN({Payload('b'), Payload('c')}))

RSocketResponder {

requestChannel(Payload initialPayload, Flowable request) {
	std::cout << "Initial request: " << initialPayload.cloneDataToString()
              << std::endl;
	return request->map([](Payload p) {
      std::cout << "Request Stream: " << p.cloneDataToString() << std::endl;
      std::stringstream ss;
      ss << "Hello " << p.moveDataToString() << "!";
      std::string s = ss.str();
      return Payload(s);
    });
}

Produce the following output

Initial request a
Request Stream: b
Request Stream: c

RSocket-Java RSocketResponder#requestChannel(Payload, Publisher)

rsocketRequester.requestChanne(Flux.just(
    DefaultPayload.create("a"), 
    DefaultPayload.create("b"), 
    DefaultPayload.create("c")
)...;

ResponderRSocket {
   requestChannel(Payload initialPayload, Flux<Payload> request) {
       System.out.println("InitialPayload: " + initialPayload.getDataUtf8());
       return request.doOnNext(p -> {
           System.out.println("Request Stream: " + p.getDataUtf8());
       });
   }
}

Produce the following output:

Initial request a
Request Stream: a
Request Stream: b
Request Stream: c

As we may see, RSocket-Java sends the initial payload along with the rest of the stream so it may be unexpected compares to RSocket-CPP which does not include initial with the rest of the stream

Expected

Need to discuss

cc/ @rstoyanchev @linux-china @smaldini @simonbasle @rdegnan @yschimke

[yschimke]

This sounds familiar #465

[OlegDokuka]

@yschimke yeah. The same to me. I'm reworking internals right now in #761 and rewriting all streams handlers from scratch that is why asking obvious questions.

Thanks for linking the issue 🙏

[OlegDokuka]

As it was discussed at one of our TSC calls we agreed that every implementation should decide on whether to handle that via more verbose requestChannel(payload, publisher) API or with support of fluent a libraries API which allows intercepting the first payload in the stream (e.g Flux#swifchOnFirst)