Repository: salesforce/reactive-grpc
Pull request #151 (2019-03-03 00:14:50)

Feature/jmh+improvements
This PR provides the following:

* bug fix for the case when the assembly hooks are enabled in Reactor so the cancellation may end up exceptionally with NPE
* provides some fixes in tests
* improves the performance of `AbstractStreamObserverAndPublisher` by specifying the lowerTild for next prefetch (it means that we don't have to wait for full prefetch fulfillment so upstream have some space to send, and downstream does not call request to frequently)
* provides reference JMH benchmarks
* updates versions of libs


Patch:
diff --git a/benchmarks/README.md b/benchmarks/README.md
new file mode 100644
index 0000000..f44fd93
--- /dev/null
+++ b/benchmarks/README.md
@@ -0,0 +1,1 @@
+This module contains Reference JMH tests which can be run in order to compare performance of Reactive-GRPC with the original library
diff --git a/benchmarks/pom.xml b/benchmarks/pom.xml
new file mode 100644
index 0000000..c171ecd
--- /dev/null
+++ b/benchmarks/pom.xml
@@ -0,0 +1,153 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~  Copyright (c) 2019, Salesforce.com, Inc.
+  ~  All rights reserved.
+  ~  Licensed under the BSD 3-Clause license.
+  ~  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>com.salesforce.servicelibs</groupId>
+        <artifactId>reactive-grpc</artifactId>
+        <version>0.9.1-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>benchmarks</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.reactivex.rxjava2</groupId>
+            <artifactId>rxjava</artifactId>
+            <version>${rxjava.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>reactor-grpc-stub</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>rxgrpc-stub</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-netty</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-stub</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-protobuf</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-buffer</artifactId>
+            <version>4.1.33.Final</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.openjdk.jmh</groupId>
+            <artifactId>jmh-core</artifactId>
+            <version>1.21</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.openjdk.jmh</groupId>
+            <artifactId>jmh-generator-annprocess</artifactId>
+            <version>1.21</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.hdrhistogram</groupId>
+            <artifactId>HdrHistogram</artifactId>
+            <version>2.1.10</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <extensions>
+            <extension>
+                <groupId>kr.motd.maven</groupId>
+                <artifactId>os-maven-plugin</artifactId>
+                <version>1.6.0</version>
+            </extension>
+        </extensions>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-install-plugin</artifactId>
+                <version>2.5.2</version>
+                <configuration>
+                    <!-- Don't install, since there isn't anything to install -->
+                    <skip>true</skip>
+                </configuration>
+            </plugin>
+
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>${compiler.plugin.version}</version>
+                <configuration>
+                    <source>1.8</source>
+                    <target>1.8</target>
+                </configuration>
+            </plugin>
+
+            <plugin>
+                <groupId>org.xolstice.maven.plugins</groupId>
+                <artifactId>protobuf-maven-plugin</artifactId>
+                <version>${protobuf.plugin.version}</version>
+                <configuration>
+                    <protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
+                    <pluginId>grpc-java</pluginId>
+                    <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}</pluginArtifact>
+                </configuration>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>test-compile</goal>
+                            <goal>test-compile-custom</goal>
+                        </goals>
+                        <configuration>
+                            <protocPlugins>
+                                <protocPlugin>
+                                    <id>rxgrpc</id>
+                                    <groupId>${project.groupId}</groupId>
+                                    <artifactId>rxgrpc</artifactId>
+                                    <version>${project.version}</version>
+                                    <mainClass>com.salesforce.rxgrpc.RxGrpcGenerator</mainClass>
+                                </protocPlugin>
+                                <protocPlugin>
+                                    <id>reactor-grpc</id>
+                                    <groupId>com.salesforce.servicelibs</groupId>
+                                    <artifactId>reactor-grpc</artifactId>
+                                    <version>${project.version}</version>
+                                    <mainClass>com.salesforce.reactorgrpc.ReactorGrpcGenerator</mainClass>
+                                </protocPlugin>
+                            </protocPlugins>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>
\ No newline at end of file
diff --git a/common/reactive-grpc-common/pom.xml b/common/reactive-grpc-common/pom.xml
index b2367e1..915858c 100644
--- a/common/reactive-grpc-common/pom.xml
+++ b/common/reactive-grpc-common/pom.xml
@@ -29,7 +29,7 @@
         <dependency>
             <groupId>org.reactivestreams</groupId>
             <artifactId>reactive-streams</artifactId>
-            <version>1.0.1</version>
+            <version>${reactive.streams.version}</version>
         </dependency>
         <dependency>
             <groupId>io.grpc</groupId>
diff --git a/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/AbstractStreamObserverAndPublisher.java b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/AbstractStreamObserverAndPublisher.java
index 18cd360..348f216 100644
--- a/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/AbstractStreamObserverAndPublisher.java
+++ b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/AbstractStreamObserverAndPublisher.java
@@ -21,21 +21,30 @@ import org.reactivestreams.Subscription;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
-// TODO FIX DOCS
 /**
- * ReactiveStreamObserverPublisher bridges the manual flow control idioms of gRPC and Reactive Streams. This class takes
- * messages off of a {@link StreamObserver} and feeds them into a {@link Publisher} while respecting backpressure. This
- * class is the inverse of {@link AbstractSubscriberAndProducer}.
- * <p>
- * When a {@link Publisher} is subscribed to by a {@link Subscriber}, the {@code Publisher} hands the {@code Subscriber}
- * a {@link Subscription}. When the {@code Subscriber} wants more messages from the {@code Publisher}, the
- * {@code Subscriber} calls {@link Subscription#request(long)}.
- * <p>
- * gRPC also uses the {@link CallStreamObserver#request(int)} idiom to request more messages from the stream.
- * <p>
- * To bridge the two idioms: this class implements a {@code Publisher} which delegates calls to {@code request()} to
- * a {@link CallStreamObserver} set in the constructor. When a message is generated as a response, the message is
- * delegated in the reverse so the {@code Publisher} can announce it to the Reactive Streams implementation.
+ *
+ * {@link AbstractStreamObserverAndPublisher} bridges the manual flow control idioms of
+ * gRPC and Reactive Streams. This class takes messages off of a {@link CallStreamObserver}
+ * and feeds them into a {@link Publisher} while respecting backpressure.
+ *
+ * In order to keep backpressure working at the async boundary between
+ * {@link CallStreamObserver} as an upstream, the {@link AbstractSubscriberAndProducer}
+ * enable manual flow-control by calling
+ * {@link CallStreamObserver#disableAutoInboundFlowControl()} at subscription time and
+ * prefetch specified by {@link AbstractStreamObserverAndPublisher#prefetch} number of
+ * elements at first and then keeps the number of requested element on the same level
+ * once the upstream sent amount of elements reach the specified
+ * {@link AbstractStreamObserverAndPublisher#limit}.
+ *
+ * In addition, {@link AbstractStreamObserverAndPublisher} take an additional care for
+ * some exotic cases when upstream does not respect backpressure of the
+ * Reactive-Streams {@link Subscriber} as the downstream, so
+ * {@link AbstractStreamObserverAndPublisher} employees {@link Queue} that enqueue each
+ * incoming elements and in that way keeps excess of items. In turn, to avoid
+ * {@link Queue} overwhelming, it is recommended to uses {@link Queue} with a fixed
+ * size so in the case {@link Queue#offer(Object)} returns false the sender thread will
+ * be blocked with a busy spin using {@link LockSupport#parkNanos(long)} try to enqueue
+ * element into {@link Queue} until downstream deque element from it.
  *
  * @param <T>
  */
@@ -55,7 +64,8 @@ public abstract class AbstractStreamObserverAndPublisher<T> extends AbstractUnim
     };
 
 
-    protected static final int DEFAULT_CHUNK_SIZE = 16;
+    protected static final int DEFAULT_CHUNK_SIZE = 512;
+    protected static final int TWO_THIRDS_OF_DEFAULT_CHUNK_SIZE = DEFAULT_CHUNK_SIZE * 2 / 3;
 
     private static final int UNSUBSCRIBED_STATE    = 0;
     private static final int SUBSCRIBED_ONCE_STATE = 1;
@@ -67,6 +77,7 @@ public abstract class AbstractStreamObserverAndPublisher<T> extends AbstractUnim
 
     private final Queue<T> queue;
     private final int prefetch;
+    private final int limit;
 
     private final Consumer<CallStreamObserver<?>> onSubscribe;
 
