Merge pull request #429 from NiteshKant/0.4.x

Fixes #419

Author: NiteshKant

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

@@ -12,6 +12,7 @@
  * 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.protocol.http.client;
@@ -276,7 +277,7 @@ public void onCompleted() {
             public void onError(Throwable e) {
                 eventsSubject.onEvent(HttpClientMetricsEvent.REQUEST_CONTENT_SOURCE_ERROR, e);
                 promise.tryFailure(e);
-                rxRequest.onWriteComplete();
+                rxRequest.onWriteFailed(e);
             }
 
             @Override

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 Netflix, Inc.
+ * Copyright 2015 Netflix, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -12,6 +12,7 @@
  * 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.protocol.http.client;
@@ -29,6 +30,7 @@
 import io.reactivex.netty.channel.ContentTransformer;
 import rx.Observable;
 import rx.functions.Action0;
+import rx.functions.Action1;
 import rx.functions.Func1;
 
 import java.nio.charset.Charset;
@@ -44,6 +46,7 @@ public class HttpClientRequest<T> {
     private Observable<ByteBuf> rawContentSource;
     private String absoluteUri;
     private Action0 onWriteCompleteAction;
+    private Action1<Throwable> onWriteFailedAction;
 
     HttpClientRequest(HttpRequest nettyRequest) {
         this.nettyRequest = nettyRequest;
@@ -196,12 +199,22 @@ void doOnWriteComplete(Action0 onWriteCompleteAction) {
         this.onWriteCompleteAction = onWriteCompleteAction;
     }
 
+    void doOnWriteFailed(Action1<Throwable> onWriteFailedAction) {
+        this.onWriteFailedAction = onWriteFailedAction;
+    }
+
     void onWriteComplete() {
         if (null != onWriteCompleteAction) {
             onWriteCompleteAction.call();
         }
     }
 
+    void onWriteFailed(Throwable throwable) {
+        if (null != onWriteFailedAction) {
+            onWriteFailedAction.call(throwable);
+        }
+    }
+
     /*Set by HttpClient*/void setDynamicUriParts(String host, int port, boolean secure) {
         absoluteUri = secure ? "https" : "http" + "://" + host + ':' + port; // Uri in netty always starts with a slash
     }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 Netflix, Inc.
+ * Copyright 2015 Netflix, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -12,6 +12,7 @@
  * 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.protocol.http.client;
@@ -89,6 +90,16 @@ public void call(Throwable throwable) {
                                              }
                                          })
                                          .subscribe(child)); //subscribe the child for response.
+
+                        request.doOnWriteFailed(new Action1<Throwable>() {
+                            @Override
+                            public void call(Throwable throwable) {
+                                eventsSubject.onEvent(HttpClientMetricsEvent.REQUEST_WRITE_FAILED,
+                                                      Clock.onEndMillis(startTimeMillis), throwable);
+                                child.onError(throwable);
+                            }
+                        });
+
                         request.doOnWriteComplete(new Action0() {
                             @Override
                             public void call() {

rxnetty/src/test/java/io/reactivex/netty/protocol/http/client/HttpClientTest.java

@@ -12,6 +12,7 @@
  * 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.protocol.http.client;
@@ -45,6 +46,7 @@
 import rx.functions.Action0;
 import rx.functions.Action1;
 import rx.functions.Func1;
+import rx.observers.TestSubscriber;
 
 import java.net.ConnectException;
 import java.nio.charset.Charset;
@@ -119,6 +121,25 @@ public void testChunkedStreaming() throws Exception {
         assertEquals(RequestProcessor.smallStreamContent, result);
     }
 
+    @Test(timeout = 60000)
+    public void testWriteWithErrorContent() throws Exception {
+        HttpClient<ByteBuf, ByteBuf> client = RxNetty.createHttpClient("localhost", port);
+        Observable<ByteBuf> errSource = Observable.error(new NullPointerException());
+
+        HttpClientRequest<ByteBuf> request = HttpClientRequest.createPost("/")
+                                                              .withContentSource(errSource);
+
+        Observable<HttpClientResponse<ByteBuf>> response = client.submit(request);
+
+        final List<String> result = new ArrayList<String>();
+
+        TestSubscriber<HttpClientResponse<ByteBuf>> testSub = new TestSubscriber<HttpClientResponse<ByteBuf>>();
+        response.subscribe(testSub);
+
+        testSub.awaitTerminalEvent();
+        testSub.assertError(NullPointerException.class);
+    }
+
     @Test
     public void testMultipleChunks() throws Exception {
         HttpClient<ByteBuf, ServerSentEvent> client = RxNetty.createHttpClient("localhost", port,