update to the latest RSocket-Java

Author: OlegDokuka

rsocket-rpc-core/src/main/java/io/rsocket/rpc/AbstractRSocketService.java

@@ -2,6 +2,7 @@
 
 import io.rsocket.AbstractRSocket;
 import io.rsocket.Payload;
+import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 
 public abstract class AbstractRSocketService extends AbstractRSocket implements RSocketRpcService {
@@ -11,7 +12,7 @@ public String getService() {
   }
 
   @Override
-  public Flux<Payload> requestChannel(Payload payload, Flux<Payload> publisher) {
+  public Flux<Payload> requestChannel(Payload payload, Publisher<Payload> publisher) {
     return Flux.error(new UnsupportedOperationException("Request-Channel not implemented."));
   }
 

rsocket-rpc-core/src/main/java/io/rsocket/rpc/RSocketRpcService.java

@@ -1,11 +1,7 @@
 package io.rsocket.rpc;
 
-import io.rsocket.Payload;
-import io.rsocket.RSocket;
-import reactor.core.publisher.Flux;
+import io.rsocket.ResponderRSocket;
 
-public interface RSocketRpcService extends RSocket {
+public interface RSocketRpcService extends ResponderRSocket {
   String getService();
-
-  Flux<Payload> requestChannel(Payload payload, Flux<Payload> publisher);
 }

rsocket-rpc-core/src/main/java/io/rsocket/rpc/rsocket/RequestHandlingRSocket.java

@@ -4,7 +4,7 @@
 import io.netty.util.ReferenceCountUtil;
 import io.rsocket.AbstractRSocket;
 import io.rsocket.Payload;
-import io.rsocket.internal.SwitchTransformFlux;
+import io.rsocket.ResponderRSocket;
 import io.rsocket.rpc.RSocketRpcService;
 import io.rsocket.rpc.exception.ServiceNotFound;
 import io.rsocket.rpc.frames.Metadata;
@@ -14,7 +14,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-public class RequestHandlingRSocket extends AbstractRSocket {
+public class RequestHandlingRSocket extends AbstractRSocket implements ResponderRSocket {
   private final ConcurrentMap<String, RSocketRpcService> registeredServices =
       new ConcurrentHashMap<>();
 
@@ -102,25 +102,50 @@ public Flux<Payload> requestStream(Payload payload) {
 
   @Override
   public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
-    return new SwitchTransformFlux<>(
-        payloads,
-        (payload, flux) -> {
-          try {
-            ByteBuf metadata = payload.sliceMetadata();
-            String service = Metadata.getService(metadata);
-
-            RSocketRpcService rsocketService = registeredServices.get(service);
-
-            if (rsocketService == null) {
-              ReferenceCountUtil.safeRelease(payload);
-              return Flux.error(new ServiceNotFound(service));
-            }
-
-            return rsocketService.requestChannel(payload, flux);
-          } catch (Throwable t) {
-            ReferenceCountUtil.safeRelease(payload);
-            return Flux.error(t);
-          }
-        });
+    return Flux.from(payloads)
+        .switchOnFirst(
+            (firstSignal, flux) -> {
+              if (firstSignal.hasValue()) {
+                Payload payload = firstSignal.get();
+                try {
+                  ByteBuf metadata = payload.sliceMetadata();
+                  String service = Metadata.getService(metadata);
+
+                  RSocketRpcService rsocketService = registeredServices.get(service);
+
+                  if (rsocketService == null) {
+                    ReferenceCountUtil.safeRelease(payload);
+                    return Flux.error(new ServiceNotFound(service));
+                  }
+
+                  return rsocketService.requestChannel(payload, flux);
+                } catch (Throwable t) {
+                  ReferenceCountUtil.safeRelease(payload);
+                  return Flux.error(t);
+                }
+              }
+
+              return flux;
+            });
+  }
+
+  @Override
+  public Flux<Payload> requestChannel(Payload payload, Publisher<Payload> payloads) {
+    try {
+      ByteBuf metadata = payload.sliceMetadata();
+      String service = Metadata.getService(metadata);
+
+      RSocketRpcService rsocketService = registeredServices.get(service);
+
+      if (rsocketService == null) {
+        ReferenceCountUtil.safeRelease(payload);
+        return Flux.error(new ServiceNotFound(service));
+      }
+
+      return rsocketService.requestChannel(payload, payloads);
+    } catch (Throwable t) {
+      ReferenceCountUtil.safeRelease(payload);
+      return Flux.error(t);
+    }
   }
 }

rsocket-rpc-protobuf/src/java_plugin/cpp/blocking_java_generator.cpp

@@ -924,7 +924,7 @@ static void PrintServer(const ServiceDescriptor* service,
   p->Print(
       *vars,
       "@$Override$\n"
-      "public $Flux$<$Payload$> requestChannel($Payload$ payload, $Flux$<$Payload$> publisher) {\n");
+      "public $Flux$<$Payload$> requestChannel($Payload$ payload, $Publisher$<$Payload$> publisher) {\n");
   p->Indent();
   if (request_channel.empty()) {
     p->Print(
@@ -956,7 +956,7 @@ static void PrintServer(const ServiceDescriptor* service,
       p->Indent();
       p->Print(
           *vars,
-          "publisher.map(deserializer($input_type$.parser()));\n");
+          "$Flux$.from(publisher).map(deserializer($input_type$.parser()));\n");
       p->Outdent();
       if (method->server_streaming()) {
         p->Print(

rsocket-rpc-protobuf/src/java_plugin/cpp/java_generator.cpp

@@ -1311,7 +1311,7 @@ static void PrintServer(const ServiceDescriptor* service,
   p->Print(
       *vars,
       "@$Override$\n"
-      "public $Flux$<$Payload$> requestChannel($Payload$ payload, $Flux$<$Payload$> publisher) {\n");
+      "public $Flux$<$Payload$> requestChannel($Payload$ payload, $Publisher$<$Payload$> publisher) {\n");
   p->Indent();
   if (request_channel.empty()) {
     p->Print(
@@ -1344,7 +1344,7 @@ static void PrintServer(const ServiceDescriptor* service,
       p->Indent();
       p->Print(
           *vars,
-          "publisher.map(deserializer($input_type$.parser()));\n");
+          "$Flux$.from(publisher).map(deserializer($input_type$.parser()));\n");
       p->Outdent();
       if (method->server_streaming()) {
         p->Print(