Merge pull request #32 from rsocket/feature/latest-rsocket-java

OlegDokuka · web-flow · commit 008d5ac12855 · 2019-02-14T16:19:46.000+02:00
Feature/latest rsocket java
diff --git a/build.gradle b/build.gradle
@@ -1,8 +1,8 @@
 plugins {
-    id 'com.gradle.build-scan' version '1.16'
+    id 'com.gradle.build-scan' version '2.1'
     id 'com.google.osdetector' version '1.4.0'
-    id 'com.github.sherter.google-java-format' version '0.7.1' apply false
-    id 'com.github.johnrengelman.shadow' version '2.0.1' apply false
+    id 'com.github.sherter.google-java-format' version '0.8' apply false
+    id 'com.github.johnrengelman.shadow' version '4.0.2' apply false
     id 'com.jfrog.artifactory' version '4.7.3' apply false
     id 'com.jfrog.bintray' version '1.8.4' apply false
     id 'me.champeau.gradle.jmh' version '0.4.7' apply false
@@ -13,7 +13,10 @@ apply from: 'artifactory.gradle'
 apply from: 'bintray.gradle'
 apply from: 'publication.gradle'
 
-buildScan { licenseAgreementUrl = 'https://gradle.com/terms-of-service'; licenseAgree = 'yes' }
+buildScan {     
+    termsOfServiceUrl = 'https://gradle.com/terms-of-service'
+    termsOfServiceAgree = 'yes'
+}
 
 subprojects {
     apply plugin: 'idea'
@@ -30,7 +33,7 @@ subprojects {
     sourceCompatibility = 1.8
     targetCompatibility = 1.8
 
-    ext['reactor-bom.version'] = 'Californium-SR1'
+    ext['reactor-bom.version'] = 'Californium-SR5'
     ext['rsocket.version'] = '0.11.16'
 
     ext['protobuf.version'] = '3.6.1'
diff --git a/gradle.properties b/gradle.properties
@@ -1,2 +1,2 @@
 group=io.rsocket.rpc
-version=0.2.12
+version=0.2.13
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-4.10.2-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-5.2.1-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/rsocket-rpc-core/build.gradle b/rsocket-rpc-core/build.gradle
@@ -9,6 +9,7 @@ description = 'RSocket RPC Library'
 
 dependencies {
     implementation 'com.google.protobuf:protobuf-java'
+    implementation 'org.slf4j:slf4j-api'
 
     api 'io.opentracing:opentracing-api'
     api 'javax.inject:javax.inject'
@@ -54,6 +55,8 @@ dependencyManagement {
             entry "log4j-core"
             entry "log4j-slf4j-impl"
         }
+        
+        dependency "org.slf4j:slf4j-api:${ext['slf4j.version']}"
         dependency "org.slf4j:slf4j-simple:${ext['slf4j.version']}"
 
         dependencySet(group: 'io.rsocket', version: ext['rsocket.version']) {
diff --git a/rsocket-rpc-core/src/main/java/io/rsocket/rpc/AbstractRSocketService.java b/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."));
   }
 
diff --git a/rsocket-rpc-core/src/main/java/io/rsocket/rpc/RSocketRpcService.java b/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);
 }
diff --git a/rsocket-rpc-core/src/main/java/io/rsocket/rpc/rsocket/RequestHandlingRSocket.java b/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);
+    }
   }
 }
diff --git a/rsocket-rpc-protobuf/src/java_plugin/cpp/blocking_java_generator.cpp b/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(
diff --git a/rsocket-rpc-protobuf/src/java_plugin/cpp/java_generator.cpp b/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(

Original file line number	Diff line number	Diff line change
`@@ -1,2 +1,2 @@`
`1`	`1`	`group=io.rsocket.rpc`
`2`		`-version=0.2.12`
	`2`	`+version=0.2.13`
Original file line number	Diff line number	Diff line change
`@@ -2,6 +2,7 @@`
`2`	`2`
`3`	`3`	`import io.rsocket.AbstractRSocket;`
`4`	`4`	`import io.rsocket.Payload;`
	`5`	`+import org.reactivestreams.Publisher;`
`5`	`6`	`import reactor.core.publisher.Flux;`
`6`	`7`
`7`	`8`	`public abstract class AbstractRSocketService extends AbstractRSocket implements RSocketRpcService {`
`@@ -11,7 +12,7 @@ public String getService() {`
`11`	`12`	`}`
`12`	`13`
`13`	`14`	`@Override`
`14`		`- public Flux<Payload> requestChannel(Payload payload, Flux<Payload> publisher) {`
	`15`	`+ public Flux<Payload> requestChannel(Payload payload, Publisher<Payload> publisher) {`
`15`	`16`	`return Flux.error(new UnsupportedOperationException("Request-Channel not implemented."));`
`16`	`17`	`}`
`17`	`18`