Merge pull request ReactiveX#435 from mattrjacobs/http-chunked

rx-apache-http recognizes "Transfer-Encoding: chunked" as an HTTP stream

rxjava-contrib/rxjava-apache-http/src/main/java/rx/apache/http/consumers/ResponseConsumerDelegate.java

@@ -17,6 +17,7 @@
 
 import java.io.IOException;
 
+import org.apache.http.Header;
 import org.apache.http.HttpEntity;
 import org.apache.http.HttpException;
 import org.apache.http.HttpResponse;
@@ -52,9 +53,7 @@ public ResponseConsumerDelegate(final Observer<? super ObservableHttpResponse> o
     @Override
     protected void onResponseReceived(HttpResponse response) throws HttpException, IOException {
         // when we receive the response with headers we evaluate what type of consumer we want
-        if (response.getFirstHeader("Content-Type").getValue().contains("text/event-stream")) {
-            // use 'contains' instead of equals since Content-Type can contain additional information
-            // such as charset ... see here: http://www.w3.org/International/O-HTTP-charset
+        if (responseIsStreamLike(response)) {
             consumer = new ResponseConsumerEventStream(observer, subscription);
         } else {
             consumer = new ResponseConsumerBasic(observer, subscription);
@@ -63,6 +62,20 @@ protected void onResponseReceived(HttpResponse response) throws HttpException, I
         consumer._onResponseReceived(response);
     }
 
+    private boolean responseIsStreamLike(HttpResponse response) {
+        final Header contentType = response.getFirstHeader("Content-Type");
+        // use 'contains' instead of equals since Content-Type can contain additional information
+        // such as charset ... see here: http://www.w3.org/International/O-HTTP-charset
+        if (contentType != null && contentType.getValue().contains("text/event-stream")) {
+            return true;
+        }
+        final Header transferEncoding = response.getFirstHeader("Transfer-Encoding");
+        if (transferEncoding != null && transferEncoding.getValue().equals("chunked")) {
+            return true;
+        }
+        return false;
+    }
+
     @Override
     protected void onContentReceived(ContentDecoder decoder, IOControl ioctrl) throws IOException {
         consumer._onContentReceived(decoder, ioctrl);