Merge pull request #178 from NiteshKant/master

NiteshKant · NiteshKant · commit 895fd7b35795 · 2014-07-05T01:58:48.000-07:00
Fixes issue #177
diff --git a/rx-netty-contexts/src/test/java/io/reactivex/netty/contexts/http/ContextPropagationTest.java b/rx-netty-contexts/src/test/java/io/reactivex/netty/contexts/http/ContextPropagationTest.java
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package io.reactivex.netty.contexts.http;
 
 import com.netflix.server.context.ContextSerializationException;
@@ -249,7 +250,7 @@ public Observable<Void> handle(HttpServerRequest<ByteBuf> request,
                             @Override
                             public Observable<Void> call(HttpClientResponse<ByteBuf> response) {
                                 serverResponse.setStatus(response.getStatus());
-                                return Observable.empty();
+                                return serverResponse.close(true);
                             }
                         });
             }
diff --git a/rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/helloworld/HelloWorldServer.java b/rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/helloworld/HelloWorldServer.java
@@ -46,7 +46,7 @@ public HttpServer<ByteBuf, ByteBuf> createServer() {
             public Observable<Void> handle(HttpServerRequest<ByteBuf> request, final HttpServerResponse<ByteBuf> response) {
                 printRequestHeader(request);
                 response.writeString("Welcome!!");
-                return response.close();
+                return response.close(false);
             }
         }).pipelineConfigurator(PipelineConfigurators.<ByteBuf, ByteBuf>httpServerConfigurator()).build();
 
diff --git a/rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/post/SimplePostServer.java b/rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/post/SimplePostServer.java
@@ -59,7 +59,7 @@ public String call(String accumulator, String value) {
                     @Override
                     public Observable<Void> call(String clientMessage) {
                         response.writeString(clientMessage.toUpperCase());
-                        return response.close();
+                        return response.close(false);
                     }
                 });
             }
diff --git a/rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/ssl/SslHelloWorldServer.java b/rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/ssl/SslHelloWorldServer.java
@@ -43,7 +43,7 @@ public HttpServer<ByteBuf, ByteBuf> createServer() {
             @Override
             public Observable<Void> handle(HttpServerRequest<ByteBuf> request, final HttpServerResponse<ByteBuf> response) {
                 response.writeStringAndFlush("Welcome!!");
-                return response.close();
+                return response.close(false);
             }
         }).withSslEngineFactory(DefaultFactories.SELF_SIGNED).build();
 
diff --git a/rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/wordcounter/WordCounterServer.java b/rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/wordcounter/WordCounterServer.java
@@ -60,7 +60,7 @@ public String call(ByteBuf content) {
                                   @Override
                                   public Observable<Void> call(Integer counter) {
                                       response.writeString(counter.toString());
-                                      return response.close();
+                                      return response.close(false);
                                   }
                               });
             }
diff --git a/rx-netty/src/main/java/io/reactivex/netty/channel/ChannelWriter.java b/rx-netty/src/main/java/io/reactivex/netty/channel/ChannelWriter.java
@@ -47,4 +47,8 @@ public interface ChannelWriter<O> {
     Observable<Void> writeBytesAndFlush(byte[] msg);
 
     Observable<Void> writeStringAndFlush(String msg);
+
+    Observable<Void> close();
+
+    Observable<Void> close(boolean flush);
 }
diff --git a/rx-netty/src/main/java/io/reactivex/netty/channel/DefaultChannelWriter.java b/rx-netty/src/main/java/io/reactivex/netty/channel/DefaultChannelWriter.java
@@ -23,7 +23,7 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.reactivex.netty.metrics.Clock;
 import io.reactivex.netty.metrics.MetricEventsSubject;
-import io.reactivex.netty.protocol.http.MultipleFutureListener;
+import io.reactivex.netty.util.MultipleFutureListener;
 import rx.Observable;
 import rx.functions.Action0;
 import rx.functions.Action1;
@@ -147,7 +147,6 @@ public ChannelHandlerContext getChannelHandlerContext() {
         return ctx;
     }
 
