Fixes #474 (disable auto release leaks buffers when content not subscribed) (#497)

When auto-release it turned off and the HTTP content is not subscribed, the buffers which are disposed from `UnicastContentSubject` are not released (which otherwise would have been released by the subscriber).

This change properly release the disposed buffers when auto-release is turned off.

Author: NiteshKant

rxnetty/src/main/java/io/reactivex/netty/protocol/http/UnicastContentSubject.java

@@ -92,7 +92,7 @@ private UnicastContentSubject(final State<T> state, long noSubscriptionTimeout,
      * @return The new instance of {@link UnicastContentSubject}
      */
     public static <T> UnicastContentSubject<T> createWithoutNoSubscriptionTimeout(Action0 onUnsubscribe) {
-        State<T> state = new State<T>(onUnsubscribe);
+        State<T> state = new State<T>(onUnsubscribe, null);
         return new UnicastContentSubject<T>(state);
     }
 
@@ -125,7 +125,13 @@ public static <T> UnicastContentSubject<T> create(long noSubscriptionTimeout, Ti
 
     public static <T> UnicastContentSubject<T> create(long noSubscriptionTimeout, TimeUnit timeUnit,
                                                       Scheduler timeoutScheduler, Action0 onUnsubscribe) {
-        State<T> state = new State<T>(onUnsubscribe);
+        return create(noSubscriptionTimeout, timeUnit, timeoutScheduler, onUnsubscribe, null);
+    }
+
+    public static <T> UnicastContentSubject<T> create(long noSubscriptionTimeout, TimeUnit timeUnit,
+                                                      Scheduler timeoutScheduler, Action0 onUnsubscribe,
+                                                      Action1<T> onDispose) {
+        State<T> state = new State<T>(onUnsubscribe, onDispose);
         return new UnicastContentSubject<T>(state, noSubscriptionTimeout, timeUnit, timeoutScheduler);
     }
 
@@ -140,7 +146,8 @@ public static <T> UnicastContentSubject<T> create(long noSubscriptionTimeout, Ti
      */
     public boolean disposeIfNotSubscribed() {
         if (state.casState(State.STATES.UNSUBSCRIBED, State.STATES.DISPOSED)) {
-            state.bufferedObservable.subscribe(Subscribers.empty()); // Drain all items so that ByteBuf gets released.
+            Subscriber<T> sub = null == state.onDispose ? Subscribers.<T>empty() : Subscribers.create(state.onDispose);
+            state.bufferedObservable.subscribe(sub); // Drain all items so that ByteBuf gets released.
             return true;
         }
         return false;
@@ -156,10 +163,12 @@ public void updateTimeoutIfNotScheduled(long noSubscriptionTimeout, TimeUnit tim
     private static final class State<T> {
 
         private final Action0 onUnsubscribe;
+        private final Action1<T> onDispose;
         private volatile Subscription releaseSubscription;
 
-        private State(Action0 onUnsubscribe) {
+        private State(Action0 onUnsubscribe, Action1<T> onDispose) {
             this.onUnsubscribe = onUnsubscribe;
+            this.onDispose = onDispose;
             final BufferUntilSubscriber<T> bufferedSubject = BufferUntilSubscriber.create();
             bufferedObservable = bufferedSubject.lift(new AutoReleaseByteBufOperator<T>()); // Always auto-release
             bufferedObserver = bufferedSubject;

rxnetty/src/main/java/io/reactivex/netty/protocol/http/client/ClientRequestResponseConverter.java

@@ -52,6 +52,8 @@
 import rx.Observer;
 import rx.Subscriber;
 import rx.functions.Action0;
+import rx.functions.Action1;
+import rx.schedulers.Schedulers;
 
 /**
  * A channel handler for {@link HttpClient} to convert netty's http request/response objects to {@link HttpClient}'s
@@ -371,7 +373,15 @@ public void call() {
             };
 
             if (subscriptionTimeout > 0) {
-                contentSubject = UnicastContentSubject.create(subscriptionTimeout, subscriptionTimeoutUnit, onUnsubscribe);
+                contentSubject = UnicastContentSubject.create(subscriptionTimeout, subscriptionTimeoutUnit,
+                                                              Schedulers.computation(), onUnsubscribe, new Action1() {
+                            @Override
+                            public void call(Object o) {
+                                if (!autoReleaseBuffers) {
+                                    ReferenceCountUtil.release(o);
+                                }
+                            }
+                        });
             } else {
                 contentSubject = UnicastContentSubject.createWithoutNoSubscriptionTimeout(onUnsubscribe);
             }

rxnetty/src/main/java/io/reactivex/netty/protocol/http/server/ServerRequestResponseConverter.java

@@ -42,6 +42,10 @@
 import io.reactivex.netty.server.ServerMetricsEvent;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import rx.functions.Action0;
+import rx.functions.Action1;
+import rx.functions.Actions;
+import rx.schedulers.Schedulers;
 
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
@@ -217,7 +221,16 @@ private final class RequestState {
 
         @SuppressWarnings({"rawtypes", "unchecked"})
         private void createRxRequest(ChannelHandlerContext ctx, HttpRequest httpRequest) {
-            contentSubject = UnicastContentSubject.create(requestContentSubscriptionTimeoutMs, TimeUnit.MILLISECONDS);
+            contentSubject = UnicastContentSubject.create(requestContentSubscriptionTimeoutMs, TimeUnit.MILLISECONDS,
+                                                          Schedulers.computation(), Actions.empty(), new Action1() {
+                        @Override
+                        public void call(Object o) {
+                            if (!autoReleaseBuffers) {
+                                ReferenceCountUtil.release(o);
+                            }
+
+                        }
+                    });
             rxRequest = new HttpServerRequest(ctx.channel(), httpRequest, contentSubject);
         }
 

rxnetty/src/test/java/io/reactivex/netty/protocol/http/UnicastContentSubjectTest.java

@@ -212,6 +212,26 @@ public void call() {
         Assert.assertTrue("Inner subscriber did not unsubscribe on inner completion.", innerUnsubscribe.get());
     }
 
+    @Test
+    public void testDispose() throws Exception {
+        TestScheduler scheduler = Schedulers.test();
+        UnicastContentSubject<ByteBuf> sub = UnicastContentSubject.create(1, TimeUnit.HOURS, scheduler, null,
+                                                                          new Action1<ByteBuf>() {
+                                                                              @Override
+                                                                              public void call(ByteBuf bb) {
+                                                                                  bb.release();
+                                                                              }
+                                                                          });
+        ByteBuf bb = Unpooled.buffer();
+        sub.onNext(bb);
+
+        Assert.assertEquals("Unexpected ref count for buffer before dispose.", 2, bb.refCnt());
+
+        scheduler.advanceTimeBy(1, TimeUnit.HOURS);
+
+        Assert.assertEquals("Unexpected ref count for buffer post dispose.", 0, bb.refCnt());
+    }
+
     private static class OnUnsubscribeAction implements Action0 {
 
         private volatile boolean called;