Merge pull request #33 from rsocket/feature/overrideError

Override Error

Author: OlegDokuka

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

@@ -2,7 +2,6 @@
 
 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 {
@@ -12,7 +11,7 @@ public String getService() {
   }
 
   @Override
-  public Flux<Payload> requestChannel(Payload payload, Publisher<Payload> publisher) {
+  public Flux<Payload> requestChannel(Payload payload, Flux<Payload> publisher) {
     return Flux.error(new UnsupportedOperationException("Request-Channel not implemented."));
   }
 

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

@@ -1,7 +1,16 @@
 package io.rsocket.rpc;
 
+import io.rsocket.Payload;
 import io.rsocket.ResponderRSocket;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 
 public interface RSocketRpcService extends ResponderRSocket {
   String getService();
+
+  Flux<Payload> requestChannel(Payload payload, Flux<Payload> publisher);
+
+  default Flux<Payload> requestChannel(Payload payload, Publisher<Payload> payloads) {
+    return requestChannel(payload, Flux.from(payloads));
+  }
 }

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, $Publisher$<$Payload$> publisher) {\n");
+      "public $Flux$<$Payload$> requestChannel($Payload$ payload, $Flux$<$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,
-          "$Flux$.from(publisher).map(deserializer($input_type$.parser()));\n");
+          "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, $Publisher$<$Payload$> publisher) {\n");
+      "public $Flux$<$Payload$> requestChannel($Payload$ payload, $Flux$<$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,
-          "$Flux$.from(publisher).map(deserializer($input_type$.parser()));\n");
+          "publisher.map(deserializer($input_type$.parser()));\n");
       p->Outdent();
       if (method->server_streaming()) {
         p->Print(