Merge pull request #455 from NiteshKant/0.4.x

NiteshKant · NiteshKant · commit b99f8191280f · 2015-12-15T13:20:07.000-08:00
Disable auto-release buffers option.
diff --git a/build.gradle b/build.gradle
@@ -32,4 +32,10 @@ configure(subprojects) {
 
     testCompile 'junit:junit-dep:4.10'
   }
+
+  test {
+    testLogging {
+      showStandardStreams = true
+    }
+  }
 }
diff --git a/rxnetty/src/main/java/io/reactivex/netty/channel/ObservableConnection.java b/rxnetty/src/main/java/io/reactivex/netty/channel/ObservableConnection.java
@@ -20,6 +20,7 @@
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.util.AttributeKey;
 import io.reactivex.netty.metrics.Clock;
 import io.reactivex.netty.metrics.MetricEventsSubject;
 import io.reactivex.netty.pipeline.ReadTimeoutPipelineConfigurator;
@@ -38,6 +39,8 @@
  */
 public class ObservableConnection<I, O> extends DefaultChannelWriter<O> {
 
+    public static AttributeKey<Boolean> AUTO_RELEASE_BUFFERS = AttributeKey.valueOf("rxnetty_auto_release_buffers");
+
     private Subject<I, I> inputSubject;
     @SuppressWarnings("rawtypes")private final MetricEventsSubject eventsSubject;
     private final ChannelMetricEventProvider metricEventProvider;
diff --git a/rxnetty/src/main/java/io/reactivex/netty/client/AbstractClientBuilder.java b/rxnetty/src/main/java/io/reactivex/netty/client/AbstractClientBuilder.java
@@ -214,6 +214,11 @@ public B withSslEngineFactory(SSLEngineFactory sslEngineFactory) {
         return returnBuilder();
     }
 
+    public B disableAutoReleaseBuffers() {
+        bootstrap.attr(ObservableConnection.AUTO_RELEASE_BUFFERS, false);
+        return returnBuilder();
+    }
+
     public Bootstrap getBootstrap() {
         return bootstrap;
     }
diff --git a/rxnetty/src/main/java/io/reactivex/netty/pipeline/ObservableAdapter.java b/rxnetty/src/main/java/io/reactivex/netty/pipeline/ObservableAdapter.java
@@ -19,6 +19,7 @@
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.util.ReferenceCountUtil;
 import io.reactivex.netty.channel.NewRxConnectionEvent;
+import io.reactivex.netty.channel.ObservableConnection;
 import io.reactivex.netty.client.ConnectionReuseEvent;
 import rx.Observer;
 
@@ -32,6 +33,15 @@ public class ObservableAdapter extends ChannelInboundHandlerAdapter {
     @SuppressWarnings("rawtypes")
     /*Nullable*/ private Observer bridgedObserver; /*This actually is an Rx Subject*/
 
+    private boolean autoReleaseBuffers;
+
+    @Override
+    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+        Boolean autoRelease = ctx.channel().attr(ObservableConnection.AUTO_RELEASE_BUFFERS).get();
+        autoReleaseBuffers = null == autoRelease || autoRelease;
+        super.handlerAdded(ctx);
+    }
+
     @SuppressWarnings("unchecked")
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
@@ -41,7 +51,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
             } catch (ClassCastException cce) {
                 bridgedObserver.onError(new RuntimeException("Mismatched message type.", cce));
             } finally {
-                ReferenceCountUtil.release(msg);
+                if (autoReleaseBuffers) {
+                    ReferenceCountUtil.release(msg);
+                }
             }
         }
     }
