Repository: salesforce/reactive-grpc
Pull request #38 (2018-01-18 07:56:44)

Feature/android
Addresses #37 

* Split common generator code from common stub code.
* Refactored common stub module to target JDK6.
* Refactored rxgrpc-stub module to target JDK6.
* Added documentation describing various levels of Android support.


Patch:
diff --git a/README.md b/README.md
index 9ca102b..ed38e87 100644
--- a/README.md
+++ b/README.md
@@ -13,6 +13,7 @@ Reactive gRPC supports the following reactive programming models:
 * [RxJava 2](https://github.com/salesforce/reactive-grpc/tree/master/rx-java)
 * [Spring Reactor](https://github.com/salesforce/reactive-grpc/tree/master/reactor)
 * (Eventually) [Java9 Flow](https://community.oracle.com/docs/DOC-1006738)
+* (Eventually) [Akka Streams](https://doc.akka.io/docs/akka/2.5/stream/index.html)
 
 # Usage
 See the readme in each technology-specific sub-directory for usage details.
@@ -26,6 +27,13 @@ See the readme in each technology-specific sub-directory for usage details.
 * [ReactorChat](https://github.com/salesforce/reactive-grpc/tree/master/demos/ReactorChat) - An IRC-like chat system using Spring Reactor and ReactorGrpc.
 * [ReactorChat-Kotlin](https://github.com/salesforce/reactive-grpc/tree/master/demos/ReactorChat-Kotlin) - ReactorChat, but showing Kotlin integration.
 
+# Android support
+Reactive gRPC supports Android to the same level of the underlying reactive technologies.
+
+* Rx-Java - Generated code targets Java 6, so it _should_ work with all versions of Android >= 2.3 (SDK 9).
+* Spring Reactor - [Not officially supported.](http://projectreactor.io/docs/core/release/reference/docs/index.html#prerequisites) 
+  "Reactor 3 does not officially support or target Android, however, it should work fine with Android SDK 26 (Android O) and above."
+
 # Back-pressure
 Reactive gRPC stubs support bi-directional streaming with back-pressure. Under the hood, Reactive gRPC is built atop the vanilla
 gRPC service stubs generated by protoc. As such, they inherit gRPC's HTTP/2-based back-pressure model.
diff --git a/reactive-grpc-common/pom.xml b/common/reactive-grpc-common/pom.xml
similarity index 58%
rename from reactive-grpc-common/pom.xml
rename to common/reactive-grpc-common/pom.xml
index 5216937..ab0a48c 100644
--- a/reactive-grpc-common/pom.xml
+++ b/common/reactive-grpc-common/pom.xml
@@ -13,21 +13,19 @@
         <groupId>com.salesforce.servicelibs</groupId>
         <artifactId>reactive-grpc</artifactId>
         <version>0.7.4-SNAPSHOT</version>
-        <relativePath>../pom.xml</relativePath>
+        <relativePath>../../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>reactive-grpc-common</artifactId>
 
+    <properties>
+        <java.version>1.6</java.version>
+        <maven.compiler.source>${java.version}</maven.compiler.source>
+        <maven.compiler.target>${java.version}</maven.compiler.target>
+    </properties>
+
     <dependencies>
-        <dependency>
-            <groupId>com.salesforce.servicelibs</groupId>
-            <artifactId>grpc-contrib</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>com.salesforce.servicelibs</groupId>
-            <artifactId>jprotoc</artifactId>
-        </dependency>
         <dependency>
             <groupId>org.reactivestreams</groupId>
             <artifactId>reactive-streams</artifactId>
@@ -60,8 +58,31 @@
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-checkstyle-plugin</artifactId>
                 <configuration>
-                    <configLocation>../checkstyle.xml</configLocation>
-                    <suppressionsLocation>../checkstyle_ignore.xml</suppressionsLocation>
+                    <configLocation>../../checkstyle.xml</configLocation>
+                    <suppressionsLocation>../../checkstyle_ignore.xml</suppressionsLocation>
+                </configuration>
+            </plugin>
+
+            <!-- Fail the build if Java7+ sneak in, because Android -->
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>animal-sniffer-maven-plugin</artifactId>
+                <version>1.7</version>
+                <executions>
+                    <execution>
+                        <id>signature-check</id>
+                        <phase>verify</phase>
+                        <goals>
+                            <goal>check</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <signature>
+                        <groupId>org.codehaus.mojo.signature</groupId>
+                        <artifactId>java16</artifactId>
+                        <version>1.0</version>
+                    </signature>
                 </configuration>
             </plugin>
         </plugins>
diff --git a/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/BiConsumer.java b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/BiConsumer.java
new file mode 100644
index 0000000..36e96be
--- /dev/null
+++ b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/BiConsumer.java
@@ -0,0 +1,25 @@
+/*
+ *  Copyright (c) 2017, 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.reactivegrpccommon;
+
+/**
+ * Java6 compatibility.
+ *
+ * @param <T>
+ * @param <U>
+ */
+public interface BiConsumer<T, U> {
+
+    /**
+     * Performs this operation on the given arguments.
+     *
+     * @param t the first input argument
+     * @param u the second input argument
+     */
+    void accept(T t, U u);
+}
diff --git a/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/CancellableStreamObserver.java b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/CancellableStreamObserver.java
similarity index 100%
rename from reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/CancellableStreamObserver.java
rename to common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/CancellableStreamObserver.java
diff --git a/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/Consumer.java b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/Consumer.java
new file mode 100644
index 0000000..a319064
--- /dev/null
+++ b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/Consumer.java
@@ -0,0 +1,23 @@
+/*
+ *  Copyright (c) 2017, 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.reactivegrpccommon;
+
+/**
+ * Java6 compatibility.
+ *
+ * @param <T>
+ */
+public interface Consumer<T> {
+
+    /**
+     * Performs this operation on the given argument.
+     *
+     * @param t the input argument
+     */
+    void accept(T t);
+}
\ No newline at end of file
diff --git a/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/Function.java b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/Function.java
new file mode 100644
index 0000000..3db02b7
--- /dev/null
+++ b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/Function.java
@@ -0,0 +1,25 @@
+/*
+ *  Copyright (c) 2017, 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.reactivegrpccommon;
+
+/**
+ * Java6 compatibility.
+ *
+ * @param <T>
+ * @param <R>
+ */
+public interface Function<T, R> {
+
+    /**
+     * Applies this function to the given argument.
+     *
+     * @param t the function argument
+     * @return the function result
+     */
+    R apply(T t);
+}
diff --git a/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveConsumerStreamObserver.java b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveConsumerStreamObserver.java
similarity index 95%
rename from reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveConsumerStreamObserver.java
rename to common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveConsumerStreamObserver.java
index 1f85d81..f81d3eb 100644
--- a/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveConsumerStreamObserver.java
+++ b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveConsumerStreamObserver.java
@@ -40,7 +40,7 @@ public abstract class ReactiveConsumerStreamObserver<TRequest, TResponse> implem
 
     @Override
     public void beforeStart(ClientCallStreamObserver<TRequest> requestStream) {
-        publisher = new ReactiveStreamObserverPublisher<>(Preconditions.checkNotNull(requestStream));
+        publisher = new ReactiveStreamObserverPublisher<TResponse>(Preconditions.checkNotNull(requestStream));
         rxConsumer = getReactiveConsumerFromPublisher(publisher);
         beforeStartCalled.countDown();
     }
diff --git a/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveExecutor.java b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveExecutor.java
similarity index 100%
rename from reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveExecutor.java
rename to common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveExecutor.java
diff --git a/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveProducerConsumerStreamObserver.java b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveProducerConsumerStreamObserver.java
similarity index 97%
rename from reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveProducerConsumerStreamObserver.java
rename to common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveProducerConsumerStreamObserver.java
index 713998f..674a9a0 100644
--- a/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveProducerConsumerStreamObserver.java
+++ b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveProducerConsumerStreamObserver.java
@@ -29,7 +29,7 @@ public abstract class ReactiveProducerConsumerStreamObserver<TRequest, TResponse
     @Override
     public void beforeStart(ClientCallStreamObserver<TRequest> requestStream) {
         super.beforeStart(Preconditions.checkNotNull(requestStream));
-        onReadyHandler = new ReactivePublisherBackpressureOnReadyHandler<>(requestStream);
+        onReadyHandler = new ReactivePublisherBackpressureOnReadyHandler<TRequest>(requestStream);
     }
 
     public void rxSubscribe() {
diff --git a/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveProducerStreamObserver.java b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveProducerStreamObserver.java
similarity index 65%
rename from reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveProducerStreamObserver.java
rename to common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveProducerStreamObserver.java
index bc67ac0..56883c9 100644
--- a/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveProducerStreamObserver.java
+++ b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveProducerStreamObserver.java
@@ -8,29 +8,28 @@
 package com.salesforce.reactivegrpccommon;
 
 import com.google.common.base.Preconditions;
-import com.salesforce.grpc.contrib.LambdaStreamObserver;
 import io.grpc.stub.ClientCallStreamObserver;
 import io.grpc.stub.ClientResponseObserver;
+import io.grpc.stub.StreamObserver;
 import org.reactivestreams.Publisher;
 
-import java.util.function.Consumer;
-
 /**
  * LambdaStreamObserver configures client-side manual flow control for the producing end of a message stream.
  *
  * @param <TRequest>
  * @param <TResponse>
  */
-public class ReactiveProducerStreamObserver<TRequest, TResponse> extends LambdaStreamObserver<TResponse> implements ClientResponseObserver<TRequest, TResponse> {
-    private Publisher<TRequest> rxProducer;
+public class ReactiveProducerStreamObserver<TRequest, TResponse> implements StreamObserver<TResponse>, ClientResponseObserver<TRequest, TResponse> {
+    private final Publisher<TRequest> rxProducer;
+    private final Consumer<TResponse> onNext;
+    private final Consumer<Throwable> onError;
+    private final Runnable onCompleted;
     private ReactivePublisherBackpressureOnReadyHandler<TRequest> onReadyHandler;
 
     public ReactiveProducerStreamObserver(Publisher<TRequest> rxProducer, Consumer<TResponse> onNext, Consumer<Throwable> onError, Runnable onCompleted) {
-        super(
-            Preconditions.checkNotNull(onNext),
-            Preconditions.checkNotNull(onError),
-            Preconditions.checkNotNull(onCompleted)
-        );
+        this.onNext = Preconditions.checkNotNull(onNext);
+        this.onError = Preconditions.checkNotNull(onError);
+        this.onCompleted = Preconditions.checkNotNull(onCompleted);
         this.rxProducer = Preconditions.checkNotNull(rxProducer);
     }
 
@@ -39,7 +38,7 @@ public class ReactiveProducerStreamObserver<TRequest, TResponse> extends LambdaS
         Preconditions.checkNotNull(producerStream);
         // Subscribe to the rxProducer with an adapter to a gRPC StreamObserver that respects backpressure
         // signals from the underlying gRPC client transport.
-        onReadyHandler = new ReactivePublisherBackpressureOnReadyHandler<>(producerStream);
+        onReadyHandler = new ReactivePublisherBackpressureOnReadyHandler<TRequest>(producerStream);
     }
 
     public void rxSubscribe() {
@@ -49,4 +48,19 @@ public class ReactiveProducerStreamObserver<TRequest, TResponse> extends LambdaS
     public void cancel() {
         onReadyHandler.cancel();
     }
+
+    @Override
+    public void onNext(TResponse tResponse) {
+        onNext.accept(tResponse);
+    }
+
+    @Override
+    public void onError(Throwable throwable) {
+        onError.accept(throwable);
+    }
+
+    @Override
+    public void onCompleted() {
+        onCompleted.run();
+    }
 }
\ No newline at end of file
diff --git a/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactivePublisherBackpressureOnReadyHandler.java b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactivePublisherBackpressureOnReadyHandler.java
similarity index 96%
rename from reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactivePublisherBackpressureOnReadyHandler.java
rename to common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactivePublisherBackpressureOnReadyHandler.java
index 2288e92..e5c1118 100644
--- a/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactivePublisherBackpressureOnReadyHandler.java
+++ b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactivePublisherBackpressureOnReadyHandler.java
@@ -55,7 +55,12 @@ public class ReactivePublisherBackpressureOnReadyHandler<T> implements Subscribe
     public ReactivePublisherBackpressureOnReadyHandler(ServerCallStreamObserver<T> requestStream) {
         this.requestStream = Preconditions.checkNotNull(requestStream);
         requestStream.setOnReadyHandler(this);
-        requestStream.setOnCancelHandler(() -> subscription.cancel());
+        requestStream.setOnCancelHandler(new Runnable() {
+            @Override
+            public void run() {
+                subscription.cancel();
+            }
+        });
     }
 
     @Override
diff --git a/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveStreamObserverPublisher.java b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveStreamObserverPublisher.java
similarity index 98%
rename from reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveStreamObserverPublisher.java
rename to common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveStreamObserverPublisher.java
index 4e6bbb5..f6f734f 100644
--- a/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveStreamObserverPublisher.java
+++ b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveStreamObserverPublisher.java
@@ -61,7 +61,7 @@ public class ReactiveStreamObserverPublisher<T> implements Publisher<T>, StreamO
             @Override
             public void request(long l) {
                 // Some Reactive Streams implementations use Long.MAX_VALUE to indicate "all messages"; gRPC uses Integer.MAX_VALUE.
-                int i = (int) Long.min(l, Integer.MAX_VALUE);
+                int i = (int) Math.min(l, Integer.MAX_VALUE);
 
                 // Very rarely, request() gets called before the client has finished setting up its stream. If this
                 // happens, wait momentarily and try again.
diff --git a/common/reactive-grpc-gencommon/pom.xml b/common/reactive-grpc-gencommon/pom.xml
new file mode 100644
index 0000000..694bd5f
--- /dev/null
+++ b/common/reactive-grpc-gencommon/pom.xml
@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<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>
+        <artifactId>reactive-grpc</artifactId>
+        <groupId>com.salesforce.servicelibs</groupId>
+        <version>0.7.4-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>reactive-grpc-gencommon</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.salesforce.servicelibs</groupId>
+            <artifactId>grpc-contrib</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.salesforce.servicelibs</groupId>
+            <artifactId>jprotoc</artifactId>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-checkstyle-plugin</artifactId>
+                <configuration>
+                    <configLocation>../../checkstyle.xml</configLocation>
+                    <suppressionsLocation>../../checkstyle_ignore.xml</suppressionsLocation>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>
\ No newline at end of file
diff --git a/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveGrpcGenerator.java b/common/reactive-grpc-gencommon/src/main/java/com/salesforce/reactivegrpcgencommon/ReactiveGrpcGenerator.java
similarity index 98%
rename from reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveGrpcGenerator.java
rename to common/reactive-grpc-gencommon/src/main/java/com/salesforce/reactivegrpcgencommon/ReactiveGrpcGenerator.java
index 7f5bf5d..4010626 100644
--- a/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpccommon/ReactiveGrpcGenerator.java
+++ b/common/reactive-grpc-gencommon/src/main/java/com/salesforce/reactivegrpcgencommon/ReactiveGrpcGenerator.java
@@ -5,9 +5,10 @@
  *  For full license text, see LICENSE.txt file in the repo root  or https://opensource.org/licenses/BSD-3-Clause
  */
 
-package com.salesforce.reactivegrpccommon;
+package com.salesforce.reactivegrpcgencommon;
 
 import com.google.common.base.Strings;
+import com.google.common.html.HtmlEscapers;
 import com.google.protobuf.DescriptorProtos.FileDescriptorProto;
 import com.google.protobuf.DescriptorProtos.FileOptions;
 import com.google.protobuf.DescriptorProtos.MethodDescriptorProto;
@@ -24,8 +25,6 @@ import java.util.List;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import static com.google.common.html.HtmlEscapers.htmlEscaper;
-
 /**
  * Abstract class for protoc generators generating Reactive Streams bindings for gRPC.
  */
@@ -173,7 +172,7 @@ public abstract class ReactiveGrpcGenerator extends Generator {
         if (!comments.isEmpty()) {
             StringBuilder builder = new StringBuilder("/**\n")
                     .append(prefix).append(" * <pre>\n");
-            Arrays.stream(htmlEscaper().escape(comments).split("\n"))
+            Arrays.stream(HtmlEscapers.htmlEscaper().escape(comments).split("\n"))
                     .forEach(line -> builder.append(prefix).append(" * ").append(line).append("\n"));
             builder
                     .append(prefix).append(" * <pre>\n")
diff --git a/demos/ReactorChat-Kotlin/pom.xml b/demos/ReactorChat-Kotlin/pom.xml
index 8e325e6..ac86021 100644
--- a/demos/ReactorChat-Kotlin/pom.xml
+++ b/demos/ReactorChat-Kotlin/pom.xml
@@ -30,6 +30,10 @@
             <groupId>io.grpc</groupId>
             <artifactId>grpc-netty</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-protobuf</artifactId>
+        </dependency>
         <dependency>
             <groupId>${project.groupId}</groupId>
             <artifactId>reactor-grpc-stub</artifactId>
diff --git a/demos/ReactorChat/pom.xml b/demos/ReactorChat/pom.xml
index 68ab22c..cc99aa3 100644
--- a/demos/ReactorChat/pom.xml
+++ b/demos/ReactorChat/pom.xml
@@ -30,6 +30,10 @@
             <groupId>io.grpc</groupId>
             <artifactId>grpc-netty</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-protobuf</artifactId>
+        </dependency>
         <dependency>
             <groupId>${project.groupId}</groupId>
             <artifactId>reactor-grpc-stub</artifactId>
diff --git a/demos/RxChat/pom.xml b/demos/RxChat/pom.xml
index 68d9364..b07a738 100644
--- a/demos/RxChat/pom.xml
+++ b/demos/RxChat/pom.xml
@@ -30,6 +30,10 @@
             <groupId>io.grpc</groupId>
             <artifactId>grpc-netty</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-protobuf</artifactId>
+        </dependency>
         <dependency>
             <groupId>${project.groupId}</groupId>
             <artifactId>rxgrpc-stub</artifactId>
diff --git a/pom.xml b/pom.xml
index b3df933..a699c84 100644
--- a/pom.xml
+++ b/pom.xml
@@ -35,7 +35,8 @@
     </developers>
 
     <modules>
-        <module>reactive-grpc-common</module>
+        <module>common/reactive-grpc-common</module>
+        <module>common/reactive-grpc-gencommon</module>
         <module>rx-java/rxgrpc</module>
         <module>rx-java/rxgrpc-stub</module>
         <module>rx-java/rxgrpc-tck</module>
diff --git a/reactor/README.md b/reactor/README.md
index 83c3303..299e9a6 100644
--- a/reactor/README.md
+++ b/reactor/README.md
@@ -2,6 +2,11 @@ Overview
 ========
 Reactor-gRPC is a set of gRPC bindings for reactive programming with [Reactor](http://projectreactor.io/).
 
+### Android support
+Reactive gRPC supports Android to the same level of the underlying reactive technologies. Spring Reactor 
+does [not officially support Android](http://projectreactor.io/docs/core/release/reference/docs/index.html#prerequisites), 
+however, "it should work fine with Android SDK 26 (Android O) and above."
+
 Installation
 =====
 ### Maven
diff --git a/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/ClientCalls.java b/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/ClientCalls.java
index 7355cb7..015105b 100644
--- a/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/ClientCalls.java
+++ b/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/ClientCalls.java
@@ -8,7 +8,6 @@
 package com.salesforce.reactorgrpc.stub;
 
 import com.google.common.util.concurrent.Runnables;
-import com.salesforce.grpc.contrib.LambdaStreamObserver;
 import com.salesforce.reactivegrpccommon.CancellableStreamObserver;
 import com.salesforce.reactivegrpccommon.ReactiveProducerStreamObserver;
 import io.grpc.stub.StreamObserver;
@@ -38,11 +37,22 @@ public final class ClientCalls {
         try {
             return Mono
                     .<TResponse>create(emitter -> rxRequest.subscribe(
-                        request -> delegate.accept(request, new LambdaStreamObserver<TResponse>(
-                            emitter::success,
-                            emitter::error,
-                            Runnables.doNothing()
-                        )),
+                        request -> delegate.accept(request, new StreamObserver<TResponse>() {
+                            @Override
+                            public void onNext(TResponse tResponse) {
+                                emitter.success(tResponse);
+                            }
+
+                            @Override
+                            public void onError(Throwable throwable) {
+                                emitter.error(throwable);
+                            }
+
+                            @Override
+                            public void onCompleted() {
+                                // Do nothing
+                            }
+                        }),
                         emitter::error
                     ))
                     .transform(Operators.lift(new SubscribeOnlyOnceLifter<TResponse>()));
diff --git a/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/ServerCalls.java b/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/ServerCalls.java
index 5200740..3056dc2 100644
--- a/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/ServerCalls.java
+++ b/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/ServerCalls.java
@@ -8,7 +8,6 @@
 package com.salesforce.reactorgrpc.stub;
 
 import com.google.common.base.Preconditions;
-import com.salesforce.grpc.contrib.LambdaStreamObserver;
 import com.salesforce.reactivegrpccommon.ReactiveExecutor;
 import com.salesforce.reactivegrpccommon.ReactivePublisherBackpressureOnReadyHandler;
 import com.salesforce.reactivegrpccommon.ReactiveStreamObserverPublisher;
@@ -111,10 +110,7 @@ public final class ServerCalls {
             responseObserver.onError(prepareError(throwable));
         }
 
-        return new LambdaStreamObserver<>(
-                streamObserverPublisher::onNext,
-                streamObserverPublisher::onError,
-                streamObserverPublisher::onCompleted);
+        return streamObserverPublisher;
     }
 
     /**
@@ -156,10 +152,7 @@ public final class ServerCalls {
             responseObserver.onError(prepareError(throwable));
         }
 
-        return new LambdaStreamObserver<>(
-                streamObserverPublisher::onNext,
-                streamObserverPublisher::onError,
-                streamObserverPublisher::onCompleted);
+        return streamObserverPublisher;
     }
 
     private static Throwable prepareError(Throwable throwable) {
diff --git a/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/SubscribeOnlyOnceLifter.java b/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/SubscribeOnlyOnceLifter.java
index b3e3d1e..02a4a37 100644
--- a/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/SubscribeOnlyOnceLifter.java
+++ b/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/SubscribeOnlyOnceLifter.java
@@ -30,7 +30,7 @@ public class SubscribeOnlyOnceLifter<T> implements BiFunction<Scannable, CoreSub
             public void onSubscribe(Subscription subscription) {
                 if (subscribedOnce.getAndSet(true)) {
                     throw new NullPointerException("You cannot directly subscribe to a gRPC service multiple times " +
-                            "concurrently. Use Flowable.share() instead.");
+                            "concurrently. Use Flux.share() instead.");
                 } else {
                     coreSubscriber.onSubscribe(subscription);
                 }
diff --git a/reactor/reactor-grpc/pom.xml b/reactor/reactor-grpc/pom.xml
index a3e141f..e3b59ff 100644
--- a/reactor/reactor-grpc/pom.xml
+++ b/reactor/reactor-grpc/pom.xml
@@ -22,7 +22,7 @@
     <dependencies>
         <dependency>
             <groupId>${project.groupId}</groupId>
-            <artifactId>reactive-grpc-common</artifactId>
+            <artifactId>reactive-grpc-gencommon</artifactId>
             <version>${project.version}</version>
         </dependency>
     </dependencies>
diff --git a/reactor/reactor-grpc/src/main/java/com/salesforce/reactorgrpc/ReactorGrpcGenerator.java b/reactor/reactor-grpc/src/main/java/com/salesforce/reactorgrpc/ReactorGrpcGenerator.java
index b0a7a84..1481377 100644
--- a/reactor/reactor-grpc/src/main/java/com/salesforce/reactorgrpc/ReactorGrpcGenerator.java
+++ b/reactor/reactor-grpc/src/main/java/com/salesforce/reactorgrpc/ReactorGrpcGenerator.java
@@ -8,7 +8,7 @@
 package com.salesforce.reactorgrpc;
 
 import com.salesforce.jprotoc.ProtocPlugin;
-import com.salesforce.reactivegrpccommon.ReactiveGrpcGenerator;
+import com.salesforce.reactivegrpcgencommon.ReactiveGrpcGenerator;
 
 /**
  * A protoc generator for generating Reactor bindings for gRPC.
diff --git a/rx-java/README.md b/rx-java/README.md
index d79a8da..4320c92 100644
--- a/rx-java/README.md
+++ b/rx-java/README.md
@@ -2,6 +2,12 @@ Overview
 ========
 RxGrpc is a new set of gRPC bindings for reactive programming with [RxJava](https://github.com/ReactiveX/RxJava).
 
+### Android support
+Reactive gRPC supports Android to the same level of the underlying reactive technologies. The generated Rx-Java binding
+code targets Java 6, so it _should_ work with all versions of Android >= 2.3 (SDK 9).
+
+See: https://github.com/ReactiveX/RxJava#version-2x-javadoc
+
 Installation
 ============
 ### Maven
diff --git a/rx-java/rxgrpc-stub/pom.xml b/rx-java/rxgrpc-stub/pom.xml
index 7fc3f76..5c96076 100644
--- a/rx-java/rxgrpc-stub/pom.xml
+++ b/rx-java/rxgrpc-stub/pom.xml
@@ -19,6 +19,12 @@
 
     <artifactId>rxgrpc-stub</artifactId>
 
+    <properties>
+        <java.version>1.6</java.version>
+        <maven.compiler.source>${java.version}</maven.compiler.source>
+        <maven.compiler.target>${java.version}</maven.compiler.target>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>${project.groupId}</groupId>
@@ -57,6 +63,29 @@
                     <suppressionsLocation>../../checkstyle_ignore.xml</suppressionsLocation>
                 </configuration>
             </plugin>
+
+            <!-- Fail the build if Java7+ sneak in, because Android -->
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>animal-sniffer-maven-plugin</artifactId>
+                <version>1.7</version>
+                <executions>
+                    <execution>
+                        <id>signature-check</id>
+                        <phase>verify</phase>
+                        <goals>
+                            <goal>check</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <signature>
+                        <groupId>org.codehaus.mojo.signature</groupId>
+                        <artifactId>java16</artifactId>
+                        <version>1.0</version>
+                    </signature>
+                </configuration>
+            </plugin>
         </plugins>
     </build>
 </project>
\ No newline at end of file
diff --git a/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/ClientCalls.java b/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/ClientCalls.java
index 5116c17..a9ac596 100644
--- a/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/ClientCalls.java
+++ b/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/ClientCalls.java
@@ -8,15 +8,17 @@
 package com.salesforce.rxgrpc.stub;
 
 import com.google.common.util.concurrent.Runnables;
-import com.salesforce.grpc.contrib.LambdaStreamObserver;
 import com.salesforce.reactivegrpccommon.CancellableStreamObserver;
 import com.salesforce.reactivegrpccommon.ReactiveProducerStreamObserver;
 import io.grpc.stub.StreamObserver;
 import io.reactivex.Flowable;
 import io.reactivex.Single;
 
-import java.util.function.BiConsumer;
-import java.util.function.Function;
+import com.salesforce.reactivegrpccommon.BiConsumer;
+import com.salesforce.reactivegrpccommon.Function;
+import io.reactivex.SingleEmitter;
+import io.reactivex.SingleOnSubscribe;
+import io.reactivex.functions.Consumer;
 
 /**
  * Utility functions for processing different client call idioms. We have one-to-one correspondence
@@ -32,19 +34,45 @@ public final class ClientCalls {
      * Implements a unary -> unary call using {@link Single} -> {@link Single}.
      */
     public static <TRequest, TResponse> Single<TResponse> oneToOne(
-            Single<TRequest> rxRequest,
-            BiConsumer<TRequest, StreamObserver<TResponse>> delegate) {
+            final Single<TRequest> rxRequest,
+            final BiConsumer<TRequest, StreamObserver<TResponse>> delegate) {
         try {
             return Single
-                    .<TResponse>create(emitter -> rxRequest.subscribe(
-                        request -> delegate.accept(request, new LambdaStreamObserver<TResponse>(
-                            emitter::onSuccess,
-                            emitter::onError,
-                            Runnables.doNothing()
-                        )),
-                        emitter::onError
-                    ))
-                    .lift(new SubscribeOnlyOnceSingleOperator<>());
+                .create(new SingleOnSubscribe<TResponse>() {
+                    @Override
+                    public void subscribe(final SingleEmitter<TResponse> emitter) {
+                        rxRequest.subscribe(
+                            new Consumer<TRequest>() {
+                                @Override
+                                public void accept(TRequest request) {
+                                    delegate.accept(request, new StreamObserver<TResponse>() {
+                                        @Override
+                                        public void onNext(TResponse tResponse) {
+                                            emitter.onSuccess(tResponse);
+                                        }
+
+                                        @Override
+                                        public void onError(Throwable throwable) {
+                                            emitter.onError(throwable);
+                                        }
+
+                                        @Override
+                                        public void onCompleted() {
+                                            // Do nothing
+                                        }
+                                    });
+                                }
+                            },
+                            new Consumer<Throwable>() {
+                                @Override
+                                public void accept(Throwable t) {
+                                    emitter.onError(t);
+                                }
+                            }
+                        );
+                    }
+                })
+                .lift(new SubscribeOnlyOnceSingleOperator<TResponse>());
         } catch (Throwable throwable) {
             return Single.error(throwable);
         }
@@ -56,12 +84,17 @@ public final class ClientCalls {
      */
     public static <TRequest, TResponse> Flowable<TResponse> oneToMany(
             Single<TRequest> rxRequest,
-            BiConsumer<TRequest, StreamObserver<TResponse>> delegate) {
+            final BiConsumer<TRequest, StreamObserver<TResponse>> delegate) {
         try {
-            RxConsumerStreamObserver<TRequest, TResponse> consumerStreamObserver = new RxConsumerStreamObserver<>();
-            rxRequest.subscribe(request -> delegate.accept(request, consumerStreamObserver));
+            final RxConsumerStreamObserver<TRequest, TResponse> consumerStreamObserver = new RxConsumerStreamObserver<TRequest, TResponse>();
+            rxRequest.subscribe(new Consumer<TRequest>() {
+                @Override
+                public void accept(TRequest request) {
+                    delegate.accept(request, consumerStreamObserver);
+                }
+            });
             return ((Flowable<TResponse>) consumerStreamObserver.getRxConsumer())
-                    .lift(new SubscribeOnlyOnceFlowableOperator<>());
+                    .lift(new SubscribeOnlyOnceFlowableOperator<TResponse>());
         } catch (Throwable throwable) {
             return Flowable.error(throwable);
         }
@@ -72,21 +105,39 @@ public final class ClientCalls {
      * messages.
      */
     public static <TRequest, TResponse> Single<TResponse> manyToOne(
-            Flowable<TRequest> rxRequest,
-            Function<StreamObserver<TResponse>, StreamObserver<TRequest>> delegate) {
+            final Flowable<TRequest> rxRequest,
+            final Function<StreamObserver<TResponse>, StreamObserver<TRequest>> delegate) {
         try {
             return Single
-                    .<TResponse>create(emitter -> {
-                        ReactiveProducerStreamObserver<TRequest, TResponse> rxProducerStreamObserver = new ReactiveProducerStreamObserver<>(
-                                rxRequest,
-                                emitter::onSuccess,
-                                emitter::onError,
-                                Runnables.doNothing());
+                .create(new SingleOnSubscribe<TResponse>() {
+                    @Override
+                    public void subscribe(final SingleEmitter<TResponse> emitter) {
+                        final ReactiveProducerStreamObserver<TRequest, TResponse> rxProducerStreamObserver = new ReactiveProducerStreamObserver<TRequest, TResponse>(
+                            rxRequest,
+                            new com.salesforce.reactivegrpccommon.Consumer<TResponse>() {
+                                @Override
+                                public void accept(TResponse t) {
+                                    emitter.onSuccess(t);
+                                }
+                            },
+                            new com.salesforce.reactivegrpccommon.Consumer<Throwable>() {
+                                @Override
+                                public void accept(Throwable t) {
+                                    emitter.onError(t);
+                                }
+                            },
+                            Runnables.doNothing());
                         delegate.apply(
-                                new CancellableStreamObserver<>(rxProducerStreamObserver,
-                                rxProducerStreamObserver::cancel));
+                            new CancellableStreamObserver<TRequest, TResponse>(rxProducerStreamObserver,
+                                new Runnable() {
+                                    @Override
+                                    public void run() {
+                                        rxProducerStreamObserver.cancel();
+                                    }
+                                }));
                         rxProducerStreamObserver.rxSubscribe();
-                    }).lift(new SubscribeOnlyOnceSingleOperator<>());
+                    }
+                }).lift(new SubscribeOnlyOnceSingleOperator<TResponse>());
         } catch (Throwable throwable) {
             return Single.error(throwable);
         }
@@ -100,11 +151,16 @@ public final class ClientCalls {
             Flowable<TRequest> rxRequest,
             Function<StreamObserver<TResponse>, StreamObserver<TRequest>> delegate) {
         try {
-            RxProducerConsumerStreamObserver<TRequest, TResponse> consumerStreamObserver = new RxProducerConsumerStreamObserver<>(rxRequest);
-            delegate.apply(new CancellableStreamObserver<>(consumerStreamObserver, consumerStreamObserver::cancel));
+            final RxProducerConsumerStreamObserver<TRequest, TResponse> consumerStreamObserver = new RxProducerConsumerStreamObserver<TRequest, TResponse>(rxRequest);
+            delegate.apply(new CancellableStreamObserver<TRequest, TResponse>(consumerStreamObserver, new Runnable() {
+                @Override
+                public void run() {
+                    consumerStreamObserver.cancel();
+                }
+            }));
             consumerStreamObserver.rxSubscribe();
             return ((Flowable<TResponse>) consumerStreamObserver.getRxConsumer())
-                    .lift(new SubscribeOnlyOnceFlowableOperator<>());
+                    .lift(new SubscribeOnlyOnceFlowableOperator<TResponse>());
         } catch (Throwable throwable) {
             return Flowable.error(throwable);
         }
diff --git a/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/ServerCalls.java b/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/ServerCalls.java
index aa33a69..b137311 100644
--- a/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/ServerCalls.java
+++ b/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/ServerCalls.java
@@ -8,7 +8,7 @@
 package com.salesforce.rxgrpc.stub;
 
 import com.google.common.base.Preconditions;
-import com.salesforce.grpc.contrib.LambdaStreamObserver;
+import com.salesforce.reactivegrpccommon.Function;
 import com.salesforce.reactivegrpccommon.ReactiveExecutor;
 import com.salesforce.reactivegrpccommon.ReactivePublisherBackpressureOnReadyHandler;
 import com.salesforce.reactivegrpccommon.ReactiveStreamObserverPublisher;
@@ -20,10 +20,11 @@ import io.grpc.stub.ServerCallStreamObserver;
 import io.grpc.stub.StreamObserver;
 import io.reactivex.Flowable;
 import io.reactivex.Single;
+import io.reactivex.functions.Action;
+import io.reactivex.functions.Consumer;
 import io.reactivex.schedulers.Schedulers;
 import org.reactivestreams.Subscriber;
-
-import java.util.function.Function;
+import org.reactivestreams.Subscription;
 
 /**
  * Utility functions for processing different server call idioms. We have one-to-one correspondence
@@ -39,22 +40,30 @@ public final class ServerCalls {
      * Implements a unary -> unary call using {@link Single} -> {@link Single}.
      */
     public static <TRequest, TResponse> void oneToOne(
-            TRequest request, StreamObserver<TResponse> responseObserver,
+            TRequest request, final StreamObserver<TResponse> responseObserver,
             Function<Single<TRequest>, Single<TResponse>> delegate) {
         try {
             Single<TRequest> rxRequest = Single.just(request);
 
             Single<TResponse> rxResponse = Preconditions.checkNotNull(delegate.apply(rxRequest));
             rxResponse.subscribe(
-                value -> {
-                    // Don't try to respond if the server has already canceled the request
-                    if (responseObserver instanceof ServerCallStreamObserver && ((ServerCallStreamObserver) responseObserver).isCancelled()) {
-                        return;
-                    }
-                    responseObserver.onNext(value);
-                    responseObserver.onCompleted();
-                },
-                throwable -> responseObserver.onError(prepareError(throwable)));
+                    new Consumer<TResponse>() {
+                        @Override
+                        public void accept(TResponse value) {
+                            // Don't try to respond if the server has already canceled the request
+                            if (responseObserver instanceof ServerCallStreamObserver && ((ServerCallStreamObserver) responseObserver).isCancelled()) {
+                                return;
+                            }
+                            responseObserver.onNext(value);
+                            responseObserver.onCompleted();
+                        }
+                    },
+                    new Consumer<Throwable>() {
+                        @Override
+                        public void accept(Throwable throwable) {
+                            responseObserver.onError(prepareError(throwable));
+                        }
+                    });
         } catch (Throwable throwable) {
             responseObserver.onError(prepareError(throwable));
         }
@@ -71,7 +80,7 @@ public final class ServerCalls {
             Single<TRequest> rxRequest = Single.just(request);
 
             Flowable<TResponse> rxResponse = Preconditions.checkNotNull(delegate.apply(rxRequest));
-            rxResponse.subscribe(new ReactivePublisherBackpressureOnReadyHandler<>(
+            rxResponse.subscribe(new ReactivePublisherBackpressureOnReadyHandler<TResponse>(
                     (ServerCallStreamObserver<TResponse>) responseObserver));
         } catch (Throwable throwable) {
             responseObserver.onError(prepareError(throwable));
@@ -83,38 +92,41 @@ public final class ServerCalls {
      * messages.
      */
     public static <TRequest, TResponse> StreamObserver<TRequest> manyToOne(
-            StreamObserver<TResponse> responseObserver,
+            final StreamObserver<TResponse> responseObserver,
             Function<Flowable<TRequest>, Single<TResponse>> delegate) {
-        ReactiveStreamObserverPublisher<TRequest> streamObserverPublisher =
-                new ReactiveStreamObserverPublisher<>((CallStreamObserver<TResponse>) responseObserver);
+        final ReactiveStreamObserverPublisher<TRequest> streamObserverPublisher =
+                new ReactiveStreamObserverPublisher<TRequest>((CallStreamObserver<TResponse>) responseObserver);
 
         try {
             Single<TResponse> rxResponse = Preconditions.checkNotNull(delegate.apply(
                     Flowable.unsafeCreate(streamObserverPublisher)
                             .observeOn(Schedulers.from(ReactiveExecutor.getSerializingExecutor()))));
             rxResponse.subscribe(
-                value -> {
-                    // Don't try to respond if the server has already canceled the request
-                    if (!streamObserverPublisher.isCanceled()) {
-                        responseObserver.onNext(value);
-                        responseObserver.onCompleted();
+                    new Consumer<TResponse>() {
+                        @Override
+                        public void accept(TResponse value) {
+                            // Don't try to respond if the server has already canceled the request
+                            if (!streamObserverPublisher.isCanceled()) {
+                                responseObserver.onNext(value);
+                                responseObserver.onCompleted();
+                            }
+                        }
+                    },
+                    new Consumer<Throwable>() {
+                        @Override
+                        public void accept(Throwable throwable) {
+                            // Don't try to respond if the server has already canceled the request
+                            if (!streamObserverPublisher.isCanceled()) {
+                                responseObserver.onError(prepareError(throwable));
+                            }
+                        }
                     }
-                },
-                throwable -> {
-                    // Don't try to respond if the server has already canceled the request
-                    if (!streamObserverPublisher.isCanceled()) {
-                        responseObserver.onError(prepareError(throwable));
-                    }
-                }
             );
         } catch (Throwable throwable) {
             responseObserver.onError(prepareError(throwable));
         }
 
-        return new LambdaStreamObserver<>(
-                streamObserverPublisher::onNext,
-                streamObserverPublisher::onError,
-                streamObserverPublisher::onCompleted);
+        return streamObserverPublisher;
     }
 
     /**
@@ -124,42 +136,53 @@ public final class ServerCalls {
     public static <TRequest, TResponse> StreamObserver<TRequest> manyToMany(
             StreamObserver<TResponse> responseObserver,
             Function<Flowable<TRequest>, Flowable<TResponse>> delegate) {
-        ReactiveStreamObserverPublisher<TRequest> streamObserverPublisher =
-                new ReactiveStreamObserverPublisher<>((CallStreamObserver<TResponse>) responseObserver);
+        final ReactiveStreamObserverPublisher<TRequest> streamObserverPublisher =
+                new ReactiveStreamObserverPublisher<TRequest>((CallStreamObserver<TResponse>) responseObserver);
 
         try {
             Flowable<TResponse> rxResponse = Preconditions.checkNotNull(delegate.apply(
                     Flowable.unsafeCreate(streamObserverPublisher)
                             .observeOn(Schedulers.from(ReactiveExecutor.getSerializingExecutor()))));
-            Subscriber<TResponse> subscriber = new ReactivePublisherBackpressureOnReadyHandler<>(
+            final Subscriber<TResponse> subscriber = new ReactivePublisherBackpressureOnReadyHandler<TResponse>(
                     (ServerCallStreamObserver<TResponse>) responseObserver);
             // Don't try to respond if the server has already canceled the request
             rxResponse.subscribe(
-                tResponse -> {
-                    if (!streamObserverPublisher.isCanceled()) {
-                        subscriber.onNext(tResponse);
-                    }
-                },
-                throwable -> {
-                    if (!streamObserverPublisher.isCanceled()) {
-                        subscriber.onError(throwable);
-                    }
-                },
-                () -> {
-                    if (!streamObserverPublisher.isCanceled()) {
-                        subscriber.onComplete();
+                    new Consumer<TResponse>() {
+                        @Override
+                        public void accept(TResponse tResponse) {
+                            if (!streamObserverPublisher.isCanceled()) {
+                                subscriber.onNext(tResponse);
+                            }
+                        }
+                    },
+                    new Consumer<Throwable>() {
+                        @Override
+                        public void accept(Throwable throwable) {
+                            if (!streamObserverPublisher.isCanceled()) {
+                                subscriber.onError(throwable);
+                            }
+                        }
+                    },
+                    new Action() {
+                        @Override
+                        public void run() {
+                            if (!streamObserverPublisher.isCanceled()) {
+                                subscriber.onComplete();
+                            }
+                        }
+                    },
+                    new Consumer<Subscription>() {
+                        @Override
+                        public void accept(Subscription subscription) {
+                            subscriber.onSubscribe(subscription);
+                        }
                     }
-                },
-                subscriber::onSubscribe
             );
         } catch (Throwable throwable) {
             responseObserver.onError(prepareError(throwable));
         }
 
-        return new LambdaStreamObserver<>(
-                streamObserverPublisher::onNext,
-                streamObserverPublisher::onError,
-                streamObserverPublisher::onCompleted);
+        return streamObserverPublisher;
     }
 
     private static Throwable prepareError(Throwable throwable) {
diff --git a/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/SubscribeOnlyOnceFlowableOperator.java b/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/SubscribeOnlyOnceFlowableOperator.java
index 990cce2..a462504 100644
--- a/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/SubscribeOnlyOnceFlowableOperator.java
+++ b/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/SubscribeOnlyOnceFlowableOperator.java
@@ -24,7 +24,7 @@ public class SubscribeOnlyOnceFlowableOperator<T> implements FlowableOperator<T,
     private AtomicBoolean subscribedOnce = new AtomicBoolean(false);
 
     @Override
-    public Subscriber<? super T> apply(Subscriber<? super T> observer) throws Exception {
+    public Subscriber<? super T> apply(final Subscriber<? super T> observer) {
         return new Subscriber<T>() {
             @Override
             public void onSubscribe(Subscription subscription) {
diff --git a/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/SubscribeOnlyOnceSingleOperator.java b/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/SubscribeOnlyOnceSingleOperator.java
index 4c3349b..4e084fe 100644
--- a/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/SubscribeOnlyOnceSingleOperator.java
+++ b/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/SubscribeOnlyOnceSingleOperator.java
@@ -23,7 +23,7 @@ public class SubscribeOnlyOnceSingleOperator<T> implements SingleOperator<T, T>
     private AtomicBoolean subscribedOnce = new AtomicBoolean(false);
 
     @Override
-    public SingleObserver<? super T> apply(SingleObserver<? super T> observer) throws Exception {
+    public SingleObserver<? super T> apply(final SingleObserver<? super T> observer) {
         return new SingleObserver<T>() {
             @Override
             public void onSubscribe(Disposable d) {
diff --git a/rx-java/rxgrpc/pom.xml b/rx-java/rxgrpc/pom.xml
index 6c9a224..53908ce 100644
--- a/rx-java/rxgrpc/pom.xml
+++ b/rx-java/rxgrpc/pom.xml
@@ -22,7 +22,7 @@
     <dependencies>
         <dependency>
             <groupId>${project.groupId}</groupId>
-            <artifactId>reactive-grpc-common</artifactId>
+            <artifactId>reactive-grpc-gencommon</artifactId>
             <version>${project.version}</version>
         </dependency>
     </dependencies>
diff --git a/rx-java/rxgrpc/src/main/java/com/salesforce/rxgrpc/RxGrpcGenerator.java b/rx-java/rxgrpc/src/main/java/com/salesforce/rxgrpc/RxGrpcGenerator.java
index 483a446..f977a71 100644
--- a/rx-java/rxgrpc/src/main/java/com/salesforce/rxgrpc/RxGrpcGenerator.java
+++ b/rx-java/rxgrpc/src/main/java/com/salesforce/rxgrpc/RxGrpcGenerator.java
@@ -8,7 +8,7 @@
 package com.salesforce.rxgrpc;
 
 import com.salesforce.jprotoc.ProtocPlugin;
-import com.salesforce.reactivegrpccommon.ReactiveGrpcGenerator;
+import com.salesforce.reactivegrpcgencommon.ReactiveGrpcGenerator;
 
 /**
  * A protoc generator for generating ReactiveX 2.0 bindings for gRPC.


Test patch:
diff --git a/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/CancellableStreamObserverTest.java b/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/CancellableStreamObserverTest.java
similarity index 68%
rename from reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/CancellableStreamObserverTest.java
rename to common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/CancellableStreamObserverTest.java
index 652ea9b..f9e99eb 100644
--- a/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/CancellableStreamObserverTest.java
+++ b/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/CancellableStreamObserverTest.java
@@ -22,9 +22,14 @@ public class CancellableStreamObserverTest {
     @Test
     public void statusExceptionTriggersHandler() {
         ClientResponseObserver<Object, Object> delegate = mock(ClientResponseObserver.class);
-        AtomicBoolean called = new AtomicBoolean(false);
+        final AtomicBoolean called = new AtomicBoolean(false);
 
-        CancellableStreamObserver<Object, Object> observer = new CancellableStreamObserver<>(delegate, () -> called.set(true));
+        CancellableStreamObserver<Object, Object> observer = new CancellableStreamObserver<Object, Object>(delegate, new Runnable() {
+            @Override
+            public void run() {
+                called.set(true);
+            }
+        });
 
         observer.onError(Status.CANCELLED.asException());
 
@@ -34,9 +39,14 @@ public class CancellableStreamObserverTest {
     @Test
     public void statusRuntimeExceptionTriggersHandler() {
         ClientResponseObserver<Object, Object> delegate = mock(ClientResponseObserver.class);
-        AtomicBoolean called = new AtomicBoolean(false);
+        final AtomicBoolean called = new AtomicBoolean(false);
 
-        CancellableStreamObserver<Object, Object> observer = new CancellableStreamObserver<>(delegate, () -> called.set(true));
+        CancellableStreamObserver<Object, Object> observer = new CancellableStreamObserver<Object, Object>(delegate, new Runnable() {
+            @Override
+            public void run() {
+                called.set(true);
+            }
+        });
 
         observer.onError(Status.CANCELLED.asRuntimeException());
 
@@ -46,9 +56,14 @@ public class CancellableStreamObserverTest {
     @Test
     public void otherExceptionDoesNotTriggersHandler() {
         ClientResponseObserver<Object, Object> delegate = mock(ClientResponseObserver.class);
-        AtomicBoolean called = new AtomicBoolean(false);
-
-        CancellableStreamObserver<Object, Object> observer = new CancellableStreamObserver<>(delegate, () -> called.set(true));
+        final AtomicBoolean called = new AtomicBoolean(false);
+
+        CancellableStreamObserver<Object, Object> observer = new CancellableStreamObserver<Object, Object>(delegate, new Runnable() {
+            @Override
+            public void run() {
+                called.set(true);
+            }
+        });
 
         observer.onError(Status.INTERNAL.asRuntimeException());
 
diff --git a/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/ReactivePublisherBackpressureOnReadyHandlerTest.java b/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/ReactivePublisherBackpressureOnReadyHandlerTest.java
similarity index 89%
rename from reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/ReactivePublisherBackpressureOnReadyHandlerTest.java
rename to common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/ReactivePublisherBackpressureOnReadyHandlerTest.java
index 5283035..f42a165 100644
--- a/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/ReactivePublisherBackpressureOnReadyHandlerTest.java
+++ b/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/ReactivePublisherBackpressureOnReadyHandlerTest.java
@@ -18,7 +18,7 @@ public class ReactivePublisherBackpressureOnReadyHandlerTest {
     @Test
     public void runPrimesThePump() {
         ClientCallStreamObserver<Object> obs = mock(ClientCallStreamObserver.class);
-        ReactivePublisherBackpressureOnReadyHandler<Object> handler = new ReactivePublisherBackpressureOnReadyHandler<>(obs);
+        ReactivePublisherBackpressureOnReadyHandler<Object> handler = new ReactivePublisherBackpressureOnReadyHandler<Object>(obs);
         Subscription sub = mock(Subscription.class);
 
         handler.onSubscribe(sub);
@@ -32,7 +32,7 @@ public class ReactivePublisherBackpressureOnReadyHandlerTest {
         ClientCallStreamObserver<Object> obs = mock(ClientCallStreamObserver.class);
         when(obs.isReady()).thenReturn(true);
 
-        ReactivePublisherBackpressureOnReadyHandler<Object> handler = new ReactivePublisherBackpressureOnReadyHandler<>(obs);
+        ReactivePublisherBackpressureOnReadyHandler<Object> handler = new ReactivePublisherBackpressureOnReadyHandler<Object>(obs);
         Subscription sub = mock(Subscription.class);
 
         handler.onSubscribe(sub);
@@ -49,7 +49,7 @@ public class ReactivePublisherBackpressureOnReadyHandlerTest {
         ClientCallStreamObserver<Object> obs = mock(ClientCallStreamObserver.class);
         when(obs.isReady()).thenReturn(false);
 
-        ReactivePublisherBackpressureOnReadyHandler<Object> handler = new ReactivePublisherBackpressureOnReadyHandler<>(obs);
+        ReactivePublisherBackpressureOnReadyHandler<Object> handler = new ReactivePublisherBackpressureOnReadyHandler<Object>(obs);
         Subscription sub = mock(Subscription.class);
 
         handler.onSubscribe(sub);
diff --git a/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/ReactiveStreamObserverPublisherTest.java b/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/ReactiveStreamObserverPublisherTest.java
similarity index 78%
rename from reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/ReactiveStreamObserverPublisherTest.java
rename to common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/ReactiveStreamObserverPublisherTest.java
index b75026f..4e722d5 100644
--- a/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/ReactiveStreamObserverPublisherTest.java
+++ b/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpccommon/ReactiveStreamObserverPublisherTest.java
@@ -9,6 +9,8 @@ package com.salesforce.reactivegrpccommon;
 
 import io.grpc.stub.CallStreamObserver;
 import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
@@ -24,7 +26,7 @@ public class ReactiveStreamObserverPublisherTest {
         CallStreamObserver<Object> obs = mock(CallStreamObserver.class);
         Subscriber<Object> sub = mock(Subscriber.class);
 
-        ReactiveStreamObserverPublisher<Object> pub = new ReactiveStreamObserverPublisher<>(obs);
+        ReactiveStreamObserverPublisher<Object> pub = new ReactiveStreamObserverPublisher<Object>(obs);
         pub.subscribe(sub);
 
         Object obj = new Object();
@@ -38,7 +40,7 @@ public class ReactiveStreamObserverPublisherTest {
         CallStreamObserver<Object> obs = mock(CallStreamObserver.class);
         Subscriber<Object> sub = mock(Subscriber.class);
 
-        ReactiveStreamObserverPublisher<Object> pub = new ReactiveStreamObserverPublisher<>(obs);
+        ReactiveStreamObserverPublisher<Object> pub = new ReactiveStreamObserverPublisher<Object>(obs);
         pub.subscribe(sub);
 
         Throwable obj = new Exception();
@@ -52,7 +54,7 @@ public class ReactiveStreamObserverPublisherTest {
         CallStreamObserver<Object> obs = mock(CallStreamObserver.class);
         Subscriber<Object> sub = mock(Subscriber.class);
 
-        ReactiveStreamObserverPublisher<Object> pub = new ReactiveStreamObserverPublisher<>(obs);
+        ReactiveStreamObserverPublisher<Object> pub = new ReactiveStreamObserverPublisher<Object>(obs);
         pub.subscribe(sub);
 
         pub.onCompleted();
@@ -64,13 +66,16 @@ public class ReactiveStreamObserverPublisherTest {
         CallStreamObserver<Object> obs = mock(CallStreamObserver.class);
         Subscriber<Object> sub = mock(Subscriber.class);
 
-        AtomicReference<Subscription> subscription = new AtomicReference<>();
-        doAnswer(invocationOnMock -> {
-            subscription.set((Subscription)invocationOnMock.getArguments()[0]);
-            return null;
+        final AtomicReference<Subscription> subscription = new AtomicReference<Subscription>();
+        doAnswer(new Answer() {
+            @Override
+            public Object answer(InvocationOnMock invocationOnMock) {
+                subscription.set((Subscription) invocationOnMock.getArguments()[0]);
+                return null;
+            }
         }).when(sub).onSubscribe(any(Subscription.class));
 
-        ReactiveStreamObserverPublisher<Object> pub = new ReactiveStreamObserverPublisher<>(obs);
+        ReactiveStreamObserverPublisher<Object> pub = new ReactiveStreamObserverPublisher<Object>(obs);
         pub.subscribe(sub);
 
         assertThat(subscription.get()).isNotNull();
diff --git a/reactor/reactor-grpc-test/pom.xml b/reactor/reactor-grpc-test/pom.xml
index 7ffc37d..a7a248e 100644
--- a/reactor/reactor-grpc-test/pom.xml
+++ b/reactor/reactor-grpc-test/pom.xml
@@ -82,6 +82,11 @@
             <artifactId>awaitility</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.salesforce.servicelibs</groupId>
+            <artifactId>grpc-contrib</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/rx-java/rxgrpc-stub/src/test/java/com/salesforce/rxgrpc/stub/SubscribeOnlyOnceTest.java b/rx-java/rxgrpc-stub/src/test/java/com/salesforce/rxgrpc/stub/SubscribeOnlyOnceTest.java
index c392c3c..afdba2e 100644
--- a/rx-java/rxgrpc-stub/src/test/java/com/salesforce/rxgrpc/stub/SubscribeOnlyOnceTest.java
+++ b/rx-java/rxgrpc-stub/src/test/java/com/salesforce/rxgrpc/stub/SubscribeOnlyOnceTest.java
@@ -9,6 +9,7 @@ package com.salesforce.rxgrpc.stub;
 
 import io.reactivex.SingleObserver;
 import io.reactivex.disposables.Disposable;
+import org.assertj.core.api.ThrowableAssert;
 import org.junit.Test;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
@@ -18,15 +19,20 @@ import static org.assertj.core.api.Assertions.*;
 
 public class SubscribeOnlyOnceTest {
     @Test
-    public void subscribeOnlyOnceFlowableOperatorErrorsWhenMultipleSubscribe() throws Exception {
-        SubscribeOnlyOnceFlowableOperator<Object> op = new SubscribeOnlyOnceFlowableOperator<>();
+    public void subscribeOnlyOnceFlowableOperatorErrorsWhenMultipleSubscribe() {
+        SubscribeOnlyOnceFlowableOperator<Object> op = new SubscribeOnlyOnceFlowableOperator<Object>();
         Subscriber<Object> innerSub = mock(Subscriber.class);
-        Subscription subscription = mock(Subscription.class);
+        final Subscription subscription = mock(Subscription.class);
 
-        Subscriber<Object> outerSub = op.apply(innerSub);
+        final Subscriber<Object> outerSub = op.apply(innerSub);
 
         outerSub.onSubscribe(subscription);
-        assertThatThrownBy(() -> outerSub.onSubscribe(subscription))
+        assertThatThrownBy(new ThrowableAssert.ThrowingCallable() {
+            @Override
+            public void call() {
+                outerSub.onSubscribe(subscription);
+            }
+        })
                 .isInstanceOf(NullPointerException.class)
                 .hasMessageContaining("cannot directly subscribe to a gRPC service multiple times");
 
@@ -34,15 +40,20 @@ public class SubscribeOnlyOnceTest {
     }
 
     @Test
-    public void subscribeOnlyOnceSingleOperatorErrorsWhenMultipleSubscribe() throws Exception {
-        SubscribeOnlyOnceSingleOperator<Object> op = new SubscribeOnlyOnceSingleOperator<>();
+    public void subscribeOnlyOnceSingleOperatorErrorsWhenMultipleSubscribe() {
+        SubscribeOnlyOnceSingleOperator<Object> op = new SubscribeOnlyOnceSingleOperator<Object>();
         SingleObserver<Object> innerSub = mock(SingleObserver.class);
-        Disposable disposable = mock(Disposable.class);
+        final Disposable disposable = mock(Disposable.class);
 
-        SingleObserver<Object> outerSub = op.apply(innerSub);
+        final SingleObserver<Object> outerSub = op.apply(innerSub);
 
         outerSub.onSubscribe(disposable);
-        assertThatThrownBy(() -> outerSub.onSubscribe(disposable))
+        assertThatThrownBy(new ThrowableAssert.ThrowingCallable() {
+            @Override
+            public void call() {
+                outerSub.onSubscribe(disposable);
+            }
+        })
                 .isInstanceOf(NullPointerException.class)
                 .hasMessageContaining("cannot directly subscribe to a gRPC service multiple times");
 
diff --git a/rx-java/rxgrpc-test/pom.xml b/rx-java/rxgrpc-test/pom.xml
index 454bbea..c9972f1 100644
--- a/rx-java/rxgrpc-test/pom.xml
+++ b/rx-java/rxgrpc-test/pom.xml
@@ -76,6 +76,11 @@
             <artifactId>awaitility</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.salesforce.servicelibs</groupId>
+            <artifactId>grpc-contrib</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>