Repository: salesforce/reactive-grpc
Pull request #86 (2018-05-30 17:54:00)

Fix server shutdown errors from gRPC 1.12.0


Patch:
diff --git a/pom.xml b/pom.xml
index 5d676b1..ef9e076 100644
--- a/pom.xml
+++ b/pom.xml
@@ -144,6 +144,12 @@
                 <version>${grpc.version}</version>
             </dependency>
             <!-- Test dependencies -->
+            <dependency>
+                <groupId>io.grpc</groupId>
+                <artifactId>grpc-testing</artifactId>
+                <version>${grpc.version}</version>
+                <scope>test</scope>
+            </dependency>
             <dependency>
                 <groupId>com.salesforce.servicelibs</groupId>
                 <artifactId>grpc-testing-contrib</artifactId>


Test patch:
diff --git a/reactor/reactor-grpc-test/pom.xml b/reactor/reactor-grpc-test/pom.xml
index 535af1d..4583b08 100644
--- a/reactor/reactor-grpc-test/pom.xml
+++ b/reactor/reactor-grpc-test/pom.xml
@@ -37,6 +37,11 @@
             <version>${reactor.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-testing</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>io.grpc</groupId>
             <artifactId>grpc-netty</artifactId>
diff --git a/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/AbstractStubTest.java b/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/AbstractStubTest.java
index e0c2548..11e0195 100644
--- a/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/AbstractStubTest.java
+++ b/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/AbstractStubTest.java
@@ -3,6 +3,8 @@ package com.salesforce.reactorgrpc;
 import io.grpc.Deadline;
 import io.grpc.ManagedChannel;
 import io.grpc.inprocess.InProcessChannelBuilder;
+import io.grpc.testing.GrpcServerRule;
+import org.junit.Rule;
 import org.junit.Test;
 
 import java.util.concurrent.TimeUnit;
@@ -10,9 +12,12 @@ import java.util.concurrent.TimeUnit;
 import static org.assertj.core.api.Assertions.assertThat;
 
 public class AbstractStubTest {
+    @Rule
+    public GrpcServerRule serverRule = new GrpcServerRule();
+
     @Test
     public void getChannelWorks() {
-        ManagedChannel channel = InProcessChannelBuilder.forName("settingCallOptionsWorks").build();
+        ManagedChannel channel = serverRule.getChannel();
         ReactorGreeterGrpc.ReactorGreeterStub stub = ReactorGreeterGrpc.newReactorStub(channel);
 
         assertThat(stub.getChannel()).isEqualTo(channel);
@@ -20,7 +25,7 @@ public class AbstractStubTest {
 
     @Test
     public void settingCallOptionsWorks() {
-        ManagedChannel channel = InProcessChannelBuilder.forName("settingCallOptionsWorks").build();
+        ManagedChannel channel = serverRule.getChannel();
         Deadline deadline = Deadline.after(42, TimeUnit.SECONDS);
 
         ReactorGreeterGrpc.ReactorGreeterStub stub = ReactorGreeterGrpc.newReactorStub(channel).withDeadline(deadline);
diff --git a/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/BackpressureIntegrationTest.java b/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/BackpressureIntegrationTest.java
index 0a9d21a..8985160 100644
--- a/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/BackpressureIntegrationTest.java
+++ b/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/BackpressureIntegrationTest.java
@@ -10,13 +10,9 @@ package com.salesforce.reactorgrpc;
 import com.google.protobuf.Empty;
 import com.salesforce.servicelibs.NumberProto;
 import com.salesforce.servicelibs.ReactorNumbersGrpc;
-import io.grpc.ManagedChannel;
-import io.grpc.Server;
-import io.grpc.inprocess.InProcessChannelBuilder;
-import io.grpc.inprocess.InProcessServerBuilder;
-import org.junit.AfterClass;
+import io.grpc.testing.GrpcServerRule;
 import org.junit.Before;
-import org.junit.BeforeClass;
+import org.junit.Rule;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -31,50 +27,44 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 @SuppressWarnings("Duplicates")
 public class BackpressureIntegrationTest {
+    @Rule
+    public GrpcServerRule serverRule = new GrpcServerRule();
+
     private static final int NUMBER_OF_STREAM_ELEMENTS = 270;
 
     private static AtomicLong lastValueTime;
     private static AtomicLong numberOfWaits;
 
-    private static Server server;
-    private static ManagedChannel channel;
-
-    @BeforeClass
-    public static void setupServer() throws Exception {
-        ReactorNumbersGrpc.NumbersImplBase svc = new ReactorNumbersGrpc.NumbersImplBase() {
-            @Override
-            public Mono<NumberProto.Number> requestPressure(Flux<NumberProto.Number> request) {
-                return request
-                        .map(proto -> proto.getNumber(0))
-                        .doOnNext(i -> System.out.println("    --> " + i))
-                        .doOnNext(i -> waitIfValuesAreEqual(i, 3))
-                        .last(-1)
-                        .map(BackpressureIntegrationTest::protoNum);
-            }
-
-            @Override
-            public Flux<NumberProto.Number> responsePressure(Mono<Empty> request) {
-                return Flux
-                        .fromIterable(IntStream.range(0, NUMBER_OF_STREAM_ELEMENTS)::iterator)
-                        .doOnNext(i -> System.out.println("   <-- " + i))
-                        .doOnNext(i -> updateNumberOfWaits(lastValueTime, numberOfWaits))
-                        .map(BackpressureIntegrationTest::protoNum);
-            }
+    private static class TestService extends ReactorNumbersGrpc.NumbersImplBase {
+        @Override
+        public Mono<NumberProto.Number> requestPressure(Flux<NumberProto.Number> request) {
+            return request
+                    .map(proto -> proto.getNumber(0))
+                    .doOnNext(i -> System.out.println("    --> " + i))
+                    .doOnNext(i -> waitIfValuesAreEqual(i, 3))
+                    .last(-1)
+                    .map(BackpressureIntegrationTest::protoNum);
+        }
 
-            @Override
-            public Flux<NumberProto.Number> twoWayRequestPressure(Flux<NumberProto.Number> request) {
-                return requestPressure(request).flux();
-            }
+        @Override
+        public Flux<NumberProto.Number> responsePressure(Mono<Empty> request) {
+            return Flux
+                    .fromIterable(IntStream.range(0, NUMBER_OF_STREAM_ELEMENTS)::iterator)
+                    .doOnNext(i -> System.out.println("   <-- " + i))
+                    .doOnNext(i -> updateNumberOfWaits(lastValueTime, numberOfWaits))
+                    .map(BackpressureIntegrationTest::protoNum);
+        }
 
-            @Override
-            public Flux<NumberProto.Number> twoWayResponsePressure(Flux<NumberProto.Number> request) {
-                request.subscribe();
-                return responsePressure(null);
-            }
-        };
+        @Override
+        public Flux<NumberProto.Number> twoWayRequestPressure(Flux<NumberProto.Number> request) {
+            return requestPressure(request).flux();
+        }
 
-        server = InProcessServerBuilder.forName("e2e").addService(svc).build().start();
-        channel = InProcessChannelBuilder.forName("e2e").usePlaintext().build();
+        @Override
+        public Flux<NumberProto.Number> twoWayResponsePressure(Flux<NumberProto.Number> request) {
+            request.subscribe();
+            return responsePressure(null);
+        }
     }
 
     @Before
@@ -83,19 +73,11 @@ public class BackpressureIntegrationTest {
         numberOfWaits = new AtomicLong(0);
     }
 
-    @AfterClass
-    public static void stopServer() throws InterruptedException {
-        server.shutdown();
-        server.awaitTermination();
-        channel.shutdown();
-
-        server = null;
-        channel = null;
-    }
-
     @Test
-    public void clientToServerBackpressure() throws InterruptedException {
-        ReactorNumbersGrpc.ReactorNumbersStub stub = ReactorNumbersGrpc.newReactorStub(channel);
+    public void clientToServerBackpressure() {
+        serverRule.getServiceRegistry().addService(new TestService());
+
+        ReactorNumbersGrpc.ReactorNumbersStub stub = ReactorNumbersGrpc.newReactorStub(serverRule.getChannel());
 
         Flux<NumberProto.Number> reactorRequest = Flux
                 .fromIterable(IntStream.range(0, NUMBER_OF_STREAM_ELEMENTS)::iterator)
@@ -114,8 +96,10 @@ public class BackpressureIntegrationTest {
     }
 
     @Test
-    public void serverToClientBackpressure() throws InterruptedException {
-        ReactorNumbersGrpc.ReactorNumbersStub stub = ReactorNumbersGrpc.newReactorStub(channel);
+    public void serverToClientBackpressure() {
+        serverRule.getServiceRegistry().addService(new TestService());
+
+        ReactorNumbersGrpc.ReactorNumbersStub stub = ReactorNumbersGrpc.newReactorStub(serverRule.getChannel());
 
         Mono<Empty> reactorRequest = Mono.just(Empty.getDefaultInstance());
 
@@ -132,8 +116,10 @@ public class BackpressureIntegrationTest {
     }
 
     @Test
-    public void bidiResponseBackpressure() throws InterruptedException {
-        ReactorNumbersGrpc.ReactorNumbersStub stub = ReactorNumbersGrpc.newReactorStub(channel);
+    public void bidiResponseBackpressure() {
+        serverRule.getServiceRegistry().addService(new TestService());
+
+        ReactorNumbersGrpc.ReactorNumbersStub stub = ReactorNumbersGrpc.newReactorStub(serverRule.getChannel());
 
         Flux<NumberProto.Number> reactorResponse = stub.twoWayResponsePressure(Flux.empty())
                 .doOnNext(n -> System.out.println(n.getNumber(0) + "  <--"))
@@ -148,8 +134,10 @@ public class BackpressureIntegrationTest {
     }
 
     @Test
-    public void bidiRequestBackpressure() throws InterruptedException {
-        ReactorNumbersGrpc.ReactorNumbersStub stub = ReactorNumbersGrpc.newReactorStub(channel);
+    public void bidiRequestBackpressure() {
+        serverRule.getServiceRegistry().addService(new TestService());
+
+        ReactorNumbersGrpc.ReactorNumbersStub stub = ReactorNumbersGrpc.newReactorStub(serverRule.getChannel());
 
         Flux<NumberProto.Number> reactorRequest = Flux
                 .fromIterable(IntStream.range(0, NUMBER_OF_STREAM_ELEMENTS)::iterator)
diff --git a/rx-java/rxgrpc-tck/src/test/java/com/salesforce/rxgrpc/tck/TckService.java b/rx-java/rxgrpc-tck/src/test/java/com/salesforce/rxgrpc/tck/TckService.java
index 2d826ae..b750b7b 100644
--- a/rx-java/rxgrpc-tck/src/test/java/com/salesforce/rxgrpc/tck/TckService.java
+++ b/rx-java/rxgrpc-tck/src/test/java/com/salesforce/rxgrpc/tck/TckService.java
@@ -26,8 +26,7 @@ public class TckService extends RxTckGrpc.TckImplBase {
                 // send back no more than 10 responses
                 .flatMap(message -> Flowable.range(1, Math.min(message.getNumber(), 10))
                         ,false, 1, 1)
-                .map(this::toMessage)
-                .doOnRequest(r -> System.out.println("Request " + r));
+                .map(this::toMessage);
     }
 
     @Override
diff --git a/rx-java/rxgrpc-test/pom.xml b/rx-java/rxgrpc-test/pom.xml
index 7ab0bd4..774faf7 100644
--- a/rx-java/rxgrpc-test/pom.xml
+++ b/rx-java/rxgrpc-test/pom.xml
@@ -56,6 +56,11 @@
             <artifactId>grpc-context</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-testing</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
diff --git a/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/AbstractStubTest.java b/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/AbstractStubTest.java
index e46b7b6..fcedab6 100644
--- a/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/AbstractStubTest.java
+++ b/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/AbstractStubTest.java
@@ -2,7 +2,8 @@ package com.salesforce.rxgrpc;
 
 import io.grpc.Deadline;
 import io.grpc.ManagedChannel;
-import io.grpc.inprocess.InProcessChannelBuilder;
+import io.grpc.testing.GrpcServerRule;
+import org.junit.Rule;
 import org.junit.Test;
 
 import java.util.concurrent.TimeUnit;
@@ -10,9 +11,12 @@ import java.util.concurrent.TimeUnit;
 import static org.assertj.core.api.Assertions.assertThat;
 
 public class AbstractStubTest {
+    @Rule
+    public GrpcServerRule serverRule = new GrpcServerRule();
+
     @Test
     public void getChannelWorks() {
-        ManagedChannel channel = InProcessChannelBuilder.forName("settingCallOptionsWorks").build();
+        ManagedChannel channel = serverRule.getChannel();
         RxGreeterGrpc.RxGreeterStub stub = RxGreeterGrpc.newRxStub(channel);
 
         assertThat(stub.getChannel()).isEqualTo(channel);
@@ -20,7 +24,7 @@ public class AbstractStubTest {
 
     @Test
     public void settingCallOptionsWorks() {
-        ManagedChannel channel = InProcessChannelBuilder.forName("settingCallOptionsWorks").build();
+        ManagedChannel channel = serverRule.getChannel();
         Deadline deadline = Deadline.after(42, TimeUnit.SECONDS);
 
         RxGreeterGrpc.RxGreeterStub stub = RxGreeterGrpc.newRxStub(channel).withDeadline(deadline);
diff --git a/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/BackpressureIntegrationTest.java b/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/BackpressureIntegrationTest.java
index ad7960f..1e5a89e 100644
--- a/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/BackpressureIntegrationTest.java
+++ b/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/BackpressureIntegrationTest.java
@@ -10,17 +10,13 @@ package com.salesforce.rxgrpc;
 import com.google.protobuf.Empty;
 import com.salesforce.servicelibs.NumberProto;
 import com.salesforce.servicelibs.RxNumbersGrpc;
-import io.grpc.ManagedChannel;
-import io.grpc.Server;
-import io.grpc.inprocess.InProcessChannelBuilder;
-import io.grpc.inprocess.InProcessServerBuilder;
+import io.grpc.testing.GrpcServerRule;
 import io.reactivex.Flowable;
 import io.reactivex.Single;
 import io.reactivex.observers.TestObserver;
 import io.reactivex.subscribers.TestSubscriber;
-import org.junit.AfterClass;
 import org.junit.Before;
-import org.junit.BeforeClass;
+import org.junit.Rule;
 import org.junit.Test;
 
 import java.util.Arrays;
@@ -32,50 +28,44 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 @SuppressWarnings("Duplicates")
 public class BackpressureIntegrationTest {
+    @Rule
+    public GrpcServerRule serverRule = new GrpcServerRule();
+
     private static final int NUMBER_OF_STREAM_ELEMENTS = 140;
 
     private static AtomicLong lastValueTime;
     private static AtomicLong numberOfWaits;
 
-    private static Server server;
-    private static ManagedChannel channel;
-
-    @BeforeClass
-    public static void setupServer() throws Exception {
-        RxNumbersGrpc.NumbersImplBase svc = new RxNumbersGrpc.NumbersImplBase() {
-            @Override
-            public Single<NumberProto.Number> requestPressure(Flowable<NumberProto.Number> request) {
-                return request
-                        .map(proto -> proto.getNumber(0))
-                        .doOnNext(i -> System.out.println("    --> " + i))
-                        .doOnNext(i -> waitIfValuesAreEqual(i, 3))
-                        .last(-1)
-                        .map(BackpressureIntegrationTest::protoNum);
-            }
-
-            @Override
-            public Flowable<NumberProto.Number> responsePressure(Single<Empty> request) {
-                return Flowable
-                        .fromIterable(IntStream.range(0, NUMBER_OF_STREAM_ELEMENTS)::iterator)
-                        .doOnNext(i -> System.out.println("   <-- " + i))
-                        .doOnNext(i -> updateNumberOfWaits(lastValueTime, numberOfWaits))
-                        .map(BackpressureIntegrationTest::protoNum);
-            }
+    private static class TestService extends RxNumbersGrpc.NumbersImplBase {
+        @Override
+        public Single<NumberProto.Number> requestPressure(Flowable<NumberProto.Number> request) {
+            return request
+                    .map(proto -> proto.getNumber(0))
+                    .doOnNext(i -> System.out.println("    --> " + i))
+                    .doOnNext(i -> waitIfValuesAreEqual(i, 3))
+                    .last(-1)
+                    .map(BackpressureIntegrationTest::protoNum);
+        }
 
-            @Override
-            public Flowable<NumberProto.Number> twoWayRequestPressure(Flowable<NumberProto.Number> request) {
-                return requestPressure(request).toFlowable();
-            }
+        @Override
+        public Flowable<NumberProto.Number> responsePressure(Single<Empty> request) {
+            return Flowable
+                    .fromIterable(IntStream.range(0, NUMBER_OF_STREAM_ELEMENTS)::iterator)
+                    .doOnNext(i -> System.out.println("   <-- " + i))
+                    .doOnNext(i -> updateNumberOfWaits(lastValueTime, numberOfWaits))
+                    .map(BackpressureIntegrationTest::protoNum);
+        }
 
-            @Override
-            public Flowable<NumberProto.Number> twoWayResponsePressure(Flowable<NumberProto.Number> request) {
-                request.subscribe();
-                return responsePressure(null);
-            }
-        };
+        @Override
+        public Flowable<NumberProto.Number> twoWayRequestPressure(Flowable<NumberProto.Number> request) {
+            return requestPressure(request).toFlowable();
+        }
 
-        server = InProcessServerBuilder.forName("e2e").addService(svc).build().start();
-        channel = InProcessChannelBuilder.forName("e2e").usePlaintext().build();
+        @Override
+        public Flowable<NumberProto.Number> twoWayResponsePressure(Flowable<NumberProto.Number> request) {
+            request.subscribe();
+            return responsePressure(null);
+        }
     }
 
     @Before
@@ -84,19 +74,10 @@ public class BackpressureIntegrationTest {
         numberOfWaits = new AtomicLong(0);
     }
 
-    @AfterClass
-    public static void stopServer() throws InterruptedException {
-        server.shutdown();
-        server.awaitTermination();
-        channel.shutdown();
-
-        server = null;
-        channel = null;
-    }
-
     @Test
     public void clientToServerBackpressure() {
-        RxNumbersGrpc.RxNumbersStub stub = RxNumbersGrpc.newRxStub(channel);
+        serverRule.getServiceRegistry().addService(new TestService());
+        RxNumbersGrpc.RxNumbersStub stub = RxNumbersGrpc.newRxStub(serverRule.getChannel());
 
         Flowable<NumberProto.Number> rxRequest = Flowable
                 .fromIterable(IntStream.range(0, NUMBER_OF_STREAM_ELEMENTS)::iterator)
@@ -115,7 +96,8 @@ public class BackpressureIntegrationTest {
 
     @Test
     public void serverToClientBackpressure() {
-        RxNumbersGrpc.RxNumbersStub stub = RxNumbersGrpc.newRxStub(channel);
+        serverRule.getServiceRegistry().addService(new TestService());
+        RxNumbersGrpc.RxNumbersStub stub = RxNumbersGrpc.newRxStub(serverRule.getChannel());
 
         Single<Empty> rxRequest = Single.just(Empty.getDefaultInstance());
 
@@ -133,7 +115,8 @@ public class BackpressureIntegrationTest {
 
     @Test
     public void bidiResponseBackpressure() {
-        RxNumbersGrpc.RxNumbersStub stub = RxNumbersGrpc.newRxStub(channel);
+        serverRule.getServiceRegistry().addService(new TestService());
+        RxNumbersGrpc.RxNumbersStub stub = RxNumbersGrpc.newRxStub(serverRule.getChannel());
 
         TestSubscriber<NumberProto.Number> rxResponse = Flowable.<NumberProto.Number>empty()
                 .compose(stub::twoWayResponsePressure)
@@ -150,7 +133,8 @@ public class BackpressureIntegrationTest {
 
     @Test
     public void bidiRequestBackpressure() {
-        RxNumbersGrpc.RxNumbersStub stub = RxNumbersGrpc.newRxStub(channel);
+        serverRule.getServiceRegistry().addService(new TestService());
+        RxNumbersGrpc.RxNumbersStub stub = RxNumbersGrpc.newRxStub(serverRule.getChannel());
 
         Flowable<NumberProto.Number> rxRequest = Flowable
                 .fromIterable(IntStream.range(0, NUMBER_OF_STREAM_ELEMENTS)::iterator)