Delayed write streams race condition. (#517)

`BackpressureManagingHandler` was not correctly enqueuing the writes to the eventloop when done from outside the eventloop.
Fixed the problem and also added test.

Author: NiteshKant

rxnetty-common/src/main/java/io/reactivex/netty/channel/BackpressureManagingHandler.java

@@ -503,7 +503,8 @@ public void onNext(Object nextItem) {
                 if (null == writeWorker) {
                     if (!inEL) {
                         atleastOneWriteEnqueued = true;
-                    } else if (atleastOneWriteEnqueued) {
+                    }
+                    if (atleastOneWriteEnqueued) {
                         writeWorker = Schedulers.computation().createWorker();
                     }
                 }
@@ -594,7 +595,7 @@ public void call() {
         private void onTermination(Throwable throwableIfAny) {
             int _listeningTo;
             boolean _shouldCompletePromise;
-            final boolean flush;
+            final boolean enqueueFlush;
 
             /**
              * The intent here is to NOT give listener callbacks via promise completion within the sync block.
@@ -607,7 +608,7 @@ private void onTermination(Throwable throwableIfAny) {
              * This co-oridantion is done via the flag isPromiseCompletedOnWriteComplete
              */
             synchronized (guard) {
-                flush = atleastOneWriteEnqueued;
+                enqueueFlush = atleastOneWriteEnqueued;
                 isDone = true;
                 _listeningTo = listeningTo;
                 /**
@@ -617,7 +618,7 @@ private void onTermination(Throwable throwableIfAny) {
                 _shouldCompletePromise = 0 == _listeningTo && !isPromiseCompletedOnWriteComplete;
             }
 
-            if (flush) {
+            if (enqueueFlush) {
                 writeWorker.schedule(new Action0() {
                     @Override
                     public void call() {

rxnetty-http/src/test/java/io/reactivex/netty/protocol/http/server/HttpEndToEndTest.java

@@ -0,0 +1,63 @@
+package io.reactivex.netty.protocol.http.server;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.reactivex.netty.protocol.http.client.HttpClientResponse;
+import org.junit.Rule;
+import org.junit.Test;
+import rx.Observable;
+import rx.Observable.OnSubscribe;
+import rx.Scheduler.Worker;
+import rx.Subscriber;
+import rx.functions.Action0;
+import rx.schedulers.Schedulers;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static io.reactivex.netty.protocol.http.server.HttpServerRule.*;
+import static org.hamcrest.MatcherAssert.*;
+import static org.hamcrest.Matchers.*;
+
+public class HttpEndToEndTest {
+
+    @Rule
+    public final HttpServerRule rule = new HttpServerRule();
+
+    @Test(timeout = 60000)
+    public void testDelayedWrites() throws Exception {
+
+        final AtomicReference<Throwable> errorFromWriteStreamCompletion = new AtomicReference<>();
+        final Worker worker = Schedulers.computation().createWorker();
+        rule.startServer(new RequestHandler<ByteBuf, ByteBuf>() {
+            @Override
+            public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) {
+                return response.writeString(Observable.create(new OnSubscribe<String>() {
+                    @Override
+                    public void call(final Subscriber<? super String> subscriber) {
+                        worker.schedule(new Action0() {
+                            @Override
+                            public void call() {
+                                try {
+                                    subscriber.onNext(WELCOME_SERVER_MSG);
+                                    subscriber.onCompleted();
+                                } catch (Exception e) {
+                                    errorFromWriteStreamCompletion.set(e);
+                                }
+                            }
+                        }, 1, TimeUnit.MILLISECONDS);
+
+                    }
+                }));
+            }
+        });
+
+        final HttpClientResponse<ByteBuf> response = rule.sendRequest(rule.getClient().createGet("/"));
+
+        assertThat("Unexpected response code.", response.getStatus(), is(HttpResponseStatus.OK));
+
+        rule.assertResponseContent(response);
+
+        assertThat("Unexpected exception on server.", errorFromWriteStreamCompletion.get(), is(nullValue()));
+    }
+}

rxnetty-http/src/test/java/io/reactivex/netty/protocol/http/server/HttpServerRule.java

@@ -17,8 +17,8 @@
 package io.reactivex.netty.protocol.http.server;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.handler.logging.LogLevel;
 import io.reactivex.netty.protocol.http.client.HttpClient;
-import io.reactivex.netty.protocol.http.client.HttpClientRequest;
 import io.reactivex.netty.protocol.http.client.HttpClientResponse;
 import org.junit.rules.ExternalResource;
 import org.junit.runner.Description;
@@ -46,7 +46,7 @@ public Statement apply(final Statement base, Description description) {
         return new Statement() {
             @Override
             public void evaluate() throws Throwable {
-                server = HttpServer.newServer();
+                server = HttpServer.newServer().enableWireLogging("test", LogLevel.INFO);
                 base.evaluate();
             }
         };
@@ -74,7 +74,7 @@ public void setupClient(HttpClient<ByteBuf, ByteBuf> client) {
         this.client = client;
     }
 
-    public HttpClientResponse<ByteBuf> sendRequest(HttpClientRequest<ByteBuf, ByteBuf> request) {
+    public HttpClientResponse<ByteBuf> sendRequest(Observable<HttpClientResponse<ByteBuf>> request) {
         TestSubscriber<HttpClientResponse<ByteBuf>> subscriber = new TestSubscriber<>();
 
         request.subscribe(subscriber);