diff --git a/rxnetty/src/main/java/io/reactivex/netty/protocol/http/client/ClientRequestResponseConverter.java b/rxnetty/src/main/java/io/reactivex/netty/protocol/http/client/ClientRequestResponseConverter.java
@@ -90,11 +90,20 @@ public class ClientRequestResponseConverter extends ChannelDuplexHandler {
 
     public static final IOException CONN_CLOSE_BEFORE_RESPONSE = new IOException("Connection closed by peer before sending a response.");
 
+    private boolean autoReleaseBuffers;
+
     public ClientRequestResponseConverter(MetricEventsSubject<ClientMetricsEvent<?>> eventsSubject) {
         this.eventsSubject = eventsSubject;
         responseState = new ResponseState();
     }
 
+    @Override
+    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+        Boolean autoRelease = ctx.channel().attr(ObservableConnection.AUTO_RELEASE_BUFFERS).get();
+        autoReleaseBuffers = null == autoRelease || autoRelease;
+        super.handlerAdded(ctx);
+    }
+
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
         Class<?> recievedMsgClass = msg.getClass();
@@ -244,13 +253,15 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
     }
 
     @SuppressWarnings("unchecked")
-    private static void invokeContentOnNext(Object nextObject, ResponseState stateToUse) {
+    private void invokeContentOnNext(Object nextObject, ResponseState stateToUse) {
         try {
             stateToUse.contentSubject.onNext(nextObject);
         } catch (ClassCastException e) {
             stateToUse.contentSubject.onError(e);
         } finally {
-            ReferenceCountUtil.release(nextObject);
+            if (autoReleaseBuffers) {
+                ReferenceCountUtil.release(nextObject);
+            }
         }
     }
 
diff --git a/rxnetty/src/main/java/io/reactivex/netty/protocol/http/server/ServerRequestResponseConverter.java b/rxnetty/src/main/java/io/reactivex/netty/protocol/http/server/ServerRequestResponseConverter.java
@@ -35,6 +35,7 @@
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.LastHttpContent;
 import io.netty.util.ReferenceCountUtil;
+import io.reactivex.netty.channel.ObservableConnection;
 import io.reactivex.netty.metrics.Clock;
 import io.reactivex.netty.metrics.MetricEventsSubject;
 import io.reactivex.netty.protocol.http.UnicastContentSubject;
@@ -76,6 +77,7 @@ public class ServerRequestResponseConverter extends ChannelDuplexHandler {
     private final MetricEventsSubject<ServerMetricsEvent<?>> eventsSubject;
     private final long requestContentSubscriptionTimeoutMs;
     private RequestState currentRequestState;
+    private boolean autoReleaseBuffers = true;
 
     public ServerRequestResponseConverter(MetricEventsSubject<ServerMetricsEvent<?>> eventsSubject,
                                           long requestContentSubscriptionTimeoutMs) {
@@ -84,6 +86,13 @@ public ServerRequestResponseConverter(MetricEventsSubject<ServerMetricsEvent<?>>
         currentRequestState = new RequestState();
     }
 
+    @Override
+    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+        Boolean autoRelease = ctx.channel().attr(ObservableConnection.AUTO_RELEASE_BUFFERS).get();
+        autoReleaseBuffers = null == autoRelease || autoRelease;
+        super.handlerAdded(ctx);
+    }
+
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
         Class<?> recievedMsgClass = msg.getClass();
@@ -188,13 +197,15 @@ public void operationComplete(ChannelFuture future) throws Exception {
     }
 
     @SuppressWarnings({"unchecked", "rawtypes"})
-    private static void invokeContentOnNext(Object nextObject, UnicastContentSubject contentSubject) {
+    private void invokeContentOnNext(Object nextObject, UnicastContentSubject contentSubject) {
         try {
             contentSubject.onNext(nextObject);
         } catch (ClassCastException e) {
             contentSubject.onError(e);
         } finally {
-            ReferenceCountUtil.release(nextObject);
+            if (autoReleaseBuffers) {
+                ReferenceCountUtil.release(nextObject);
+            }
         }
     }
 
diff --git a/rxnetty/src/main/java/io/reactivex/netty/server/AbstractServerBuilder.java b/rxnetty/src/main/java/io/reactivex/netty/server/AbstractServerBuilder.java
@@ -26,6 +26,7 @@
 import io.netty.util.concurrent.EventExecutorGroup;
 import io.reactivex.netty.RxNetty;
 import io.reactivex.netty.channel.ConnectionHandler;
+import io.reactivex.netty.channel.ObservableConnection;
 import io.reactivex.netty.metrics.MetricEventsListener;
 import io.reactivex.netty.metrics.MetricEventsListenerFactory;
 import io.reactivex.netty.pipeline.PipelineConfigurator;
@@ -138,6 +139,11 @@ public B withEventExecutorGroup(EventExecutorGroup eventExecutorGroup) {
         return returnBuilder();
     }
 
+    public B disableAutoReleaseBuffers() {
+        serverBootstrap.attr(ObservableConnection.AUTO_RELEASE_BUFFERS, false);
+        return returnBuilder();
+    }
+
     public PipelineConfigurator<I, O> getPipelineConfigurator() {
         return pipelineConfigurator;
     }
diff --git a/rxnetty/src/main/java/io/reactivex/netty/server/ConnectionBasedServerBuilder.java b/rxnetty/src/main/java/io/reactivex/netty/server/ConnectionBasedServerBuilder.java
@@ -25,6 +25,7 @@
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.reactivex.netty.RxNetty;
 import io.reactivex.netty.channel.ConnectionHandler;
+import io.reactivex.netty.channel.ObservableConnection;
 
 /**
  * @author Nitesh Kant
@@ -73,4 +74,10 @@ public B defaultChannelOptions() {
         childChannelOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
         return super.defaultChannelOptions();
     }
+
+    @Override
+    public B disableAutoReleaseBuffers() {
+        serverBootstrap.childAttr(ObservableConnection.AUTO_RELEASE_BUFFERS, false);
+        return returnBuilder();
+    }
 }
diff --git a/rxnetty/src/test/java/io/reactivex/netty/protocol/http/client/HttpClientTest.java b/rxnetty/src/test/java/io/reactivex/netty/protocol/http/client/HttpClientTest.java
@@ -461,6 +461,33 @@ public void call(String t1) {
         });
     }
 
+    @Test(timeout = 60000)
+    public void testDisableAutoRelease() throws Exception {
+        HttpClient<ByteBuf, ByteBuf> client = RxNetty.<ByteBuf, ByteBuf>newHttpClientBuilder("localhost", port)
+                .disableAutoReleaseBuffers()
+                .enableWireLogging(LogLevel.ERROR).build();
+        Observable<HttpClientResponse<ByteBuf>> response = client.submit(HttpClientRequest.createGet("test/singleEntity"));
+        final List<ByteBuf> result = new ArrayList<ByteBuf>();
+        response.flatMap(new Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>>() {
+            @Override
+            public Observable<ByteBuf> call(HttpClientResponse<ByteBuf> response) {
+                return response.getContent();
+            }
+        }).toBlocking().forEach(new Action1<ByteBuf>() {
+
+            @Override
+            public void call(ByteBuf t1) {
+                result.add(t1);
+            }
+        });
+
+        assertEquals("Response not found.", 1, result.size());
+        assertEquals("Hello world", result.get(0).toString(Charset.defaultCharset()));
+
+        assertEquals("Byte buf auto-released", 1, result.get(0).refCnt());
+        result.get(0).release();
+    }
+
     private static void readResponseContent(Observable<HttpClientResponse<ServerSentEvent>> response,
                                             final List<String> result) {
         response.flatMap(
diff --git a/rxnetty/src/test/java/io/reactivex/netty/protocol/http/server/DisbleAutoReleaseTest.java b/rxnetty/src/test/java/io/reactivex/netty/protocol/http/server/DisbleAutoReleaseTest.java
@@ -0,0 +1,61 @@
+package io.reactivex.netty.protocol.http.server;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.logging.LogLevel;
+import io.reactivex.netty.RxNetty;
+import io.reactivex.netty.protocol.http.client.HttpClientRequest;
+import io.reactivex.netty.protocol.http.client.HttpClientResponse;
+import org.junit.Assert;
+import org.junit.Test;
+import rx.Observable;
+import rx.functions.Action0;
+import rx.functions.Func1;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class DisbleAutoReleaseTest {
+
+    @Test(timeout = 60000)
+    public void testDisableAutoRelease() throws Exception {
+
+        final List<ByteBuf> requestBufs = new ArrayList<ByteBuf>();
+
+        HttpServer<ByteBuf, ByteBuf> server = RxNetty.newHttpServerBuilder(0, new RequestHandler<ByteBuf, ByteBuf>() {
+            @Override
+            public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) {
+                return request.getContent()
+                              .map(new Func1<ByteBuf, Void>() {
+                                  @Override
+                                  public Void call(ByteBuf byteBuf) {
+                                      requestBufs.add(byteBuf);
+                                      return null;
+                                  }
+                              })
+                              .ignoreElements()
+                              .cast(Void.class)
+                              .concatWith(response.writeStringAndFlush("Welcome!"));
+            }
+        }).disableAutoReleaseBuffers().enableWireLogging(LogLevel.DEBUG).build().start();
+
+        final CountDownLatch finishLatch = new CountDownLatch(1);
+        HttpClientResponse<ByteBuf> response = RxNetty.createHttpClient("localhost", server.getServerPort())
+                                                      .submit(HttpClientRequest.createPost("/").withContent("Hello"))
+                                                      .finallyDo(new Action0() {
+                                                          @Override
+                                                          public void call() {
+                                                              finishLatch.countDown();
+                                                          }
+                                                      }).toBlocking().toFuture().get(10, TimeUnit.SECONDS);
+        Assert.assertTrue("The returned observable did not finish.", finishLatch.await(1, TimeUnit.MINUTES));
+        Assert.assertEquals("Request failed.", response.getStatus(), HttpResponseStatus.OK);
+
+        Assert.assertEquals("Unexpected request content on server.", 1, requestBufs.size());
+        Assert.assertEquals("Unexpected request content buffer ref count.", 1, requestBufs.get(0).refCnt());
+
+        requestBufs.get(0).release();
+    }
+}
diff --git a/rxnetty/src/test/java/io/reactivex/netty/protocol/http/websocket/WebSocketClientServerTest.java b/rxnetty/src/test/java/io/reactivex/netty/protocol/http/websocket/WebSocketClientServerTest.java
@@ -32,7 +32,6 @@
 import io.reactivex.netty.RxNetty;
 import io.reactivex.netty.channel.ConnectionHandler;
 import io.reactivex.netty.channel.ObservableConnection;
-import io.reactivex.netty.protocol.http.server.RequestHandler;
 import io.reactivex.netty.protocol.http.server.HttpServerRequest;
 import io.reactivex.netty.protocol.http.server.HttpServerResponse;
 import io.reactivex.netty.server.RxServer;
diff --git a/rxnetty/src/test/resources/log4j.properties b/rxnetty/src/test/resources/log4j.properties
@@ -0,0 +1,20 @@
+#
+# Copyright 2014 Netflix, Inc.
+#
+# 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.
+#
+log4j.rootLogger=DEBUG, stdout
+
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%5p [%t] (%F:%L) - %m%n

Original file line number	Diff line number	Diff line change
`@@ -32,4 +32,10 @@ configure(subprojects) {`
`32`	`32`
`33`	`33`	`testCompile 'junit:junit-dep:4.10'`
`34`	`34`	`}`
	`35`	`+`
	`36`	`+ test {`
	`37`	`+ testLogging {`
	`38`	`+ showStandardStreams = true`
	`39`	`+ }`
	`40`	`+ }`
`35`	`41`	`}`
Original file line number	Diff line number	Diff line change
`@@ -214,6 +214,11 @@ public B withSslEngineFactory(SSLEngineFactory sslEngineFactory) {`
`214`	`214`	`return returnBuilder();`
`215`	`215`	`}`
`216`	`216`
	`217`	`+ public B disableAutoReleaseBuffers() {`
	`218`	`+ bootstrap.attr(ObservableConnection.AUTO_RELEASE_BUFFERS, false);`
	`219`	`+ return returnBuilder();`
	`220`	`+ }`
	`221`	`+`
`217`	`222`	`public Bootstrap getBootstrap() {`
`218`	`223`	`return bootstrap;`
`219`	`224`	`}`