diff --git a/rx-netty/src/main/java/io/reactivex/netty/protocol/http/client/HttpClientResponse.java b/rx-netty/src/main/java/io/reactivex/netty/protocol/http/client/HttpClientResponse.java
index 0da57895..4cd04f00 100644
--- a/rx-netty/src/main/java/io/reactivex/netty/protocol/http/client/HttpClientResponse.java
+++ b/rx-netty/src/main/java/io/reactivex/netty/protocol/http/client/HttpClientResponse.java
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package io.reactivex.netty.protocol.http.client;
 
 import io.netty.handler.codec.http.Cookie;
@@ -23,7 +24,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import rx.Observable;
-import rx.subjects.PublishSubject;
+import rx.subjects.Subject;
 
 import java.util.Map;
 import java.util.Set;
@@ -47,13 +48,13 @@ public class HttpClientResponse<T> {
     public static final String KEEP_ALIVE_TIMEOUT_HEADER_ATTR = "timeout";
 
     private final HttpResponse nettyResponse;
-    private final PublishSubject<T> contentSubject;
+    private final Subject<T, T> contentSubject;
     private final HttpResponseHeaders responseHeaders;
     private final HttpVersion httpVersion;
     private final HttpResponseStatus status;
     private final CookiesHolder cookiesHolder;
 
-    public HttpClientResponse(HttpResponse nettyResponse, PublishSubject<T> contentSubject) {
+    public HttpClientResponse(HttpResponse nettyResponse, Subject<T, T> contentSubject) {
         this.nettyResponse = nettyResponse;
         this.contentSubject = contentSubject;
         httpVersion = this.nettyResponse.getProtocolVersion();
diff --git a/rx-netty/src/main/java/io/reactivex/netty/protocol/http/server/HttpServerRequest.java b/rx-netty/src/main/java/io/reactivex/netty/protocol/http/server/HttpServerRequest.java
index 2baf4c9e..54e50c91 100644
--- a/rx-netty/src/main/java/io/reactivex/netty/protocol/http/server/HttpServerRequest.java
+++ b/rx-netty/src/main/java/io/reactivex/netty/protocol/http/server/HttpServerRequest.java
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package io.reactivex.netty.protocol.http.server;
 
 import io.netty.handler.codec.http.Cookie;
@@ -21,7 +22,7 @@
 import io.netty.handler.codec.http.HttpVersion;
 import io.reactivex.netty.protocol.http.CookiesHolder;
 import rx.Observable;
-import rx.subjects.PublishSubject;
+import rx.subjects.Subject;
 
 import java.util.List;
 import java.util.Map;
@@ -34,13 +35,13 @@ public class HttpServerRequest<T> {
 
     private final HttpRequest nettyRequest;
     private final HttpRequestHeaders headers;
-    private final PublishSubject<T> contentSubject;
+    private final Subject<T, T> contentSubject;
     private final HttpMethod method;
     private final HttpVersion protocolVersion;
     private final UriInfoHolder uriInfoHolder;
     private final CookiesHolder cookiesHolder;
 
-    public HttpServerRequest(HttpRequest nettyRequest, PublishSubject<T> contentSubject) {
+    public HttpServerRequest(HttpRequest nettyRequest, Subject<T, T> contentSubject) {
         this.nettyRequest = nettyRequest;
         headers = new HttpRequestHeaders(this.nettyRequest);
         method = this.nettyRequest.getMethod();