Only trigger the timeoutScheduler when not yet subscribed.

This changeset makes sure that the (costly) delayed execution task
is schedule when noone is subscribed yet - since the whole purpose
of its existence is to dispose if not subscribed yet.

Note that a future improvement might be also to unsubscribe if one
subscribes before the cleanup actually runs.

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

@@ -270,10 +270,13 @@ public void onError(Throwable e) {
 
     @Override
     public void onNext(T t) {
-        ReferenceCountUtil.retain(t); // Retain so that post-buffer, the ByteBuf does not get released. Release will be done after reading from the subject.
+        // Retain so that post-buffer, the ByteBuf does not get released.
+        // Release will be done after reading from the subject.
+        ReferenceCountUtil.retain(t);
         state.bufferedSubject.onNext(t);
 
-        if (state.casTimeoutScheduled()) {// Schedule timeout once.
+        // Schedule timeout once and when not subscribed yet.
+        if (state.casTimeoutScheduled() && state.state == State.STATES.UNSUBSCRIBED.ordinal()) {
             timeoutScheduler.subscribe(new Action1<Long>() { // Schedule timeout after the first content arrives.
                 @Override
                 public void call(Long aLong) {