@@ -103,29 +114,32 @@ public abstract class AbstractStreamObserverAndPublisher<T> extends AbstractUnim
     AbstractStreamObserverAndPublisher(
             Queue<T> queue,
             Consumer<CallStreamObserver<?>> onSubscribe) {
-        this(queue, DEFAULT_CHUNK_SIZE, onSubscribe);
+        this(queue, DEFAULT_CHUNK_SIZE, TWO_THIRDS_OF_DEFAULT_CHUNK_SIZE, onSubscribe);
     }
 
     AbstractStreamObserverAndPublisher(
             Queue<T> queue,
             Consumer<CallStreamObserver<?>> onSubscribe,
             Runnable onTerminate) {
-        this(queue, DEFAULT_CHUNK_SIZE, onSubscribe, onTerminate);
+        this(queue, DEFAULT_CHUNK_SIZE, TWO_THIRDS_OF_DEFAULT_CHUNK_SIZE, onSubscribe, onTerminate);
     }
 
     AbstractStreamObserverAndPublisher(
             Queue<T> queue,
             int prefetch,
+            int lowTide,
             Consumer<CallStreamObserver<?>> onSubscribe) {
-        this(queue, prefetch, onSubscribe, null);
+        this(queue, prefetch, lowTide, onSubscribe, null);
     }
 
     AbstractStreamObserverAndPublisher(
             Queue<T> queue,
             int prefetch,
+            int lowTide,
             Consumer<CallStreamObserver<?>> onSubscribe,
             Runnable onTerminate) {
         this.prefetch = prefetch;
+        this.limit = lowTide;
         this.queue = queue;
         this.onSubscribe = onSubscribe;
         this.onTerminate = onTerminate;
@@ -178,7 +192,7 @@ public abstract class AbstractStreamObserverAndPublisher<T> extends AbstractUnim
 
                 sent++;
 
-                if (sent == prefetch) {
+                if (sent == limit) {
                     if (r != Long.MAX_VALUE) {
                         r = REQUESTED.addAndGet(this, -sent);
                     }
@@ -418,7 +432,7 @@ public abstract class AbstractStreamObserverAndPublisher<T> extends AbstractUnim
         T v = queue.poll();
         if (v != null) {
             int p = produced + 1;
-            if (p == prefetch) {
+            if (p == limit) {
                 produced = 0;
                 subscription.request(p);
             } else {
diff --git a/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/AbstractSubscriberAndProducer.java b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/AbstractSubscriberAndProducer.java
index 96c2051..bde2e09 100644
--- a/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/AbstractSubscriberAndProducer.java
+++ b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/AbstractSubscriberAndProducer.java
@@ -113,7 +113,7 @@ public abstract class AbstractSubscriberAndProducer<T> implements Subscriber<T>,
             s.cancel();
 
             if (WIP.getAndIncrement(this) == 0) {
-                if (s instanceof Queue) {
+                if (sourceMode != NONE) {
                     ((Queue) s).clear();
                 }
             }
diff --git a/demos/backpressure-demo/src/main/java/demo/backpressure/BackpressureController.java b/demos/backpressure-demo/src/main/java/demo/backpressure/BackpressureController.java
index ca26088..0c89082 100644
--- a/demos/backpressure-demo/src/main/java/demo/backpressure/BackpressureController.java
+++ b/demos/backpressure-demo/src/main/java/demo/backpressure/BackpressureController.java
@@ -100,7 +100,7 @@ public class BackpressureController extends RxBackpressureDemoGrpc.BackpressureD
 
     @FXML
     public void initialize() throws Exception {
-        Server server = ServerBuilder.forPort(0).addService(this).build().start();
+        Server server = ServerBuilder.forPort(9000).addService(this).build().start();
         Channel channel = ManagedChannelBuilder.forAddress("localhost", server.getPort()).usePlaintext().build();
         stub = RxBackpressureDemoGrpc.newRxStub(channel);
 
diff --git a/pom.xml b/pom.xml
index 9a0ea14..d8d53e3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -32,6 +32,11 @@
             <name>Christophe Bornet</name>
             <email>cbornet@hotmail.com</email>
         </developer>
+        <developer>
+            <id>olegdokuka</id>
+            <name>Oleh Dokuka</name>
+            <email>shadowgun@i.ua</email>
+        </developer>
     </developers>
 
     <modules>
@@ -46,16 +51,20 @@
         <module>reactor/reactor-grpc-tck</module>
         <module>reactor/reactor-grpc-test</module>
         <module>reactor/reactor-grpc-test-32</module>
+        <module>benchmarks</module>
     </modules>
 
     <properties>
         <!-- Dependency Versions -->
-        <grpc.version>1.12.0</grpc.version>
-        <protoc.version>3.5.1</protoc.version> <!-- Same version as grpc-proto -->
+        <reactive.streams.version>1.0.2</reactive.streams.version>
+        <grpc.version>1.19.0</grpc.version>
+        <protoc.version>3.6.1</protoc.version> <!-- Same version as grpc-proto -->
+        <protobuf.plugin.version>0.6.1</protobuf.plugin.version>
+        <compiler.plugin.version>3.8.0</compiler.plugin.version>
         <grpc.contrib.version>0.8.1</grpc.contrib.version>
         <jprotoc.version>0.9.1</jprotoc.version>
-        <rxjava.version>2.1.10</rxjava.version>
-        <reactor.version>3.1.14.RELEASE</reactor.version>
+        <rxjava.version>2.2.7</rxjava.version>
+        <reactor.version>3.1.15.RELEASE</reactor.version>
         <reactor.extra.version>3.1.9.RELEASE</reactor.extra.version>
 
         <!-- Test Dependency Versions -->
@@ -187,7 +196,7 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
-                <version>3.6.1</version>
+                <version>${compiler.plugin.version}</version>
                 <configuration>
                     <source>${maven.compiler.source}</source>
                     <target>${maven.compiler.target}</target>
@@ -225,7 +234,7 @@
             <plugin>
                 <groupId>org.jacoco</groupId>
                 <artifactId>jacoco-maven-plugin</artifactId>
-                <version>0.7.9</version>
+                <version>0.8.3</version>
                 <executions>
                     <execution>
                         <goals>
@@ -252,7 +261,7 @@
                     <plugin>
                         <groupId>org.apache.maven.plugins</groupId>
                         <artifactId>maven-javadoc-plugin</artifactId>
-                        <version>2.10.4</version>
+                        <version>3.0.1</version>
                         <configuration>
                             <additionalparam>-Xdoclint:none</additionalparam>
                         </configuration>
diff --git a/reactor/reactor-grpc-stub/pom.xml b/reactor/reactor-grpc-stub/pom.xml
index f45d4b6..00ede5a 100644
--- a/reactor/reactor-grpc-stub/pom.xml
+++ b/reactor/reactor-grpc-stub/pom.xml
@@ -72,7 +72,7 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
-                <version>3.6.1</version>
+                <version>${compiler.plugin.version}</version>
                 <configuration>
                     <source>1.8</source>
                     <target>1.8</target>
diff --git a/reactor/reactor-grpc-tck/pom.xml b/reactor/reactor-grpc-tck/pom.xml
index bdae7a1..02e22ee 100644
--- a/reactor/reactor-grpc-tck/pom.xml
+++ b/reactor/reactor-grpc-tck/pom.xml
@@ -49,9 +49,19 @@
         <dependency>
             <groupId>org.reactivestreams</groupId>
             <artifactId>reactive-streams-tck</artifactId>
-            <version>1.0.1</version>
+            <version>${reactive.streams.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-buffer</artifactId>
+            <version>4.1.33.Final</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 
     <build>
@@ -59,7 +69,7 @@
             <extension>
                 <groupId>kr.motd.maven</groupId>
                 <artifactId>os-maven-plugin</artifactId>
-                <version>1.4.1.Final</version>
+                <version>1.6.0</version>
             </extension>
         </extensions>
         <plugins>
@@ -76,7 +86,7 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
-                <version>3.6.1</version>
+                <version>${compiler.plugin.version}</version>
                 <configuration>
                     <source>1.8</source>
                     <target>1.8</target>
@@ -86,7 +96,7 @@
             <plugin>
                 <groupId>org.xolstice.maven.plugins</groupId>
                 <artifactId>protobuf-maven-plugin</artifactId>
-                <version>0.5.0</version>
+                <version>${protobuf.plugin.version}</version>
                 <configuration>
                     <protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
                     <pluginId>grpc-java</pluginId>
diff --git a/rx-java/rxgrpc-tck/pom.xml b/rx-java/rxgrpc-tck/pom.xml
index fcc0847..4c84bd4 100644
--- a/rx-java/rxgrpc-tck/pom.xml
+++ b/rx-java/rxgrpc-tck/pom.xml
@@ -49,7 +49,7 @@
         <dependency>
             <groupId>org.reactivestreams</groupId>
             <artifactId>reactive-streams-tck</artifactId>
-            <version>1.0.2</version>
+            <version>${reactive.streams.version}</version>
             <scope>test</scope>
         </dependency>
     </dependencies>
@@ -59,7 +59,7 @@
             <extension>
                 <groupId>kr.motd.maven</groupId>
                 <artifactId>os-maven-plugin</artifactId>
-                <version>1.4.1.Final</version>
+                <version>1.6.0</version>
             </extension>
         </extensions>
         <plugins>
@@ -76,7 +76,7 @@
             <plugin>
                 <groupId>org.xolstice.maven.plugins</groupId>
                 <artifactId>protobuf-maven-plugin</artifactId>
-                <version>0.5.0</version>
+                <version>${protobuf.plugin.version}</version>
                 <configuration>
                     <protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
                     <pluginId>grpc-java</pluginId>


Test patch:
diff --git a/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/BenchmarkGRpcServerServiceImpl.java b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/BenchmarkGRpcServerServiceImpl.java
new file mode 100644
index 0000000..1aa4a4b
--- /dev/null
+++ b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/BenchmarkGRpcServerServiceImpl.java
@@ -0,0 +1,111 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.rxgrpc.jmh;
+
+import java.util.Arrays;
+import java.util.List;
+
+import com.salesforce.reactivegrpc.jmh.proto.BenchmarkServiceGrpc;
+import com.salesforce.reactivegrpc.jmh.proto.Messages;
+import io.grpc.stub.ServerCallStreamObserver;
+import io.grpc.stub.StreamObserver;
+
+public class BenchmarkGRpcServerServiceImpl
+    extends BenchmarkServiceGrpc.BenchmarkServiceImplBase {
+
+    final Messages.SimpleResponse       response;
+    final List<Messages.SimpleResponse> responses;
+
+    public BenchmarkGRpcServerServiceImpl(int times) {
+        Messages.SimpleResponse[] array = new Messages.SimpleResponse[times];
+        Arrays.fill(array, Messages.SimpleResponse.getDefaultInstance());
+
+        this.responses = Arrays.asList(array);
+        this.response = Messages.SimpleResponse.getDefaultInstance();
+    }
+
+    @Override
+    public void unaryCall(Messages.SimpleRequest request,
+        StreamObserver<Messages.SimpleResponse> responseObserver) {
+        responseObserver.onNext(response);
+        responseObserver.onCompleted();
+    }
+
+    @Override
+    public void streamingFromServer(Messages.SimpleRequest request, StreamObserver<Messages.SimpleResponse> responseObserver) {
+        final ServerCallStreamObserver<Messages.SimpleResponse> callStreamObserver = (ServerCallStreamObserver<Messages.SimpleResponse>) responseObserver;
+
+        for (Messages.SimpleResponse response : responses) {
+            if (callStreamObserver.isCancelled()) {
+                return;
+            }
+
+            callStreamObserver.onNext(response);
+        }
+
+        if (!callStreamObserver.isCancelled()) {
+            callStreamObserver.onCompleted();
+        }
+    }
+
+    @Override
+    public StreamObserver<Messages.SimpleRequest> streamingFromClient(StreamObserver<Messages.SimpleResponse> responseObserver) {
+        return new StreamObserver<Messages.SimpleRequest>() {
+            @Override
+            public void onNext(Messages.SimpleRequest value) {
+
+            }
+
+            @Override
+            public void onError(Throwable t) {
+                responseObserver.onError(t);
+            }
+
+            @Override
+            public void onCompleted() {
+                responseObserver.onNext(response);
+                responseObserver.onCompleted();
+            }
+        };
+    }
+
+    @Override
+    public StreamObserver<Messages.SimpleRequest> streamingBothWays(StreamObserver<Messages.SimpleResponse> responseObserver) {
+        final ServerCallStreamObserver<Messages.SimpleResponse> callStreamObserver = (ServerCallStreamObserver<Messages.SimpleResponse>) responseObserver;
+
+        callStreamObserver.setOnReadyHandler(() -> {
+            for (Messages.SimpleResponse response : responses) {
+                if (callStreamObserver.isCancelled()) {
+                    return;
+                }
+
+                callStreamObserver.onNext(response);
+            }
+
+            if (!callStreamObserver.isCancelled()) {
+                callStreamObserver.onCompleted();
+            }
+        });
+
+        return new StreamObserver<Messages.SimpleRequest>() {
+            @Override
+            public void onNext(Messages.SimpleRequest value) {
+
+            }
+
+            @Override
+            public void onError(Throwable t) {
+
+            }
+
+            @Override
+            public void onCompleted() {
+
+            }
+        };
+    }
+}
diff --git a/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/BenchmarkReactorServerServiceImpl.java b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/BenchmarkReactorServerServiceImpl.java
new file mode 100644
index 0000000..ea2fbad
--- /dev/null
+++ b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/BenchmarkReactorServerServiceImpl.java
@@ -0,0 +1,50 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactivegrpc.jmh;
+
+import java.util.Arrays;
+
+import com.salesforce.reactivegrpc.jmh.proto.Messages;
+import com.salesforce.reactivegrpc.jmh.proto.ReactorBenchmarkServiceGrpc;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class BenchmarkReactorServerServiceImpl extends
+                                        ReactorBenchmarkServiceGrpc.BenchmarkServiceImplBase {
+
+    final Mono<Messages.SimpleResponse> responseMono;
+    final Flux<Messages.SimpleResponse> responseFlux;
+
+    public BenchmarkReactorServerServiceImpl(int times) {
+        Messages.SimpleResponse[] array = new Messages.SimpleResponse[times];
+        Arrays.fill(array, Messages.SimpleResponse.getDefaultInstance());
+
+        this.responseFlux = Flux.fromArray(array);
+        this.responseMono = Mono.just(Messages.SimpleResponse.getDefaultInstance());
+    }
+
+    @Override
+    public Mono<Messages.SimpleResponse> unaryCall(Mono<Messages.SimpleRequest> request) {
+        return request.then(responseMono);
+    }
+
+    @Override
+    public Flux<Messages.SimpleResponse> streamingFromServer(Mono<Messages.SimpleRequest> request) {
+        return request.thenMany(responseFlux);
+    }
+
+    @Override
+    public Mono<Messages.SimpleResponse> streamingFromClient(Flux<Messages.SimpleRequest> request) {
+        return request.then(responseMono);
+    }
+
+    @Override
+    public Flux<Messages.SimpleResponse> streamingBothWays(Flux<Messages.SimpleRequest> request) {
+        request.subscribe();
+        return responseFlux;
+    }
+}
diff --git a/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/BenchmarkRxServerServiceImpl.java b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/BenchmarkRxServerServiceImpl.java
new file mode 100644
index 0000000..1f0bfc4
--- /dev/null
+++ b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/BenchmarkRxServerServiceImpl.java
@@ -0,0 +1,50 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactivegrpc.jmh;
+
+import java.util.Arrays;
+
+import com.salesforce.reactivegrpc.jmh.proto.Messages;
+import com.salesforce.reactivegrpc.jmh.proto.RxBenchmarkServiceGrpc;
+import io.reactivex.Flowable;
+import io.reactivex.Single;
+
+public class BenchmarkRxServerServiceImpl extends
+                                        RxBenchmarkServiceGrpc.BenchmarkServiceImplBase {
+
+    final Single<Messages.SimpleResponse>   responseMono;
+    final Flowable<Messages.SimpleResponse> responseFlux;
+
+    public BenchmarkRxServerServiceImpl(int times) {
+        Messages.SimpleResponse[] array = new Messages.SimpleResponse[times];
+        Arrays.fill(array, Messages.SimpleResponse.getDefaultInstance());
+
+        this.responseFlux = Flowable.fromArray(array);
+        this.responseMono = Single.just(Messages.SimpleResponse.getDefaultInstance());
+    }
+
+    @Override
+    public Single<Messages.SimpleResponse> unaryCall(Single<Messages.SimpleRequest> request) {
+        return request.ignoreElement().andThen(responseMono);
+    }
+
+    @Override
+    public Flowable<Messages.SimpleResponse> streamingFromServer(Single<Messages.SimpleRequest> request) {
+        return request.ignoreElement().andThen(responseFlux);
+    }
+
+    @Override
+    public Single<Messages.SimpleResponse> streamingFromClient(Flowable<Messages.SimpleRequest> request) {
+        return request.ignoreElements().andThen(responseMono);
+    }
+
+    @Override
+    public Flowable<Messages.SimpleResponse> streamingBothWays(Flowable<Messages.SimpleRequest> request) {
+        request.subscribe();
+        return responseFlux;
+    }
+}
diff --git a/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/JMHRunner.java b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/JMHRunner.java
new file mode 100644
index 0000000..63018c3
--- /dev/null
+++ b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/JMHRunner.java
@@ -0,0 +1,18 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactivegrpc.jmh;
+
+import java.io.IOException;
+
+import org.openjdk.jmh.runner.RunnerException;
+
+public class JMHRunner {
+
+    public static void main(String... args) throws IOException, RunnerException {
+        org.openjdk.jmh.Main.main(args);
+    }
+}
diff --git a/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/PerfObserver.java b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/PerfObserver.java
new file mode 100644
index 0000000..e22fd1b
--- /dev/null
+++ b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/PerfObserver.java
@@ -0,0 +1,49 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactivegrpc.jmh;
+
+import java.util.concurrent.CountDownLatch;
+
+import com.salesforce.reactivegrpc.jmh.proto.Messages;
+import io.grpc.stub.ClientCallStreamObserver;
+import io.grpc.stub.ClientResponseObserver;
+import org.openjdk.jmh.infra.Blackhole;
+
+public class PerfObserver implements ClientResponseObserver<Messages.SimpleRequest, Messages.SimpleResponse> {
+
+    final Blackhole      blackhole;
+    final CountDownLatch latch;
+
+    ClientCallStreamObserver observer;
+
+    public PerfObserver(Blackhole blackhole) {
+        this.blackhole = blackhole;
+        this.latch = new CountDownLatch(1);
+    }
+
+    @Override
+    public void beforeStart(ClientCallStreamObserver<Messages.SimpleRequest> observer) {
+        this.observer = observer;
+    }
+
+    @Override
+    public void onNext(Messages.SimpleResponse response) {
+        blackhole.consume(response);
+    }
+
+    @Override
+    public void onError(Throwable throwable) {
+        throwable.printStackTrace();
+        blackhole.consume(throwable);
+        latch.countDown();
+    }
+
+    @Override
+    public void onCompleted() {
+        latch.countDown();
+    }
+}
diff --git a/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/PerfSingleObserver.java b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/PerfSingleObserver.java
new file mode 100644
index 0000000..61e84e1
--- /dev/null
+++ b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/PerfSingleObserver.java
@@ -0,0 +1,40 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactivegrpc.jmh;
+
+import java.util.concurrent.CountDownLatch;
+
+import io.reactivex.SingleObserver;
+import io.reactivex.disposables.Disposable;
+import org.openjdk.jmh.infra.Blackhole;
+
+public final class PerfSingleObserver implements SingleObserver<Object> {
+
+    final Blackhole bh;
+    final CountDownLatch latch;
+
+    public PerfSingleObserver(Blackhole bh) {
+        this.bh = bh;
+        this.latch = new CountDownLatch(1);
+    }
+
+    @Override
+    public void onSubscribe(Disposable subscription) {
+    }
+
+    @Override
+    public void onSuccess(Object item) {
+        bh.consume(item);
+    }
+
+    @Override
+    public void onError(Throwable throwable) {
+        throwable.printStackTrace();
+        bh.consume(throwable);
+        latch.countDown();
+    }
+}
diff --git a/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/PerfSubscriber.java b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/PerfSubscriber.java
new file mode 100644
index 0000000..395feb7
--- /dev/null
+++ b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/PerfSubscriber.java
@@ -0,0 +1,51 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactivegrpc.jmh;
+
+import java.util.concurrent.CountDownLatch;
+
+import org.openjdk.jmh.infra.Blackhole;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+public final class PerfSubscriber implements Subscriber<Object> {
+
+    final Blackhole bh;
+    final CountDownLatch latch;
+
+    Subscription subscription;
+
+    public PerfSubscriber(Blackhole bh) {
+        this.bh = bh;
+        this.latch = new CountDownLatch(1);
+    }
+
+    @Override
+    public void onSubscribe(Subscription subscription) {
+        this.subscription = subscription;
+        subscription.request(1);
+    }
+
+    @Override
+    public void onNext(Object item) {
+        bh.consume(item);
+        subscription.request(1);
+    }
+
+    @Override
+    public void onError(Throwable throwable) {
+        throwable.printStackTrace();
+        bh.consume(throwable);
+        latch.countDown();
+    }
+
+    @Override
+    public void onComplete() {
+        latch.countDown();
+    }
+
+}
diff --git a/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/ReactorGRpcBenchmark.java b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/ReactorGRpcBenchmark.java
new file mode 100644
index 0000000..d27e696
--- /dev/null
+++ b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/ReactorGRpcBenchmark.java
@@ -0,0 +1,137 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactivegrpc.jmh;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import com.salesforce.reactivegrpc.jmh.proto.Messages;
+import com.salesforce.reactivegrpc.jmh.proto.ReactorBenchmarkServiceGrpc;
+import io.grpc.ManagedChannel;
+import io.grpc.Server;
+import io.grpc.inprocess.InProcessChannelBuilder;
+import io.grpc.inprocess.InProcessServerBuilder;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@BenchmarkMode(Mode.Throughput)
+@Warmup(iterations = 10)
+@Measurement(iterations = 10, time = 10, timeUnit = TimeUnit.SECONDS)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Fork(1)
+@State(Scope.Benchmark)
+public class ReactorGRpcBenchmark {
+
+    static final Mono<Messages.SimpleRequest> MONO_REQUEST =
+        Mono.just(Messages.SimpleRequest.getDefaultInstance());
+
+    static final Flux<Messages.SimpleRequest> FLUX_REQUEST;
+
+    static {
+        Messages.SimpleRequest[] array = new Messages.SimpleRequest[100000];
+        Arrays.fill(array, Messages.SimpleRequest.getDefaultInstance());
+
+        FLUX_REQUEST = Flux.fromArray(array);
+    }
+
+
+    Server         reactiveServer;
+    ManagedChannel reactiveChannel;
+
+    ReactorBenchmarkServiceGrpc.ReactorBenchmarkServiceStub reactiveClient;
+
+    @Setup
+    public void setup() throws IOException {
+        System.out.println("---------- SETUP ONCE -------------");
+        ScheduledExecutorService scheduledExecutorService =
+            Executors.newScheduledThreadPool(Runtime.getRuntime()
+                                                    .availableProcessors());
+        reactiveServer =
+            InProcessServerBuilder.forName("benchmark-reactiveServer")
+                                  .scheduledExecutorService(scheduledExecutorService)
+                                  .addService(new BenchmarkReactorServerServiceImpl(100000))
+                                  .build()
+                                  .start();
+
+        reactiveChannel = InProcessChannelBuilder.forName("benchmark-reactiveServer")
+                                                 .build();
+        reactiveClient = ReactorBenchmarkServiceGrpc.newReactorStub(reactiveChannel);
+    }
+
+    @TearDown
+    public void tearDown() throws InterruptedException {
+        System.out.println("---------- TEAR DOWN ONCE -------------");
+        reactiveServer.shutdownNow();
+        reactiveChannel.shutdownNow();
+        reactiveServer.awaitTermination(1000, TimeUnit.MILLISECONDS);
+        reactiveChannel.awaitTermination(1000, TimeUnit.MILLISECONDS);
+    }
+
+    @Benchmark
+    public Object reactiveUnaryCall(Blackhole blackhole) throws InterruptedException {
+        PerfSubscriber actual = new PerfSubscriber(blackhole);
+
+        reactiveClient.unaryCall(MONO_REQUEST)
+                      .subscribe(actual);
+        actual.latch.await();
+
+        return actual;
+    }
+
+    @Benchmark
+    public Object reactiveServerStreamingCall(Blackhole blackhole)
+        throws InterruptedException {
+        PerfSubscriber actual = new PerfSubscriber(blackhole);
+
+        reactiveClient.streamingFromServer(MONO_REQUEST)
+                      .subscribe(actual);
+
+        actual.latch.await();
+
+        return actual;
+    }
+
+    @Benchmark
+    public Object reactiveClientStreamingCall(Blackhole blackhole) throws InterruptedException {
+        PerfSubscriber actual = new PerfSubscriber(blackhole);
+
+        reactiveClient.streamingFromClient(FLUX_REQUEST)
+                      .subscribe(actual);
+
+        actual.latch.await();
+
+        return actual;
+    }
+
+
+    @Benchmark
+    public Object reactiveBothWaysStreamingCall(Blackhole blackhole) throws InterruptedException {
+        PerfSubscriber actual = new PerfSubscriber(blackhole);
+
+        reactiveClient.streamingBothWays(FLUX_REQUEST)
+                      .subscribe(actual);
+
+        actual.latch.await();
+
+        return actual;
+    }
+}
diff --git a/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/ReferenceGRpcBenchmark.java b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/ReferenceGRpcBenchmark.java
new file mode 100644
index 0000000..f7a5874
--- /dev/null
+++ b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/ReferenceGRpcBenchmark.java
@@ -0,0 +1,148 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactivegrpc.jmh;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import com.salesforce.reactivegrpc.jmh.proto.BenchmarkServiceGrpc;
+import com.salesforce.reactivegrpc.jmh.proto.Messages;
+import io.grpc.ManagedChannel;
+import io.grpc.Server;
+import io.grpc.inprocess.InProcessChannelBuilder;
+import io.grpc.inprocess.InProcessServerBuilder;
+import io.grpc.stub.ClientCallStreamObserver;
+import io.grpc.stub.StreamObserver;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+@BenchmarkMode(Mode.Throughput)
+@Warmup(iterations = 10)
+@Measurement(iterations = 10, time = 10, timeUnit = TimeUnit.SECONDS)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Fork(1)
+@State(Scope.Benchmark)
+public class ReferenceGRpcBenchmark {
+
+    static final Messages.SimpleRequest   REQUEST       = Messages.SimpleRequest.getDefaultInstance();
+    static final Messages.SimpleRequest[] ARRAY_REQUEST = new Messages.SimpleRequest[100000];
+
+    static {
+        Arrays.fill(ARRAY_REQUEST, Messages.SimpleRequest.getDefaultInstance());
+    }
+
+    Server         gRpcServer;
+    ManagedChannel gRpcChannel;
+
+    BenchmarkServiceGrpc.BenchmarkServiceStub gRpcClient;
+
+    @Setup
+    public void setup() throws IOException {
+        System.out.println("---------- SETUP ONCE -------------");
+        ScheduledExecutorService scheduledExecutorService =
+            Executors.newScheduledThreadPool(Runtime.getRuntime()
+                                                    .availableProcessors());
+        gRpcServer =
+            InProcessServerBuilder.forName("benchmark-gRpcServer")
+                                  .scheduledExecutorService(scheduledExecutorService)
+                                  .addService(new com.salesforce.rxgrpc.jmh.BenchmarkGRpcServerServiceImpl(100000))
+                                  .build()
+                                  .start();
+        gRpcChannel = InProcessChannelBuilder.forName("benchmark-gRpcServer")
+                                             .build();
+        gRpcClient = BenchmarkServiceGrpc.newStub(gRpcChannel);
+    }
+
+    @TearDown
+    public void tearDown() throws InterruptedException {
+        System.out.println("---------- TEAR DOWN ONCE -------------");
+        gRpcServer.shutdownNow();
+        gRpcChannel.shutdownNow();
+        gRpcServer.awaitTermination(1000, TimeUnit.MILLISECONDS);
+        gRpcChannel.awaitTermination(1000, TimeUnit.MILLISECONDS);
+    }
+
+    @Benchmark
+    public Object gRpcUnaryCall(Blackhole blackhole) throws InterruptedException {
+        PerfObserver observer = new PerfObserver(blackhole);
+
+        gRpcClient.unaryCall(REQUEST, observer);
+
+        observer.latch.await();
+
+        return observer;
+    }
+
+    @Benchmark
+    public Object gRpcServerStreamingCall(Blackhole blackhole) throws InterruptedException {
+        PerfObserver observer = new PerfObserver(blackhole);
+
+        gRpcClient.streamingFromServer(REQUEST, observer);
+
+        observer.latch.await();
+
+        return observer;
+    }
+
+    @Benchmark
+    public Object gRpcClientStreamingCall(Blackhole blackhole) throws InterruptedException {
+        PerfObserver observer = new PerfObserver(blackhole);
+        StreamObserver<Messages.SimpleRequest> sender = gRpcClient.streamingFromClient(observer);
+
+        for (Messages.SimpleRequest request : ARRAY_REQUEST) {
+            sender.onNext(request);
+        }
+
+        sender.onCompleted();
+
+        observer.latch.await();
+
+        return observer;
+    }
+
+    @Benchmark
+    public Object gRpcBothWaysStreamingCall(Blackhole blackhole) throws InterruptedException {
+        PerfObserver observer = new PerfObserver(blackhole) {
+            boolean done;
+            @Override
+            public void beforeStart(ClientCallStreamObserver<Messages.SimpleRequest> sender) {
+                sender.setOnReadyHandler(() -> {
+                    if (done) {
+                        return;
+                    }
+
+                    for (Messages.SimpleRequest request : ARRAY_REQUEST) {
+                        sender.onNext(request);
+                    }
+
+                    sender.onCompleted();
+                    done = true;
+                });
+                super.beforeStart(observer);
+            }
+        };
+
+        gRpcClient.streamingFromClient(observer);
+
+        observer.latch.await();
+
+        return observer;
+    }
+}
diff --git a/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/RxGRpcBenchmark.java b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/RxGRpcBenchmark.java
new file mode 100644
index 0000000..2c5c7d8
--- /dev/null
+++ b/benchmarks/src/test/java/com/salesforce/reactivegrpc/jmh/RxGRpcBenchmark.java
@@ -0,0 +1,137 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactivegrpc.jmh;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import com.salesforce.reactivegrpc.jmh.proto.Messages;
+import com.salesforce.reactivegrpc.jmh.proto.RxBenchmarkServiceGrpc;
+import io.grpc.ManagedChannel;
+import io.grpc.Server;
+import io.grpc.inprocess.InProcessChannelBuilder;
+import io.grpc.inprocess.InProcessServerBuilder;
+import io.reactivex.Flowable;
+import io.reactivex.Single;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+@BenchmarkMode(Mode.Throughput)
+@Warmup(iterations = 10)
+@Measurement(iterations = 10, time = 10, timeUnit = TimeUnit.SECONDS)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Fork(1)
+@State(Scope.Benchmark)
+public class RxGRpcBenchmark {
+
+    static final Single<Messages.SimpleRequest> MONO_REQUEST =
+        Single.just(Messages.SimpleRequest.getDefaultInstance());
+
+    static final Flowable<Messages.SimpleRequest> FLUX_REQUEST;
+
+    static {
+        Messages.SimpleRequest[] array = new Messages.SimpleRequest[100000];
+        Arrays.fill(array, Messages.SimpleRequest.getDefaultInstance());
+
+        FLUX_REQUEST = Flowable.fromArray(array);
+    }
+
+
+    Server         reactiveServer;
+    ManagedChannel reactiveChannel;
+
+    RxBenchmarkServiceGrpc.RxBenchmarkServiceStub reactiveClient;
+
+    @Setup
+    public void setup() throws IOException {
+        System.out.println("---------- SETUP ONCE -------------");
+        ScheduledExecutorService scheduledExecutorService =
+            Executors.newScheduledThreadPool(Runtime.getRuntime()
+                                                    .availableProcessors());
+        reactiveServer =
+            InProcessServerBuilder.forName("benchmark-reactiveServer")
+                                  .scheduledExecutorService(scheduledExecutorService)
+                                  .addService(new BenchmarkRxServerServiceImpl(100000))
+                                  .build()
+                                  .start();
+
+        reactiveChannel = InProcessChannelBuilder.forName("benchmark-reactiveServer")
+                                                 .build();
+        reactiveClient = RxBenchmarkServiceGrpc.newRxStub(reactiveChannel);
+    }
+
+    @TearDown
+    public void tearDown() throws InterruptedException {
+        System.out.println("---------- TEAR DOWN ONCE -------------");
+        reactiveServer.shutdownNow();
+        reactiveChannel.shutdownNow();
+        reactiveServer.awaitTermination(1000, TimeUnit.MILLISECONDS);
+        reactiveChannel.awaitTermination(1000, TimeUnit.MILLISECONDS);
+    }
+
+    @Benchmark
+    public Object reactiveUnaryCall(Blackhole blackhole) throws InterruptedException {
+        PerfSingleObserver actual = new PerfSingleObserver(blackhole);
+
+        reactiveClient.unaryCall(MONO_REQUEST)
+                      .subscribe(actual);
+        actual.latch.await();
+
+        return actual;
+    }
+
+    @Benchmark
+    public Object reactiveServerStreamingCall(Blackhole blackhole)
+        throws InterruptedException {
+        PerfSubscriber actual = new PerfSubscriber(blackhole);
+
+        reactiveClient.streamingFromServer(MONO_REQUEST)
+                      .subscribe(actual);
+
+        actual.latch.await();
+
+        return actual;
+    }
+
+    @Benchmark
+    public Object reactiveClientStreamingCall(Blackhole blackhole) throws InterruptedException {
+        PerfSingleObserver actual = new PerfSingleObserver(blackhole);
+
+        reactiveClient.streamingFromClient(FLUX_REQUEST)
+                      .subscribe(actual);
+
+        actual.latch.await();
+
+        return actual;
+    }
+
+
+    @Benchmark
+    public Object reactiveBothWaysStreamingCall(Blackhole blackhole) throws InterruptedException {
+        PerfSubscriber actual = new PerfSubscriber(blackhole);
+
+        reactiveClient.streamingBothWays(FLUX_REQUEST)
+                      .subscribe(actual);
+
+        actual.latch.await();
+
+        return actual;
+    }
+}
diff --git a/benchmarks/src/test/proto/grpc/testing/control.proto b/benchmarks/src/test/proto/grpc/testing/control.proto
new file mode 100755
index 0000000..69c841e
--- /dev/null
+++ b/benchmarks/src/test/proto/grpc/testing/control.proto
@@ -0,0 +1,265 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+// Copyright 2015-2016 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+syntax = "proto3";
+
+import "grpc/testing/payloads.proto";
+import "grpc/testing/stats.proto";
+
+package grpc.testing;
+
+option java_package = "com.salesforce.reactivegrpc.jmh.proto";
+option java_outer_classname = "Control";
+
+enum ClientType {
+  // Many languages support a basic distinction between using
+  // sync or async client, and this allows the specification
+  SYNC_CLIENT = 0;
+  ASYNC_CLIENT = 1;
+  OTHER_CLIENT = 2; // used for some language-specific variants
+}
+
+enum ServerType {
+  SYNC_SERVER = 0;
+  ASYNC_SERVER = 1;
+  ASYNC_GENERIC_SERVER = 2;
+  OTHER_SERVER = 3; // used for some language-specific variants
+}
+
+enum RpcType {
+  UNARY = 0;
+  STREAMING = 1;
+  STREAMING_FROM_CLIENT = 2;
+  STREAMING_FROM_SERVER = 3;
+  STREAMING_BOTH_WAYS = 4;
+}
+
+// Parameters of poisson process distribution, which is a good representation
+// of activity coming in from independent identical stationary sources.
+message PoissonParams {
+  // The rate of arrivals (a.k.a. lambda parameter of the exp distribution).
+  double offered_load = 1;
+}
+
+// Once an RPC finishes, immediately start a new one.
+// No configuration parameters needed.
+message ClosedLoopParams {}
+
+message LoadParams {
+  oneof load {
+    ClosedLoopParams closed_loop = 1;
+    PoissonParams poisson = 2;
+  };
+}
+
+// presence of SecurityParams implies use of TLS
+message SecurityParams {
+  bool use_test_ca = 1;
+  string server_host_override = 2;
+}
+
+message ChannelArg {
+  string name = 1;
+  oneof value {
+    string str_value = 2;
+    int32 int_value = 3;
+  }
+}
+
+message ClientConfig {
+  // List of targets to connect to. At least one target needs to be specified.
+  repeated string server_targets = 1;
+  ClientType client_type = 2;
+  SecurityParams security_params = 3;
+  // How many concurrent RPCs to start for each channel.
+  // For synchronous client, use a separate thread for each outstanding RPC.
+  int32 outstanding_rpcs_per_channel = 4;
+  // Number of independent client channels to create.
+  // i-th channel will connect to server_target[i % server_targets.size()]
+  int32 client_channels = 5;
+  // Only for async client. Number of threads to use to start/manage RPCs.
+  int32 async_client_threads = 7;
+  RpcType rpc_type = 8;
+  // The requested load for the entire client (aggregated over all the threads).
+  LoadParams load_params = 10;
+  PayloadConfig payload_config = 11;
+  HistogramParams histogram_params = 12;
+
+  // Specify the cores we should run the client on, if desired
+  repeated int32 core_list = 13;
+  int32 core_limit = 14;
+
+  // If we use an OTHER_CLIENT client_type, this string gives more detail
+  string other_client_api = 15;
+
+  repeated ChannelArg channel_args = 16;
+
+  // Number of messages on a stream before it gets finished/restarted
+  int32 messages_per_stream = 18;
+}
+
+message ClientStatus { ClientStats stats = 1; }
+
+// Request current stats
+message Mark {
+  // if true, the stats will be reset after taking their snapshot.
+  bool reset = 1;
+}
+
+message ClientArgs {
+  oneof argtype {
+    ClientConfig setup = 1;
+    Mark mark = 2;
+  }
+}
+
+message ServerConfig {
+  ServerType server_type = 1;
+  SecurityParams security_params = 2;
+  // Port on which to listen. Zero means pick unused port.
+  int32 port = 4;
+  // Only for async server. Number of threads used to serve the requests.
+  int32 async_server_threads = 7;
+  // Specify the number of cores to limit server to, if desired
+  int32 core_limit = 8;
+  // payload config, used in generic server.
+  // Note this must NOT be used in proto (non-generic) servers. For proto servers,
+  // 'response sizes' must be configured from the 'response_size' field of the
+  // 'SimpleRequest' objects in RPC requests.
+  PayloadConfig payload_config = 9;
+
+  // Specify the cores we should run the server on, if desired
+  repeated int32 core_list = 10;
+
+  // If we use an OTHER_SERVER client_type, this string gives more detail
+  string other_server_api = 11;
+
+  // c++-only options (for now) --------------------------------
+
+  // Buffer pool size (no buffer pool specified if unset)
+  int32 resource_quota_size = 1001;
+}
+
+message ServerArgs {
+  oneof argtype {
+    ServerConfig setup = 1;
+    Mark mark = 2;
+  }
+}
+
+message ServerStatus {
+  ServerStats stats = 1;
+  // the port bound by the server
+  int32 port = 2;
+  // Number of cores available to the server
+  int32 cores = 3;
+}
+
+message CoreRequest {
+}
+
+message CoreResponse {
+  // Number of cores available on the server
+  int32 cores = 1;
+}
+
+message Void {
+}
+
+// A single performance scenario: input to qps_json_driver
+message Scenario {
+  // Human readable name for this scenario
+  string name = 1;
+  // Client configuration
+  ClientConfig client_config = 2;
+  // Number of clients to start for the test
+  int32 num_clients = 3;
+  // Server configuration
+  ServerConfig server_config = 4;
+  // Number of servers to start for the test
+  int32 num_servers = 5;
+  // Warmup period, in seconds
+  int32 warmup_seconds = 6;
+  // Benchmark time, in seconds
+  int32 benchmark_seconds = 7;
+  // Number of workers to spawn locally (usually zero)
+  int32 spawn_local_worker_count = 8;
+}
+
+// A set of scenarios to be run with qps_json_driver
+message Scenarios {
+  repeated Scenario scenarios = 1;
+}
+
+// Basic summary that can be computed from ClientStats and ServerStats
+// once the scenario has finished.
+message ScenarioResultSummary
+{
+  // Total number of operations per second over all clients.
+  double qps = 1;
+  // QPS per one server core.
+  double qps_per_server_core = 2;
+  // server load based on system_time (0.85 => 85%)
+  double server_system_time = 3;
+  // server load based on user_time (0.85 => 85%)
+  double server_user_time = 4;
+  // client load based on system_time (0.85 => 85%)
+  double client_system_time = 5;
+  // client load based on user_time (0.85 => 85%)
+  double client_user_time = 6;
+
+  // X% latency percentiles (in nanoseconds)
+  double latency_50 = 7;
+  double latency_90 = 8;
+  double latency_95 = 9;
+  double latency_99 = 10;
+  double latency_999 = 11;
+
+  // server cpu usage percentage
+  double server_cpu_usage = 12;
+
+  // Number of requests that succeeded/failed
+  double successful_requests_per_second = 13;
+  double failed_requests_per_second = 14;
+
+  // Number of polls called inside completion queue per request
+  double client_polls_per_request = 15;
+  double server_polls_per_request = 16;
+}
+
+// Results of a single benchmark scenario.
+message ScenarioResult {
+  // Inputs used to run the scenario.
+  Scenario scenario = 1;
+  // Histograms from all clients merged into one histogram.
+  HistogramData latencies = 2;
+  // Client stats for each client
+  repeated ClientStats client_stats = 3;
+  // Server stats for each server
+  repeated ServerStats server_stats = 4;
+  // Number of cores available to each server
+  repeated int32 server_cores = 5;
+  // An after-the-fact computed summary
+  ScenarioResultSummary summary = 6;
+  // Information on success or failure of each worker
+  repeated bool client_success = 7;
+  repeated bool server_success = 8;
+  // Number of failed requests (one row per status code seen)
+  repeated RequestResultCount request_results = 9;
+}
diff --git a/benchmarks/src/test/proto/grpc/testing/messages.proto b/benchmarks/src/test/proto/grpc/testing/messages.proto
new file mode 100755
index 0000000..8005ae4
--- /dev/null
+++ b/benchmarks/src/test/proto/grpc/testing/messages.proto
@@ -0,0 +1,176 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+// Copyright 2015 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// Message definitions to be used by integration test service definitions.
+
+syntax = "proto3";
+
+package grpc.testing;
+
+option java_package = "com.salesforce.reactivegrpc.jmh.proto";
+option java_outer_classname = "Messages";
+
+// TODO(dgq): Go back to using well-known types once
+// https://github.com/grpc/grpc/issues/6980 has been fixed.
+// import "google/protobuf/wrappers.proto";
+message BoolValue {
+  // The bool value.
+  bool value = 1;
+}
+
+// DEPRECATED, don't use. To be removed shortly.
+// The type of payload that should be returned.
+enum PayloadType {
+  // Compressable text format.
+  COMPRESSABLE = 0;
+}
+
+// A block of data, to simply increase gRPC message size.
+message Payload {
+  // DEPRECATED, don't use. To be removed shortly.
+  // The type of data in body.
+  PayloadType type = 1;
+  // Primary contents of payload.
+  bytes body = 2;
+}
+
+// A protobuf representation for grpc status. This is used by test
+// clients to specify a status that the server should attempt to return.
+message EchoStatus {
+  int32 code = 1;
+  string message = 2;
+}
+
+// Unary request.
+message SimpleRequest {
+  // DEPRECATED, don't use. To be removed shortly.
+  // Desired payload type in the response from the server.
+  // If response_type is RANDOM, server randomly chooses one from other formats.
+  PayloadType response_type = 1;
+
+  // Desired payload size in the response from the server.
+  int32 response_size = 2;
+
+  // Optional input payload sent along with the request.
+  Payload payload = 3;
+
+  // Whether SimpleResponse should include username.
+  bool fill_username = 4;
+
+  // Whether SimpleResponse should include OAuth scope.
+  bool fill_oauth_scope = 5;
+
+  // Whether to request the server to compress the response. This field is
+  // "nullable" in order to interoperate seamlessly with clients not able to
+  // implement the full compression tests by introspecting the call to verify
+  // the response's compression status.
+  BoolValue response_compressed = 6;
+
+  // Whether server should return a given status
+  EchoStatus response_status = 7;
+
+  // Whether the server should expect this request to be compressed.
+  BoolValue expect_compressed = 8;
+}
+
+// Unary response, as configured by the request.
+message SimpleResponse {
+  // Payload to increase message size.
+  Payload payload = 1;
+  // The user the request came from, for verifying authentication was
+  // successful when the client expected it.
+  string username = 2;
+  // OAuth scope.
+  string oauth_scope = 3;
+}
+
+// Client-streaming request.
+message StreamingInputCallRequest {
+  // Optional input payload sent along with the request.
+  Payload payload = 1;
+
+  // Whether the server should expect this request to be compressed. This field
+  // is "nullable" in order to interoperate seamlessly with servers not able to
+  // implement the full compression tests by introspecting the call to verify
+  // the request's compression status.
+  BoolValue expect_compressed = 2;
+
+  // Not expecting any payload from the response.
+}
+
+// Client-streaming response.
+message StreamingInputCallResponse {
+  // Aggregated size of payloads received from the client.
+  int32 aggregated_payload_size = 1;
+}
+
+// Configuration for a particular response.
+message ResponseParameters {
+  // Desired payload sizes in responses from the server.
+  int32 size = 1;
+
+  // Desired interval between consecutive responses in the response stream in
+  // microseconds.
+  int32 interval_us = 2;
+
+  // Whether to request the server to compress the response. This field is
+  // "nullable" in order to interoperate seamlessly with clients not able to
+  // implement the full compression tests by introspecting the call to verify
+  // the response's compression status.
+  BoolValue compressed = 3;
+}
+
+// Server-streaming request.
+message StreamingOutputCallRequest {
+  // DEPRECATED, don't use. To be removed shortly.
+  // Desired payload type in the response from the server.
+  // If response_type is RANDOM, the payload from each response in the stream
+  // might be of different types. This is to simulate a mixed type of payload
+  // stream.
+  PayloadType response_type = 1;
+
+  // Configuration for each expected response message.
+  repeated ResponseParameters response_parameters = 2;
+
+  // Optional input payload sent along with the request.
+  Payload payload = 3;
+
+  // Whether server should return a given status
+  EchoStatus response_status = 7;
+}
+
+// Server-streaming response, as configured by the request and parameters.
+message StreamingOutputCallResponse {
+  // Payload to increase response size.
+  Payload payload = 1;
+}
+
+// For reconnect interop test only.
+// Client tells server what reconnection parameters it used.
+message ReconnectParams {
+  int32 max_reconnect_backoff_ms = 1;
+}
+
+// For reconnect interop test only.
+// Server tells client whether its reconnects are following the spec and the
+// reconnect backoffs it saw.
+message ReconnectInfo {
+  bool passed = 1;
+  repeated int32 backoff_ms = 2;
+}
diff --git a/benchmarks/src/test/proto/grpc/testing/payloads.proto b/benchmarks/src/test/proto/grpc/testing/payloads.proto
new file mode 100755
index 0000000..2cf3816
--- /dev/null
+++ b/benchmarks/src/test/proto/grpc/testing/payloads.proto
@@ -0,0 +1,48 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+// Copyright 2015 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+syntax = "proto3";
+
+package grpc.testing;
+
+option java_package = "com.salesforce.reactivegrpc.jmh.proto";
+option java_outer_classname = "Payloads";
+
+message ByteBufferParams {
+  int32 req_size = 1;
+  int32 resp_size = 2;
+}
+
+message SimpleProtoParams {
+  int32 req_size = 1;
+  int32 resp_size = 2;
+}
+
+message ComplexProtoParams {
+  // TODO (vpai): Fill this in once the details of complex, representative
+  //              protos are decided
+}
+
+message PayloadConfig {
+  oneof payload {
+    ByteBufferParams bytebuf_params = 1;
+    SimpleProtoParams simple_params = 2;
+    ComplexProtoParams complex_params = 3;
+  }
+}
diff --git a/benchmarks/src/test/proto/grpc/testing/services.proto b/benchmarks/src/test/proto/grpc/testing/services.proto
new file mode 100755
index 0000000..dcfee9f
--- /dev/null
+++ b/benchmarks/src/test/proto/grpc/testing/services.proto
@@ -0,0 +1,82 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+// Copyright 2015-2016 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// An integration test service that covers all the method signature permutations
+// of unary/streaming requests/responses.
+syntax = "proto3";
+
+import "grpc/testing/messages.proto";
+import "grpc/testing/control.proto";
+
+package grpc.testing;
+
+option java_package = "com.salesforce.reactivegrpc.jmh.proto";
+option java_outer_classname = "Services";
+
+service BenchmarkService {
+  // One request followed by one response.
+  // The server returns the client payload as-is.
+  rpc UnaryCall(SimpleRequest) returns (SimpleResponse);
+
+  // Repeated sequence of one request followed by one response.
+  // Should be called streaming ping-pong
+  // The server returns the client payload as-is on each response
+  rpc StreamingCall(stream SimpleRequest) returns (stream SimpleResponse);
+
+  // Single-sided unbounded streaming from client to server
+  // The server returns the client payload as-is once the client does WritesDone
+  rpc StreamingFromClient(stream SimpleRequest) returns (SimpleResponse);
+
+  // Single-sided unbounded streaming from server to client
+  // The server repeatedly returns the client payload as-is
+  rpc StreamingFromServer(SimpleRequest) returns (stream SimpleResponse);
+
+  // Two-sided unbounded streaming between server to client
+  // Both sides send the content of their own choice to the other
+  rpc StreamingBothWays(stream SimpleRequest) returns (stream SimpleResponse);
+}
+
+service WorkerService {
+  // Start server with specified workload.
+  // First request sent specifies the ServerConfig followed by ServerStatus
+  // response. After that, a "Mark" can be sent anytime to request the latest
+  // stats. Closing the stream will initiate shutdown of the test server
+  // and once the shutdown has finished, the OK status is sent to terminate
+  // this RPC.
+  rpc RunServer(stream ServerArgs) returns (stream ServerStatus);
+
+  // Start client with specified workload.
+  // First request sent specifies the ClientConfig followed by ClientStatus
+  // response. After that, a "Mark" can be sent anytime to request the latest
+  // stats. Closing the stream will initiate shutdown of the test client
+  // and once the shutdown has finished, the OK status is sent to terminate
+  // this RPC.
+  rpc RunClient(stream ClientArgs) returns (stream ClientStatus);
+
+  // Just return the core count - unary call
+  rpc CoreCount(CoreRequest) returns (CoreResponse);
+
+  // Quit this worker
+  rpc QuitWorker(Void) returns (Void);
+}
+
+service ReportQpsScenarioService {
+  // Report results of a QPS test benchmark scenario.
+  rpc ReportScenario(ScenarioResult) returns (Void);
+}
diff --git a/benchmarks/src/test/proto/grpc/testing/stats.proto b/benchmarks/src/test/proto/grpc/testing/stats.proto
new file mode 100755
index 0000000..b72d4a8
--- /dev/null
+++ b/benchmarks/src/test/proto/grpc/testing/stats.proto
@@ -0,0 +1,83 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+// Copyright 2015 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+syntax = "proto3";
+
+package grpc.testing;
+
+option java_package = "com.salesforce.reactivegrpc.jmh.proto";
+option java_outer_classname = "Stats";
+
+message ServerStats {
+  // wall clock time change in seconds since last reset
+  double time_elapsed = 1;
+
+  // change in user time (in seconds) used by the server since last reset
+  double time_user = 2;
+
+  // change in server time (in seconds) used by the server process and all
+  // threads since last reset
+  double time_system = 3;
+
+  // change in total cpu time of the server (data from proc/stat)
+  uint64 total_cpu_time = 4;
+
+  // change in idle time of the server (data from proc/stat)
+  uint64 idle_cpu_time = 5;
+
+  // Number of polls called inside completion queue
+  uint64 cq_poll_count = 6;
+}
+
+// Histogram params based on grpc/support/histogram.c
+message HistogramParams {
+  double resolution = 1;   // first bucket is [0, 1 + resolution)
+  double max_possible = 2; // use enough buckets to allow this value
+}
+
+// Histogram data based on grpc/support/histogram.c
+message HistogramData {
+  repeated uint32 bucket = 1;
+  double min_seen = 2;
+  double max_seen = 3;
+  double sum = 4;
+  double sum_of_squares = 5;
+  double count = 6;
+}
+
+message RequestResultCount {
+  int32 status_code = 1;
+  int64 count = 2;
+}
+
+message ClientStats {
+  // Latency histogram. Data points are in nanoseconds.
+  HistogramData latencies = 1;
+
+  // See ServerStats for details.
+  double time_elapsed = 2;
+  double time_user = 3;
+  double time_system = 4;
+
+  // Number of failed requests (one row per status code seen)
+  repeated RequestResultCount request_results = 5;
+
+  // Number of polls called inside completion queue
+  uint64 cq_poll_count = 6;
+}
diff --git a/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpc/common/BackpressureChunkingTest.java b/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpc/common/BackpressureChunkingTest.java
index 9f0242f..ebbcadf 100644
--- a/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpc/common/BackpressureChunkingTest.java
+++ b/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpc/common/BackpressureChunkingTest.java
@@ -17,15 +17,17 @@ import io.reactivex.subscribers.TestSubscriber;
 import org.junit.Test;
 import org.reactivestreams.Publisher;
 
+import static com.salesforce.reactivegrpc.common.AbstractStreamObserverAndPublisher.DEFAULT_CHUNK_SIZE;
+import static com.salesforce.reactivegrpc.common.AbstractStreamObserverAndPublisher.TWO_THIRDS_OF_DEFAULT_CHUNK_SIZE;
 import static org.assertj.core.api.Assertions.assertThat;
 
 public class BackpressureChunkingTest {
     @Test
     public void chunkOperatorCorrectlyChunksInfiniteRequest() {
-        int chunkSize = AbstractStreamObserverAndPublisher.DEFAULT_CHUNK_SIZE;
+        int chunkSize = DEFAULT_CHUNK_SIZE;
 
-        int halfChunk = chunkSize / 2;
-        int num = chunkSize * 2 + halfChunk;
+        int partOfChunk = TWO_THIRDS_OF_DEFAULT_CHUNK_SIZE;
+        int num = chunkSize * 2;
 
         AbstractStreamObserverAndPublisher<Long> source =
                 new TestStreamObserverAndPublisherWithFusion<Long>(new ConcurrentLinkedQueue<Long>(), null);
@@ -38,16 +40,16 @@ public class BackpressureChunkingTest {
         testSubscriber.awaitTerminalEvent();
         testSubscriber.assertComplete();
 
-        assertThat(observer.requestsQueue).containsExactly(chunkSize, chunkSize, chunkSize);
+        assertThat(observer.requestsQueue).containsExactly(chunkSize, partOfChunk, partOfChunk, partOfChunk);
         assertThat(source.outputFused).isFalse();
     }
 
     @Test
     public void chunkOperatorCorrectlyChunksFiniteRequest() {
-        int chunkSize = AbstractStreamObserverAndPublisher.DEFAULT_CHUNK_SIZE;
+        int chunkSize = DEFAULT_CHUNK_SIZE;
 
-        int halfChunk = chunkSize / 2;
-        int num = chunkSize * 2 + halfChunk;
+        int partOfChunk = TWO_THIRDS_OF_DEFAULT_CHUNK_SIZE;
+        int num = chunkSize * 2;
 
         AbstractStreamObserverAndPublisher<Long> source =
                 new TestStreamObserverAndPublisherWithFusion<Long>(new ConcurrentLinkedQueue<Long>(), null);
@@ -59,16 +61,16 @@ public class BackpressureChunkingTest {
         testSubscriber.awaitTerminalEvent();
         testSubscriber.assertComplete();
 
-        assertThat(observer.requestsQueue).containsExactly(chunkSize, chunkSize, chunkSize);
+        assertThat(observer.requestsQueue).containsExactly(chunkSize, partOfChunk, partOfChunk, partOfChunk);
         assertThat(source.outputFused).isFalse();
     }
 
     @Test
     public void chunkOperatorCorrectlyChunksInfiniteRequestFusion() {
-        int chunkSize = AbstractStreamObserverAndPublisher.DEFAULT_CHUNK_SIZE;
+        int chunkSize = DEFAULT_CHUNK_SIZE;
 
-        int halfChunk = chunkSize / 2;
-        int num = chunkSize * 2 + halfChunk;
+        int partOfChunk = TWO_THIRDS_OF_DEFAULT_CHUNK_SIZE;
+        int num = chunkSize * 2;
 
         AbstractStreamObserverAndPublisher<Long> source =
                 new TestStreamObserverAndPublisherWithFusion<Long>(new ConcurrentLinkedQueue<Long>(), null);
@@ -82,16 +84,16 @@ public class BackpressureChunkingTest {
         testSubscriber.awaitTerminalEvent();
         testSubscriber.assertComplete();
 
-        assertThat(observer.requestsQueue).containsExactly(chunkSize, chunkSize, chunkSize);
+        assertThat(observer.requestsQueue).containsExactly(chunkSize, partOfChunk, partOfChunk, partOfChunk);
         assertThat(source.outputFused).isTrue();
     }
 
     @Test
     public void chunkOperatorCorrectlyChunksFiniteRequestFusion() {
-        int chunkSize = AbstractStreamObserverAndPublisher.DEFAULT_CHUNK_SIZE;
+        int chunkSize = DEFAULT_CHUNK_SIZE;
 
-        int halfChunk = chunkSize / 2;
-        int num = chunkSize * 2 + halfChunk;
+        int partOfChunk = TWO_THIRDS_OF_DEFAULT_CHUNK_SIZE;
+        int num = chunkSize * 2;
 
         AbstractStreamObserverAndPublisher<Long> source =
                 new TestStreamObserverAndPublisherWithFusion<Long>(new ConcurrentLinkedQueue<Long>(), null);
@@ -104,7 +106,7 @@ public class BackpressureChunkingTest {
         testSubscriber.awaitTerminalEvent();
         testSubscriber.assertComplete();
 
-        assertThat(observer.requestsQueue).containsExactly(chunkSize, chunkSize, chunkSize);
+        assertThat(observer.requestsQueue).containsExactly(chunkSize, partOfChunk, partOfChunk, partOfChunk);
         assertThat(source.outputFused).isTrue();
     }
 
@@ -113,7 +115,7 @@ public class BackpressureChunkingTest {
      */
     @Test
     public void chunkOperatorWorksWithConcatMap() {
-        int chunkSize = AbstractStreamObserverAndPublisher.DEFAULT_CHUNK_SIZE;
+        int chunkSize = DEFAULT_CHUNK_SIZE;
 
         AbstractStreamObserverAndPublisher<Long> source =
                 new AbstractStreamObserverAndPublisher<Long>(new ConcurrentLinkedQueue<Long>(), null){};
@@ -131,6 +133,6 @@ public class BackpressureChunkingTest {
         testSubscriber.awaitTerminalEvent();
         testSubscriber.assertNoErrors();
 
-        assertThat(observer.requestsQueue).containsExactly(chunkSize, chunkSize);
+        assertThat(observer.requestsQueue).containsExactly(chunkSize);
     }
 }
diff --git a/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpc/common/StreamObserverAndPublisherTest.java b/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpc/common/StreamObserverAndPublisherTest.java
index a8b784f..ba7738a 100644
--- a/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpc/common/StreamObserverAndPublisherTest.java
+++ b/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpc/common/StreamObserverAndPublisherTest.java
@@ -28,9 +28,13 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
+import static com.salesforce.reactivegrpc.common.AbstractStreamObserverAndPublisher.DEFAULT_CHUNK_SIZE;
+
 @RunWith(Parameterized.class)
 public class StreamObserverAndPublisherTest {
 
+    static final int PART_OF_CHUNK = DEFAULT_CHUNK_SIZE * 2 / 3;
+
     @Parameterized.Parameters
     public static Object[][] data() {
         return new Object[1][0];
@@ -82,10 +86,14 @@ public class StreamObserverAndPublisherTest {
         testSubscriber.assertValueCount(countPerThread);
 
         Assertions.assertThat(processor.outputFused).isFalse();
-        Assertions.assertThat(observer.requestsQueue.size()).isBetween(countPerThread / 16, countPerThread / 16 + 1);
-        Integer i;
+        Assertions.assertThat(observer.requestsQueue.size()).isBetween((countPerThread - DEFAULT_CHUNK_SIZE) / PART_OF_CHUNK + 1, (countPerThread - DEFAULT_CHUNK_SIZE) / PART_OF_CHUNK + 2);
+
+        Integer i = observer.requestsQueue.poll();
+
+        Assertions.assertThat(i).isEqualTo(DEFAULT_CHUNK_SIZE);
+
         while ((i = observer.requestsQueue.poll()) != null) {
-            Assertions.assertThat(i).isEqualTo(16);
+            Assertions.assertThat(i).isEqualTo(PART_OF_CHUNK);
         }
     }
 
@@ -116,10 +124,14 @@ public class StreamObserverAndPublisherTest {
         testSubscriber.assertValueCount(countPerThread);
 
         Assertions.assertThat(processor.outputFused).isTrue();
-        Assertions.assertThat(observer.requestsQueue.size()).isBetween(countPerThread / 16, countPerThread / 16 + 1);
-        Integer i;
+        Assertions.assertThat(observer.requestsQueue.size()).isBetween((countPerThread - DEFAULT_CHUNK_SIZE) / PART_OF_CHUNK + 1, (countPerThread - DEFAULT_CHUNK_SIZE) / PART_OF_CHUNK + 2);
+
+        Integer i = observer.requestsQueue.poll();
+
+        Assertions.assertThat(i).isEqualTo(DEFAULT_CHUNK_SIZE);
+
         while ((i = observer.requestsQueue.poll()) != null) {
-            Assertions.assertThat(i).isEqualTo(16);
+            Assertions.assertThat(i).isEqualTo(PART_OF_CHUNK);
         }
     }
 
@@ -221,7 +233,7 @@ public class StreamObserverAndPublisherTest {
 
             Assertions.assertThat(upstream.requestsQueue)
                       .hasSize(1)
-                      .containsOnly(16);
+                      .containsOnly(DEFAULT_CHUNK_SIZE);
         }
     }
 
diff --git a/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/BenchmarkGRpcServerServiceImpl.java b/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/BenchmarkGRpcServerServiceImpl.java
new file mode 100644
index 0000000..331db22
--- /dev/null
+++ b/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/BenchmarkGRpcServerServiceImpl.java
@@ -0,0 +1,111 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactorgrpc.jmh;
+
+import java.util.Arrays;
+import java.util.List;
+
+import com.salesforce.reactorgrpc.jmh.proto.BenchmarkServiceGrpc;
+import com.salesforce.reactorgrpc.jmh.proto.Messages;
+import io.grpc.stub.ServerCallStreamObserver;
+import io.grpc.stub.StreamObserver;
+
+public class BenchmarkGRpcServerServiceImpl
+    extends BenchmarkServiceGrpc.BenchmarkServiceImplBase {
+
+    final Messages.SimpleResponse       response;
+    final List<Messages.SimpleResponse> responses;
+
+    public BenchmarkGRpcServerServiceImpl(int times) {
+        Messages.SimpleResponse[] array = new Messages.SimpleResponse[times];
+        Arrays.fill(array, Messages.SimpleResponse.getDefaultInstance());
+
+        this.responses = Arrays.asList(array);
+        this.response = Messages.SimpleResponse.getDefaultInstance();
+    }
+
+    @Override
+    public void unaryCall(Messages.SimpleRequest request,
+        StreamObserver<Messages.SimpleResponse> responseObserver) {
+        responseObserver.onNext(response);
+        responseObserver.onCompleted();
+    }
+
+    @Override
+    public void streamingFromServer(Messages.SimpleRequest request, StreamObserver<Messages.SimpleResponse> responseObserver) {
+        final ServerCallStreamObserver<Messages.SimpleResponse> callStreamObserver = (ServerCallStreamObserver<Messages.SimpleResponse>) responseObserver;
+
+        for (Messages.SimpleResponse response : responses) {
+            if (callStreamObserver.isCancelled()) {
+                return;
+            }
+
+            callStreamObserver.onNext(response);
+        }
+
+        if (!callStreamObserver.isCancelled()) {
+            callStreamObserver.onCompleted();
+        }
+    }
+
+    @Override
+    public StreamObserver<Messages.SimpleRequest> streamingFromClient(StreamObserver<Messages.SimpleResponse> responseObserver) {
+        return new StreamObserver<Messages.SimpleRequest>() {
+            @Override
+            public void onNext(Messages.SimpleRequest value) {
+
+            }
+
+            @Override
+            public void onError(Throwable t) {
+                responseObserver.onError(t);
+            }
+
+            @Override
+            public void onCompleted() {
+                responseObserver.onNext(response);
+                responseObserver.onCompleted();
+            }
+        };
+    }
+
+    @Override
+    public StreamObserver<Messages.SimpleRequest> streamingBothWays(StreamObserver<Messages.SimpleResponse> responseObserver) {
+        final ServerCallStreamObserver<Messages.SimpleResponse> callStreamObserver = (ServerCallStreamObserver<Messages.SimpleResponse>) responseObserver;
+
+        callStreamObserver.setOnReadyHandler(() -> {
+            for (Messages.SimpleResponse response : responses) {
+                if (callStreamObserver.isCancelled()) {
+                    return;
+                }
+
+                callStreamObserver.onNext(response);
+            }
+
+            if (!callStreamObserver.isCancelled()) {
+                callStreamObserver.onCompleted();
+            }
+        });
+
+        return new StreamObserver<Messages.SimpleRequest>() {
+            @Override
+            public void onNext(Messages.SimpleRequest value) {
+
+            }
+
+            @Override
+            public void onError(Throwable t) {
+
+            }
+
+            @Override
+            public void onCompleted() {
+
+            }
+        };
+    }
+}
diff --git a/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/BenchmarkReactorServerServiceImpl.java b/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/BenchmarkReactorServerServiceImpl.java
new file mode 100644
index 0000000..c212451
--- /dev/null
+++ b/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/BenchmarkReactorServerServiceImpl.java
@@ -0,0 +1,50 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactorgrpc.jmh;
+
+import java.util.Arrays;
+
+import com.salesforce.reactorgrpc.jmh.proto.Messages;
+import com.salesforce.reactorgrpc.jmh.proto.ReactorBenchmarkServiceGrpc;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class BenchmarkReactorServerServiceImpl extends
+                                        ReactorBenchmarkServiceGrpc.BenchmarkServiceImplBase {
+
+    final Mono<Messages.SimpleResponse> responseMono;
+    final Flux<Messages.SimpleResponse> responseFlux;
+
+    public BenchmarkReactorServerServiceImpl(int times) {
+        Messages.SimpleResponse[] array = new Messages.SimpleResponse[times];
+        Arrays.fill(array, Messages.SimpleResponse.getDefaultInstance());
+
+        this.responseFlux = Flux.fromArray(array);
+        this.responseMono = Mono.just(Messages.SimpleResponse.getDefaultInstance());
+    }
+
+    @Override
+    public Mono<Messages.SimpleResponse> unaryCall(Mono<Messages.SimpleRequest> request) {
+        return request.then(responseMono);
+    }
+
+    @Override
+    public Flux<Messages.SimpleResponse> streamingFromServer(Mono<Messages.SimpleRequest> request) {
+        return request.thenMany(responseFlux);
+    }
+
+    @Override
+    public Mono<Messages.SimpleResponse> streamingFromClient(Flux<Messages.SimpleRequest> request) {
+        return request.then(responseMono);
+    }
+
+    @Override
+    public Flux<Messages.SimpleResponse> streamingBothWays(Flux<Messages.SimpleRequest> request) {
+        request.subscribe();
+        return responseFlux;
+    }
+}
diff --git a/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/JMHRunner.java b/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/JMHRunner.java
new file mode 100644
index 0000000..c73a965
--- /dev/null
+++ b/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/JMHRunner.java
@@ -0,0 +1,18 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactorgrpc.jmh;
+
+import java.io.IOException;
+
+import org.openjdk.jmh.runner.RunnerException;
+
+public class JMHRunner {
+
+    public static void main(String... args) throws IOException, RunnerException {
+        org.openjdk.jmh.Main.main(args);
+    }
+}
diff --git a/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/PerfObserver.java b/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/PerfObserver.java
new file mode 100644
index 0000000..02e48f0
--- /dev/null
+++ b/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/PerfObserver.java
@@ -0,0 +1,49 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactorgrpc.jmh;
+
+import java.util.concurrent.CountDownLatch;
+
+import com.salesforce.reactorgrpc.jmh.proto.Messages;
+import io.grpc.stub.ClientCallStreamObserver;
+import io.grpc.stub.ClientResponseObserver;
+import org.openjdk.jmh.infra.Blackhole;
+
+public class PerfObserver implements ClientResponseObserver<Messages.SimpleRequest, Messages.SimpleResponse> {
+
+    final Blackhole      blackhole;
+    final CountDownLatch latch;
+
+    ClientCallStreamObserver observer;
+
+    public PerfObserver(Blackhole blackhole) {
+        this.blackhole = blackhole;
+        this.latch = new CountDownLatch(1);
+    }
+
+    @Override
+    public void beforeStart(ClientCallStreamObserver<Messages.SimpleRequest> observer) {
+        this.observer = observer;
+    }
+
+    @Override
+    public void onNext(Messages.SimpleResponse response) {
+        blackhole.consume(response);
+    }
+
+    @Override
+    public void onError(Throwable throwable) {
+        throwable.printStackTrace();
+        blackhole.consume(throwable);
+        latch.countDown();
+    }
+
+    @Override
+    public void onCompleted() {
+        latch.countDown();
+    }
+}
diff --git a/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/PerfSubscriber.java b/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/PerfSubscriber.java
new file mode 100644
index 0000000..5552bc5
--- /dev/null
+++ b/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/PerfSubscriber.java
@@ -0,0 +1,51 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactorgrpc.jmh;
+
+import java.util.concurrent.CountDownLatch;
+
+import org.openjdk.jmh.infra.Blackhole;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+public final class PerfSubscriber implements Subscriber<Object> {
+
+    final Blackhole bh;
+    final CountDownLatch latch;
+
+    Subscription subscription;
+
+    public PerfSubscriber(Blackhole bh) {
+        this.bh = bh;
+        this.latch = new CountDownLatch(1);
+    }
+
+    @Override
+    public void onSubscribe(Subscription subscription) {
+        this.subscription = subscription;
+        subscription.request(1);
+    }
+
+    @Override
+    public void onNext(Object item) {
+        bh.consume(item);
+        subscription.request(1);
+    }
+
+    @Override
+    public void onError(Throwable throwable) {
+        throwable.printStackTrace();
+        bh.consume(throwable);
+        latch.countDown();
+    }
+
+    @Override
+    public void onComplete() {
+        latch.countDown();
+    }
+
+}
diff --git a/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/ReferenceGRpcBenchmark.java b/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/ReferenceGRpcBenchmark.java
new file mode 100644
index 0000000..ffb1818
--- /dev/null
+++ b/reactor/reactor-grpc-jmh/src/test/java/com/salesforce/reactorgrpc/jmh/ReferenceGRpcBenchmark.java
@@ -0,0 +1,148 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactorgrpc.jmh;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import com.salesforce.reactorgrpc.jmh.proto.BenchmarkServiceGrpc;
+import com.salesforce.reactorgrpc.jmh.proto.Messages;
+import io.grpc.ManagedChannel;
+import io.grpc.Server;
+import io.grpc.inprocess.InProcessChannelBuilder;
+import io.grpc.inprocess.InProcessServerBuilder;
+import io.grpc.stub.ClientCallStreamObserver;
+import io.grpc.stub.StreamObserver;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+@BenchmarkMode(Mode.Throughput)
+@Warmup(iterations = 10)
+@Measurement(iterations = 10, time = 10, timeUnit = TimeUnit.SECONDS)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Fork(1)
+@State(Scope.Benchmark)
+public class ReferenceGRpcBenchmark {
+
+    static final Messages.SimpleRequest REQUEST = Messages.SimpleRequest.getDefaultInstance();
+    static final Messages.SimpleRequest[] ARRAY_REQUEST = new Messages.SimpleRequest[100000];
+
+    static {
+        Arrays.fill(ARRAY_REQUEST, Messages.SimpleRequest.getDefaultInstance());
+    }
+
+    Server         gRpcServer;
+    ManagedChannel gRpcChannel;
+
+    BenchmarkServiceGrpc.BenchmarkServiceStub gRpcClient;
+
+    @Setup
+    public void setup() throws IOException {
+        System.out.println("---------- SETUP ONCE -------------");
+        ScheduledExecutorService scheduledExecutorService =
+            Executors.newScheduledThreadPool(Runtime.getRuntime()
+                                                    .availableProcessors());
+        gRpcServer =
+            InProcessServerBuilder.forName("benchmark-gRpcServer")
+                                  .scheduledExecutorService(scheduledExecutorService)
+                                  .addService(new BenchmarkGRpcServerServiceImpl(100000))
+                                  .build()
+                                  .start();
+        gRpcChannel = InProcessChannelBuilder.forName("benchmark-gRpcServer")
+                                             .build();
+        gRpcClient = BenchmarkServiceGrpc.newStub(gRpcChannel);
+    }
+
+    @TearDown
+    public void tearDown() throws InterruptedException {
+        System.out.println("---------- TEAR DOWN ONCE -------------");
+        gRpcServer.shutdownNow();
+        gRpcChannel.shutdownNow();
+        gRpcServer.awaitTermination(1000, TimeUnit.MILLISECONDS);
+        gRpcChannel.awaitTermination(1000, TimeUnit.MILLISECONDS);
+    }
+
+    @Benchmark
+    public Object gRpcUnaryCall(Blackhole blackhole) throws InterruptedException {
+        PerfObserver observer = new PerfObserver(blackhole);
+
+        gRpcClient.unaryCall(REQUEST, observer);
+
+        observer.latch.await();
+
+        return observer;
+    }
+
+    @Benchmark
+    public Object gRpcServerStreamingCall(Blackhole blackhole) throws InterruptedException {
+        PerfObserver observer = new PerfObserver(blackhole);
+
+        gRpcClient.streamingFromServer(REQUEST, observer);
+
+        observer.latch.await();
+
+        return observer;
+    }
+
+    @Benchmark
+    public Object gRpcClientStreamingCall(Blackhole blackhole) throws InterruptedException {
+        PerfObserver observer = new PerfObserver(blackhole);
+        StreamObserver<Messages.SimpleRequest> sender = gRpcClient.streamingFromClient(observer);
+
+        for (Messages.SimpleRequest request : ARRAY_REQUEST) {
+            sender.onNext(request);
+        }
+
+        sender.onCompleted();
+
+        observer.latch.await();
+
+        return observer;
+    }
+
+    @Benchmark
+    public Object gRpcBothWaysStreamingCall(Blackhole blackhole) throws InterruptedException {
+        PerfObserver observer = new PerfObserver(blackhole) {
+            boolean done;
+            @Override
+            public void beforeStart(ClientCallStreamObserver<Messages.SimpleRequest> sender) {
+                sender.setOnReadyHandler(() -> {
+                    if (done) {
+                        return;
+                    }
+
+                    for (Messages.SimpleRequest request : ARRAY_REQUEST) {
+                        sender.onNext(request);
+                    }
+
+                    sender.onCompleted();
+                    done = true;
+                });
+                super.beforeStart(observer);
+            }
+        };
+
+        gRpcClient.streamingFromClient(observer);
+
+        observer.latch.await();
+
+        return observer;
+    }
+}
diff --git a/reactor/reactor-grpc-stub/src/test/java/com/salesforce/reactorgrpc/stub/ReactorClientStreamObserverAndPublisherTest.java b/reactor/reactor-grpc-stub/src/test/java/com/salesforce/reactorgrpc/stub/ReactorClientStreamObserverAndPublisherTest.java
index 2b49adc..826cbde 100644
--- a/reactor/reactor-grpc-stub/src/test/java/com/salesforce/reactorgrpc/stub/ReactorClientStreamObserverAndPublisherTest.java
+++ b/reactor/reactor-grpc-stub/src/test/java/com/salesforce/reactorgrpc/stub/ReactorClientStreamObserverAndPublisherTest.java
@@ -6,13 +6,8 @@
 
 package com.salesforce.reactorgrpc.stub;
 
-import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.atomic.AtomicLong;
-import javax.annotation.Nullable;
+import java.util.concurrent.ForkJoinPool;
 
-import io.grpc.stub.ClientCallStreamObserver;
-import io.grpc.stub.StreamObserver;
 import org.assertj.core.api.Assertions;
 import org.junit.Test;
 import reactor.core.Fuseable;
@@ -20,156 +15,40 @@ import reactor.core.publisher.Flux;
 import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 
+
 public class ReactorClientStreamObserverAndPublisherTest {
 
+    private static final int DEFAULT_CHUNK_SIZE = 512;
+    private static final int PART_OF_CHUNK = DEFAULT_CHUNK_SIZE * 2 / 3;
+
     @Test
     public void multiThreadedProducerTest() {
-        ReactorClientStreamObserverAndPublisher<Long> processor =
+        ReactorClientStreamObserverAndPublisher<Integer> processor =
                 new ReactorClientStreamObserverAndPublisher<>(null);
         int countPerThread = 100000;
-        TestCallStreamObserver observer = new TestCallStreamObserver(processor, countPerThread);
+        TestCallStreamObserverProducer observer =
+            new TestCallStreamObserverProducer(ForkJoinPool.commonPool(), processor, countPerThread);
         processor.beforeStart(observer);
         StepVerifier.create(Flux.from(processor)
                                 .subscribeOn(Schedulers.single()))
                     .expectNextCount(countPerThread)
                     .verifyComplete();
 
-        Assertions.assertThat(observer.requestsQueue.size()).isEqualTo(countPerThread / 16 + 1);
+        Assertions.assertThat(observer.requestsQueue.size()).isBetween((countPerThread - DEFAULT_CHUNK_SIZE) / PART_OF_CHUNK + 1, (countPerThread - DEFAULT_CHUNK_SIZE) / PART_OF_CHUNK + 3);
     }
 
     @Test
     public void producerFusedTest() {
-        ReactorClientStreamObserverAndPublisher<Long> processor =
+        ReactorClientStreamObserverAndPublisher<Integer> processor =
                 new ReactorClientStreamObserverAndPublisher<>(null);
         int countPerThread = 100000;
-        TestCallStreamObserver observer = new TestCallStreamObserver(processor, countPerThread);
+        TestCallStreamObserverProducer observer = new TestCallStreamObserverProducer(ForkJoinPool.commonPool(), processor, countPerThread);
         processor.beforeStart(observer);
         StepVerifier.create(Flux.from(processor))
                     .expectFusion(Fuseable.ANY, Fuseable.ASYNC)
                     .expectNextCount(countPerThread)
                     .verifyComplete();
 
-        Assertions.assertThat(observer.requestsQueue.size()).isEqualTo(countPerThread / 16 + 1);
+        Assertions.assertThat(observer.requestsQueue.size()).isBetween((countPerThread - DEFAULT_CHUNK_SIZE) / PART_OF_CHUNK + 1, (countPerThread - DEFAULT_CHUNK_SIZE) / PART_OF_CHUNK + 3);
     }
-
-    static class TestCallStreamObserver extends ClientCallStreamObserver<Long> {
-
-
-        final Queue<Integer> requestsQueue = new ConcurrentLinkedQueue<>();
-        final AtomicLong requested = new AtomicLong();
-
-        final StreamObserver<Long> subscriber;
-
-        final long max;
-
-
-        volatile boolean cancelled;
-
-        long index = 0;
-
-        TestCallStreamObserver(StreamObserver<Long> subscriber, long max) {
-            this.subscriber = subscriber;
-            this.max = max;
-        }
-
-        @Override
-        public void cancel(@Nullable String message, @Nullable Throwable cause) {
-            cancelled = true;
-        }
-
-        @Override
-        public boolean isReady() {
-            return true;
-        }
-
-        @Override
-        public void onNext(Long value) {
-
-        }
-
-        @Override
-        public void onError(Throwable t) {
-
-        }
-
-        @Override
-        public void onCompleted() {
-
-        }
-
-        @Override
-        public void setOnReadyHandler(Runnable onReadyHandler) {
-
-        }
-
-        @Override
-        public void disableAutoInboundFlowControl() {
-
-        }
-
-        @Override
-        public void request(int count) {
-            requestsQueue.add(count);
-
-            long n = count;
-
-            long initialRequested;
-
-            do {
-                initialRequested = requested.get();
-
-                if (initialRequested == Long.MAX_VALUE) {
-                    return;
-                }
-
-                n = initialRequested + n;
-
-                if (n <= 0) {
-                    n = Long.MAX_VALUE;
-                }
-
-            } while (!requested.compareAndSet(initialRequested, n));
-
-            if (initialRequested > 0) {
-                return;
-            }
-
-            new Thread(() -> {
-                int sent = 0;
-
-                while (true) {
-
-                    for (; sent < requested.get() && index < max; sent++, index++) {
-
-                        if (cancelled) {
-                            return;
-                        }
-
-                        subscriber.onNext(index);
-                    }
-
-                    if (cancelled) {
-                        return;
-                    }
-
-                    if (index == max) {
-                        subscriber.onCompleted();
-                        return;
-                    }
-
-                    if (requested.addAndGet(-sent) == 0) {
-                        return;
-                    }
-
-                    sent = 0;
-                }
-            }).start();
-        }
-
-        @Override
-        public void setMessageCompression(boolean enable) {
-
-        }
-    }
-
 }
\ No newline at end of file
diff --git a/reactor/reactor-grpc-stub/src/test/java/com/salesforce/reactorgrpc/stub/TestCallStreamObserverProducer.java b/reactor/reactor-grpc-stub/src/test/java/com/salesforce/reactorgrpc/stub/TestCallStreamObserverProducer.java
new file mode 100644
index 0000000..c1647c7
--- /dev/null
+++ b/reactor/reactor-grpc-stub/src/test/java/com/salesforce/reactorgrpc/stub/TestCallStreamObserverProducer.java
@@ -0,0 +1,147 @@
+/*
+ *  Copyright (c) 2019, Salesforce.com, Inc.
+ *  All rights reserved.
+ *  Licensed under the BSD 3-Clause license.
+ *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
+ */
+package com.salesforce.reactorgrpc.stub;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+import javax.annotation.Nullable;
+
+import io.grpc.stub.ClientCallStreamObserver;
+import io.grpc.stub.StreamObserver;
+import reactor.core.publisher.Operators;
+
+/**
+ * This class is an implementation of GRPC based Range Publisher. Note, implementation
+ * sends data on the specified ExecutorService e.g simulates observerOn behaviours
+ */
+public class TestCallStreamObserverProducer extends ClientCallStreamObserver<Integer> {
+
+    private static final long serialVersionUID = 2587302975077663557L;
+
+    final Queue<Integer>                  requestsQueue = new ConcurrentLinkedQueue<Integer>();
+    final int                             end;
+    final ExecutorService                 executorService;
+    final StreamObserver<? super Integer> actual;
+
+    int index;
+
+    volatile long requested;
+    public static final AtomicLongFieldUpdater<TestCallStreamObserverProducer> REQUESTED =
+        AtomicLongFieldUpdater.newUpdater(TestCallStreamObserverProducer.class, "requested");
+
+    volatile boolean cancelled;
+
+    TestCallStreamObserverProducer(ExecutorService executorService, StreamObserver<? super Integer> actual, int end) {
+        this.end = end;
+        this.actual = actual;
+        this.executorService = executorService;
+    }
+
+    void slowPath(long r) {
+        long e = 0;
+        int f = end;
+        int i = index;
+        StreamObserver<? super Integer> a = actual;
+
+        for (;;) {
+
+            while (e != r && i != f) {
+                if (cancelled) {
+                    return;
+                }
+
+                a.onNext(i);
+
+                e++;
+                i++;
+            }
+
+            if (i == f) {
+                if (!cancelled) {
+                    a.onCompleted();
+                }
+                return;
+            }
+
+            r = requested;
+            if (e == r) {
+                index = i;
+                r = REQUESTED.addAndGet(this, -e);
+                if (r == 0L) {
+                    return;
+                }
+                e = 0L;
+            }
+        }
+    }
+
+    @Override
+    public void request(final int n) {
+        if (Operators.validate(n)) {
+            requestsQueue.add(n);
+            if (add(this, n) == 0L) {
+                executorService.execute(() -> slowPath(n));
+            }
+        }
+    }
+
+    static long add(TestCallStreamObserverProducer o, long n) {
+        for (;;) {
+            long r = REQUESTED.get(o);
+            if (r == Long.MAX_VALUE) {
+                return Long.MAX_VALUE;
+            }
+            long u = Operators.addCap(r, n);
+            if ((REQUESTED).compareAndSet(o, r, u)) {
+                return r;
+            }
+        }
+    }
+
+    @Override
+    public boolean isReady() {
+        return true;
+    }
+
+    @Override
+    public void cancel(@Nullable String message, @Nullable Throwable cause) {
+        cancelled = true;
+    }
+
+    @Override
+    public void onNext(Integer value) {
+
+    }
+
+    @Override
+    public void onError(Throwable t) {
+
+    }
+
+    @Override
+    public void onCompleted() {
+
+    }
+
+    @Override
+    public void setMessageCompression(boolean enable) {
+
+    }
+
+    @Override
+    public void setOnReadyHandler(Runnable onReadyHandler) {
+
+    }
+
+    @Override
+    public void disableAutoInboundFlowControl() {
+
+    }
+}
+
diff --git a/reactor/reactor-grpc-test-32/pom.xml b/reactor/reactor-grpc-test-32/pom.xml
index 880633a..ae10217 100644
--- a/reactor/reactor-grpc-test-32/pom.xml
+++ b/reactor/reactor-grpc-test-32/pom.xml
@@ -78,7 +78,7 @@
             <extension>
                 <groupId>kr.motd.maven</groupId>
                 <artifactId>os-maven-plugin</artifactId>
-                <version>1.4.1.Final</version>
+                <version>1.6.0</version>
             </extension>
         </extensions>
         <plugins>
@@ -95,7 +95,7 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
-                <version>3.6.1</version>
+                <version>${compiler.plugin.version}</version>
                 <configuration>
                     <source>1.8</source>
                     <target>1.8</target>
@@ -105,7 +105,7 @@
             <plugin>
                 <groupId>org.xolstice.maven.plugins</groupId>
                 <artifactId>protobuf-maven-plugin</artifactId>
-                <version>0.5.0</version>
+                <version>${protobuf.plugin.version}</version>
                 <configuration>
                     <protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
                     <pluginId>grpc-java</pluginId>
diff --git a/reactor/reactor-grpc-test/pom.xml b/reactor/reactor-grpc-test/pom.xml
index 9df0fd0..15c1bb1 100644
--- a/reactor/reactor-grpc-test/pom.xml
+++ b/reactor/reactor-grpc-test/pom.xml
@@ -99,7 +99,7 @@
             <extension>
                 <groupId>kr.motd.maven</groupId>
                 <artifactId>os-maven-plugin</artifactId>
-                <version>1.4.1.Final</version>
+                <version>1.6.0</version>
             </extension>
         </extensions>
         <plugins>
@@ -116,7 +116,7 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
-                <version>3.6.1</version>
+                <version>${compiler.plugin.version}</version>
                 <configuration>
                     <source>1.8</source>
                     <target>1.8</target>
@@ -126,7 +126,7 @@
             <plugin>
                 <groupId>org.xolstice.maven.plugins</groupId>
                 <artifactId>protobuf-maven-plugin</artifactId>
-                <version>0.5.0</version>
+                <version>${protobuf.plugin.version}</version>
                 <configuration>
                     <protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
                     <pluginId>grpc-java</pluginId>
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 4b19bf4..c0eae2a 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
@@ -31,7 +31,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 @SuppressWarnings("Duplicates")
 @RunWith(Parameterized.class)
 public class BackpressureIntegrationTest {
-    private static final int NUMBER_OF_STREAM_ELEMENTS = 270;
+    private static final int NUMBER_OF_STREAM_ELEMENTS = 512 + 512 / 2;
 
 
     @Parameterized.Parameters
@@ -91,9 +91,9 @@ public class BackpressureIntegrationTest {
         stepVerifier
                     .expectNextMatches(v -> v.getNumber(0) == NUMBER_OF_STREAM_ELEMENTS - 1)
                     .expectComplete()
-                    .verify(Duration.ofSeconds(5));
+                    .verify(Duration.ofSeconds(15));
 
-        assertThat(numberOfWaits.get()).isEqualTo(1);
+        assertThat(numberOfWaits.get()).isBetween(1L, 2L);
     }
 
     @Test
@@ -117,9 +117,9 @@ public class BackpressureIntegrationTest {
                 .create(reactorResponse)
                 .expectNextCount(NUMBER_OF_STREAM_ELEMENTS)
                 .expectComplete()
-                .verify(Duration.ofSeconds(5));
+                .verify(Duration.ofSeconds(15));
 
-        assertThat(numberOfWaits.get()).isEqualTo(1);
+        assertThat(numberOfWaits.get()).isBetween(1L, 2L);
     }
 
     @Test
@@ -147,9 +147,9 @@ public class BackpressureIntegrationTest {
         stepVerifier
                 .expectNextCount(NUMBER_OF_STREAM_ELEMENTS)
                 .expectComplete()
-                .verify(Duration.ofSeconds(5));
+                .verify(Duration.ofSeconds(15));
 
-        assertThat(numberOfWaits.get()).isEqualTo(1);
+        assertThat(numberOfWaits.get()).isBetween(1L, 2L);
     }
 
     @Test
@@ -179,9 +179,9 @@ public class BackpressureIntegrationTest {
         stepVerifier
                 .expectNextMatches(v -> v.getNumber(0) == NUMBER_OF_STREAM_ELEMENTS - 1)
                 .expectComplete()
-                .verify(Duration.ofSeconds(5));
+                .verify(Duration.ofSeconds(15));
 
-        assertThat(numberOfWaits.get()).isEqualTo(1);
+        assertThat(numberOfWaits.get()).isBetween(1L, 2L);
     }
 
 
diff --git a/rx-java/rxgrpc-test/pom.xml b/rx-java/rxgrpc-test/pom.xml
index 64270d7..9603ecb 100644
--- a/rx-java/rxgrpc-test/pom.xml
+++ b/rx-java/rxgrpc-test/pom.xml
@@ -93,7 +93,7 @@
             <extension>
                 <groupId>kr.motd.maven</groupId>
                 <artifactId>os-maven-plugin</artifactId>
-                <version>1.4.1.Final</version>
+                <version>1.6.0</version>
             </extension>
         </extensions>
         <plugins>
@@ -110,7 +110,7 @@
             <plugin>
                 <groupId>org.xolstice.maven.plugins</groupId>
                 <artifactId>protobuf-maven-plugin</artifactId>
-                <version>0.5.0</version>
+                <version>${protobuf.plugin.version}</version>
                 <configuration>
                     <protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
                     <pluginId>grpc-java</pluginId>
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 5b90833..649ad17 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
@@ -34,7 +34,7 @@ public class BackpressureIntegrationTest {
     @Rule
     public UnhandledRxJavaErrorRule errorRule = new UnhandledRxJavaErrorRule().autoVerifyNoError();
 
-    private static final int NUMBER_OF_STREAM_ELEMENTS = 140;
+    private static final int NUMBER_OF_STREAM_ELEMENTS = 512 + 512 / 2;
 
     private static AtomicLong lastValueTime;
     private static AtomicLong numberOfWaits;
@@ -90,11 +90,11 @@ public class BackpressureIntegrationTest {
 
         TestObserver<NumberProto.Number> rxResponse = rxRequest.as(stub::requestPressure).test();
 
-        rxResponse.awaitTerminalEvent(5, TimeUnit.SECONDS);
+        rxResponse.awaitTerminalEvent(15, TimeUnit.SECONDS);
         rxResponse.assertComplete()
                 .assertValue(v -> v.getNumber(0) == NUMBER_OF_STREAM_ELEMENTS - 1);
 
-        assertThat(numberOfWaits.get()).isEqualTo(1);
+        assertThat(numberOfWaits.get()).isBetween(1L, 2L);
     }
 
     @Test
@@ -109,11 +109,11 @@ public class BackpressureIntegrationTest {
                 .doOnNext(n -> waitIfValuesAreEqual(n.getNumber(0), 3))
                 .test();
 
-        rxResponse.awaitTerminalEvent(5, TimeUnit.SECONDS);
+        rxResponse.awaitTerminalEvent(15, TimeUnit.SECONDS);
         rxResponse.assertComplete()
                 .assertValueCount(NUMBER_OF_STREAM_ELEMENTS);
 
-        assertThat(numberOfWaits.get()).isEqualTo(1);
+        assertThat(numberOfWaits.get()).isBetween(1L, 2L);
     }
 
     @Test
@@ -127,11 +127,11 @@ public class BackpressureIntegrationTest {
                 .doOnNext(n -> waitIfValuesAreEqual(n.getNumber(0), 3))
                 .test();
 
-        rxResponse.awaitTerminalEvent(5, TimeUnit.SECONDS);
+        rxResponse.awaitTerminalEvent(15, TimeUnit.SECONDS);
         rxResponse.assertComplete()
                 .assertValueCount(NUMBER_OF_STREAM_ELEMENTS);
 
-        assertThat(numberOfWaits.get()).isEqualTo(1);
+        assertThat(numberOfWaits.get()).isBetween(1L, 2L);
     }
 
     @Test
@@ -147,11 +147,11 @@ public class BackpressureIntegrationTest {
 
         TestSubscriber<NumberProto.Number> rxResponse = rxRequest.compose(stub::twoWayRequestPressure).test();
 
-        rxResponse.awaitTerminalEvent(5, TimeUnit.SECONDS);
+        rxResponse.awaitTerminalEvent(15, TimeUnit.SECONDS);
         rxResponse.assertComplete()
                 .assertValue(v -> v.getNumber(0) == NUMBER_OF_STREAM_ELEMENTS - 1);
 
-        assertThat(numberOfWaits.get()).isEqualTo(1);
+        assertThat(numberOfWaits.get()).isBetween(1L, 2L);
     }