-    @SuppressWarnings("unchecked")
     protected ChannelFuture writeOnChannel(Object msg) {
         ChannelFuture writeFuture = getChannel().write(msg); // Calling write on context will be wrong as the context will be of a component not necessarily, the tail of the pipeline.
         unflushedWritesListener.get().listen(writeFuture);
@@ -162,16 +161,21 @@ public boolean isCloseIssued() {
         return closeIssued.get();
     }
 
-    @SuppressWarnings("unchecked")
+    @Override
     public Observable<Void> close() {
+        return close(false);
+    }
+
+    @Override
+    public Observable<Void> close(boolean flush) {
         if (closeIssued.compareAndSet(false, true)) {
-            return _close();
+            return _close(flush);
         } else {
             return CONNECTION_ALREADY_CLOSED;
         }
     }
 
-    protected Observable<Void> _close() {
+    protected Observable<Void> _close(boolean flush) {
         return Observable.empty();
     }
 }
diff --git a/rx-netty/src/main/java/io/reactivex/netty/channel/ObservableConnection.java b/rx-netty/src/main/java/io/reactivex/netty/channel/ObservableConnection.java
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package io.reactivex.netty.channel;
 
 import io.netty.channel.ChannelFuture;
@@ -21,6 +22,7 @@
 import io.reactivex.netty.metrics.Clock;
 import io.reactivex.netty.metrics.MetricEventsSubject;
 import io.reactivex.netty.pipeline.ReadTimeoutPipelineConfigurator;
+import io.reactivex.netty.util.NoOpSubscriber;
 import rx.Observable;
 import rx.Subscriber;
 import rx.subjects.PublishSubject;
@@ -67,24 +69,55 @@ public Observable<Void> close() {
     }
 
     @Override
-    protected Observable<Void> _close() {
-        PublishSubject<I> thisSubject = inputSubject;
-        cleanupConnection(); // Cleanup is required irrespective of close underlying connection (pooled connection)
-        Observable<Void> toReturn = _closeChannel();
-        thisSubject.onCompleted(); // This is just to make sure we make the subject as completed after we finish
-        // closing the channel, results in more deterministic behavior for clients.
-        return toReturn;
-    }
-
-    protected void cleanupConnection() {
+    protected Observable<Void> _close(boolean flush) {
+        final PublishSubject<I> thisSubject = inputSubject;
         cancelPendingWrites(true);
         ReadTimeoutPipelineConfigurator.disableReadTimeout(getChannelHandlerContext().pipeline());
+        if (flush) {
+            Observable<Void> toReturn = flush().lift(new Observable.Operator<Void, Void>() {
+                @Override
+                public Subscriber<? super Void> call(final Subscriber<? super Void> child) {
+                    return new Subscriber<Void>() {
+                        @Override
+                        public void onCompleted() {
+                            _closeChannel().subscribe(child);
+                            thisSubject.onCompleted(); // Even though closeChannel() returns an Observable, close itself is eager.
+                            // So this makes sure we send onCompleted() on subject after close is initialized.
+                            // This results in more deterministic behavior for clients.
+                        }
+
+                        @Override
+                        public void onError(Throwable e) {
+                            child.onError(e);
+                        }
+
+                        @Override
+                        public void onNext(Void aVoid) {
+                            // Insignificant
+                        }
+                    };
+                }
+            });
+            toReturn.subscribe(new NoOpSubscriber<Void>()); // Since subscribing to returned Observable is not required
+                                                            // by the caller and we need to be subscribed to trigger the
+                                                            // close of channel (_closeChannel()), it is required to
+                                                            // subscribe to the returned Observable. We are not
+                                                            // interested in the result so NoOpSub is used.
+            return toReturn;
+        } else {
+            Observable<Void> toReturn = _closeChannel();
+            thisSubject.onCompleted(); // Even though closeChannel() returns an Observable, close itself is eager.
+            // So this makes sure we send onCompleted() on subject after close is initialized.
+            // This results in more deterministic behavior for clients.
+            return toReturn;
+        }
     }
 
     @SuppressWarnings("unchecked")
     protected Observable<Void> _closeChannel() {
         closeStartTimeMillis = Clock.newStartTimeMillis();
         eventsSubject.onEvent(metricEventProvider.getChannelCloseStartEvent());
+
         final ChannelFuture closeFuture = getChannelHandlerContext().close();
 
         /**
@@ -126,4 +159,5 @@ public void operationComplete(ChannelFuture future) throws Exception {
     protected void updateInputSubject(PublishSubject<I> newSubject) {
         inputSubject = newSubject;
     }
+
 }
diff --git a/rx-netty/src/main/java/io/reactivex/netty/protocol/http/client/ClientRequestResponseConverter.java b/rx-netty/src/main/java/io/reactivex/netty/protocol/http/client/ClientRequestResponseConverter.java
@@ -37,7 +37,7 @@
 import io.reactivex.netty.client.ConnectionReuseEvent;
 import io.reactivex.netty.metrics.Clock;
 import io.reactivex.netty.metrics.MetricEventsSubject;
-import io.reactivex.netty.protocol.http.MultipleFutureListener;
+import io.reactivex.netty.util.MultipleFutureListener;
 import rx.Observable;
 import rx.Subscriber;
 import rx.subjects.PublishSubject;
diff --git a/rx-netty/src/main/java/io/reactivex/netty/protocol/http/server/HttpConnectionHandler.java b/rx-netty/src/main/java/io/reactivex/netty/protocol/http/server/HttpConnectionHandler.java
@@ -133,7 +133,7 @@ public void onNext(I i) {
                             public void onCompleted() {
                                 eventsSubject.onEvent(HttpServerMetricsEvent.REQUEST_HANDLING_SUCCESS,
                                                       Clock.onEndMillis(startTimeMillis));
-                                response.close();
+                                response.close(false);
                             }
 
                             @Override
@@ -143,7 +143,7 @@ public void onError(Throwable throwable) {
                                 if (!response.isHeaderWritten()) {
                                     responseGenerator.updateResponse(response, throwable);
                                 }
-                                response.close();
+                                response.close(false);
                             }
 
                             @Override
diff --git a/rx-netty/src/main/java/io/reactivex/netty/protocol/http/server/HttpServerResponse.java b/rx-netty/src/main/java/io/reactivex/netty/protocol/http/server/HttpServerResponse.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.buffer.ByteBuf;
@@ -85,7 +86,27 @@ public HttpResponseStatus getStatus() {
     }
 
     @Override
-    public Observable<Void> _close() {
+    public Observable<Void> close() {
+        return close(true);
+    }
+
+    /**
+     * Closes this response with optionally flushing the writes. <br/>
+     *
+     * <b>Unless it is required by the usecase, it is generally more optimal to leave the decision of when to flush to
+     * the framework as that enables a gathering write on the underlying socket, which is more optimal.</b>
+     *
+     * @param flush If this close should also flush the writes.
+     *
+     * @return Observable representing the close result.
+     */
+    @Override
+    public Observable<Void> close(boolean flush) {
+        return super.close(flush);
+    }
+
+    @Override
+    public Observable<Void> _close(boolean flush) {
 
         writeHeadersIfNotWritten();
 
@@ -94,7 +115,7 @@ public Observable<Void> _close() {
             // sent for keep-alive connections, netty's HTTP codec will not know that the response has ended and hence
             // will ignore the subsequent HTTP header writes. See issue: https://github.com/Netflix/RxNetty/issues/130
         }
-        return flush();
+        return flush ? flush() : Observable.<Void>empty();
     }
 
     HttpResponse getNettyResponse() {
diff --git a/rx-netty/src/main/java/io/reactivex/netty/protocol/http/server/ServerRequestResponseConverter.java b/rx-netty/src/main/java/io/reactivex/netty/protocol/http/server/ServerRequestResponseConverter.java
@@ -114,7 +114,12 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
         } else {
             super.write(ctx, msg, promise); // pass through, since we do not understand this message.
         }
+    }
 
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+        super.channelReadComplete(ctx);
+        ctx.pipeline().flush(); // If there is nothing to flush, this is a short-circuit in netty.
     }
 
     private void addWriteCompleteEvents(ChannelPromise promise, final long startTimeMillis,
diff --git a/rx-netty/src/main/java/io/reactivex/netty/util/MultipleFutureListener.java b/rx-netty/src/main/java/io/reactivex/netty/util/MultipleFutureListener.java
@@ -13,7 +13,8 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.reactivex.netty.protocol.http;
+
+package io.reactivex.netty.util;
 
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
@@ -43,6 +44,9 @@ public MultipleFutureListener(final ChannelPromise completionPromise) {
         completionObservable = Observable.create(new Observable.OnSubscribe<Void>() {
             @Override
             public void call(final Subscriber<? super Void> subscriber) {
+                if (listeningToCount.get() == 0) {
+                    MultipleFutureListener.this.completionPromise.trySuccess();
+                }
                 MultipleFutureListener.this.completionPromise.addListener(new ChannelFutureListener() {
                     @Override
                     public void operationComplete(ChannelFuture future) throws Exception {
diff --git a/rx-netty/src/main/java/io/reactivex/netty/util/NoOpSubscriber.java b/rx-netty/src/main/java/io/reactivex/netty/util/NoOpSubscriber.java
@@ -0,0 +1,39 @@
+/*
+ * 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.
+ */
+
+package io.reactivex.netty.util;
+
+import rx.Subscriber;
+
+/**
+ * A subscriber that does nothing.
+ *
+ * @author Nitesh Kant
+ */
+public class NoOpSubscriber<T> extends Subscriber<T> {
+
+    @Override
+    public void onCompleted() {
+    }
+
+    @Override
+    public void onError(Throwable e) {
+    }
+
+    @Override
+    public void onNext(T next) {
+    }
+}
diff --git a/rx-netty/src/test/java/io/reactivex/netty/NoOpChannelHandlerContext.java b/rx-netty/src/test/java/io/reactivex/netty/NoOpChannelHandlerContext.java
diff --git a/rx-netty/src/test/java/io/reactivex/netty/protocol/http/server/HttpServerTest.java b/rx-netty/src/test/java/io/reactivex/netty/protocol/http/server/HttpServerTest.java
diff --git a/rx-netty/src/test/java/io/reactivex/netty/util/MultipleFutureListenerTest.java b/rx-netty/src/test/java/io/reactivex/netty/util/MultipleFutureListenerTest.java

Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,7 @@ public HttpServer<ByteBuf, ByteBuf> createServer() {`
`46`	`46`	`public Observable<Void> handle(HttpServerRequest<ByteBuf> request, final HttpServerResponse<ByteBuf> response) {`
`47`	`47`	`printRequestHeader(request);`
`48`	`48`	`response.writeString("Welcome!!");`
`49`		`- return response.close();`
	`49`	`+ return response.close(false);`
`50`	`50`	`}`
`51`	`51`	`}).pipelineConfigurator(PipelineConfigurators.<ByteBuf, ByteBuf>httpServerConfigurator()).build();`
`52`	`52`
Original file line number	Diff line number	Diff line change
`@@ -59,7 +59,7 @@ public String call(String accumulator, String value) {`
`59`	`59`	`@Override`
`60`	`60`	`public Observable<Void> call(String clientMessage) {`
`61`	`61`	`response.writeString(clientMessage.toUpperCase());`
`62`		`- return response.close();`
	`62`	`+ return response.close(false);`
`63`	`63`	`}`
`64`	`64`	`});`
`65`	`65`	`}`
Original file line number	Diff line number	Diff line change
`@@ -43,7 +43,7 @@ public HttpServer<ByteBuf, ByteBuf> createServer() {`
`43`	`43`	`@Override`
`44`	`44`	`public Observable<Void> handle(HttpServerRequest<ByteBuf> request, final HttpServerResponse<ByteBuf> response) {`
`45`	`45`	`response.writeStringAndFlush("Welcome!!");`
`46`		`- return response.close();`
	`46`	`+ return response.close(false);`
`47`	`47`	`}`
`48`	`48`	`}).withSslEngineFactory(DefaultFactories.SELF_SIGNED).build();`
`49`	`49`
Original file line number	Diff line number	Diff line change
`@@ -60,7 +60,7 @@ public String call(ByteBuf content) {`
`60`	`60`	`@Override`
`61`	`61`	`public Observable<Void> call(Integer counter) {`
`62`	`62`	`response.writeString(counter.toString());`
`63`		`- return response.close();`
	`63`	`+ return response.close(false);`
`64`	`64`	`}`
`65`	`65`	`});`
`66`	`66`	`}`
Original file line number	Diff line number	Diff line change
`@@ -47,4 +47,8 @@ public interface ChannelWriter<O> {`
`47`	`47`	`Observable<Void> writeBytesAndFlush(byte[] msg);`
`48`	`48`
`49`	`49`	`Observable<Void> writeStringAndFlush(String msg);`
	`50`	`+`
	`51`	`+ Observable<Void> close();`
	`52`	`+`
	`53`	`+ Observable<Void> close(boolean flush);`
`50`	`54`	`}`
Original file line number	Diff line number	Diff line change
`@@ -133,7 +133,7 @@ public void onNext(I i) {`
`133`	`133`	`public void onCompleted() {`
`134`	`134`	`eventsSubject.onEvent(HttpServerMetricsEvent.REQUEST_HANDLING_SUCCESS,`
`135`	`135`	`Clock.onEndMillis(startTimeMillis));`
`136`		`- response.close();`
	`136`	`+ response.close(false);`
`137`	`137`	`}`
`138`	`138`
`139`	`139`	`@Override`
`@@ -143,7 +143,7 @@ public void onError(Throwable throwable) {`
`143`	`143`	`if (!response.isHeaderWritten()) {`
`144`	`144`	`responseGenerator.updateResponse(response, throwable);`
`145`	`145`	`}`
`146`		`- response.close();`
	`146`	`+ response.close(false);`
`147`	`147`	`}`
`148`	`148`
`149`	`149`	`@Override`