Merge pull request #189 from NiteshKant/master

NiteshKant · NiteshKant · commit 284a0391e517 · 2014-07-22T15:30:11.000-07:00
Fixes #187
diff --git a/rx-netty-examples/build.gradle b/rx-netty-examples/build.gradle
@@ -16,10 +16,6 @@
 
 
 
-
-
-
-
 sourceCompatibility = JavaVersion.VERSION_1_6
 targetCompatibility = JavaVersion.VERSION_1_6
 
diff --git a/rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/helloworld/HelloWorldClient.java b/rx-netty-examples/src/main/java/io/reactivex/netty/examples/http/helloworld/HelloWorldClient.java
@@ -17,16 +17,17 @@
 package io.reactivex.netty.examples.http.helloworld;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.handler.codec.http.HttpResponseStatus;
 import io.reactivex.netty.RxNetty;
+import io.reactivex.netty.protocol.http.client.FlatResponseOperator;
 import io.reactivex.netty.protocol.http.client.HttpClientResponse;
-import rx.Observable;
-import rx.functions.Action0;
+import io.reactivex.netty.protocol.http.client.ResponseHolder;
 import rx.functions.Func1;
-import rx.functions.Func2;
 
 import java.nio.charset.Charset;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import static io.reactivex.netty.examples.http.helloworld.HelloWorldServer.DEFAULT_PORT;
 
@@ -41,29 +42,18 @@ public HelloWorldClient(int port) {
         this.port = port;
     }
 
-    public HttpResponseStatus sendHelloRequest() {
-        HttpResponseStatus statusCode = RxNetty.createHttpGet("http://localhost:" + port + "/hello")
-                .mergeMap(new Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>>() {
-                    @Override
-                    public Observable<ByteBuf> call(HttpClientResponse<ByteBuf> response) {
-                        return response.getContent();
-                    }
-                }, new Func2<HttpClientResponse<ByteBuf>, ByteBuf, HttpResponseStatus>() {
-                    @Override
-                    public HttpResponseStatus call(HttpClientResponse<ByteBuf> response, ByteBuf content) {
-                        printResponseHeader(response);
-                        System.out.println(content.toString(Charset.defaultCharset()));
-                        return response.getStatus();
-                    }
-                })
-                .doOnTerminate(new Action0() {
-                    @Override
-                    public void call() {
-                        System.out.println("=======================");
-                    }
-                }).toBlocking().last();
-
-        return statusCode;
+    public HttpClientResponse<ByteBuf> sendHelloRequest() throws InterruptedException, ExecutionException, TimeoutException {
+        return RxNetty.createHttpGet("http://localhost:" + port + "/hello")
+               .lift(FlatResponseOperator.<ByteBuf>flatResponse())
+               .map(new Func1<ResponseHolder<ByteBuf>, HttpClientResponse<ByteBuf>>() {
+                   @Override
+                   public HttpClientResponse<ByteBuf> call(ResponseHolder<ByteBuf> holder) {
+                       printResponseHeader(holder.getResponse());
+                       System.out.println(holder.getContent().toString(Charset.defaultCharset()));
+                       System.out.println("========================");
+                       return holder.getResponse();
+                   }
+               }).toBlocking().toFuture().get(1, TimeUnit.MINUTES);
     }
 
     public void printResponseHeader(HttpClientResponse<ByteBuf> response) {
@@ -76,7 +66,7 @@ public void printResponseHeader(HttpClientResponse<ByteBuf> response) {
         }
     }
 
-    public static void main(String[] args) {
+    public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
         int port = DEFAULT_PORT;
         if (args.length > 0) {
             port = Integer.parseInt(args[0]);
diff --git a/rx-netty-examples/src/test/java/io/reactivex/netty/examples/http/cpuintensive/CpuIntensiveServerTest.java b/rx-netty-examples/src/test/java/io/reactivex/netty/examples/http/cpuintensive/CpuIntensiveServerTest.java
@@ -26,6 +26,9 @@
 import org.junit.Before;
 import org.junit.Test;
 
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+
 import static io.reactivex.netty.examples.http.cpuintensive.CPUIntensiveServer.DEFAULT_PORT;
 
 /**
@@ -47,9 +50,9 @@ public void stopServer() throws InterruptedException {
     }
 
     @Test
-    public void testRequestReplySequence() {
+    public void testRequestReplySequence() throws InterruptedException, ExecutionException, TimeoutException {
         HelloWorldClient client = new HelloWorldClient(DEFAULT_PORT); // The client is no different than hello world.
-        HttpResponseStatus statusCode = client.sendHelloRequest();
+        HttpResponseStatus statusCode = client.sendHelloRequest().getStatus();
         Assert.assertEquals(HttpResponseStatus.OK, statusCode);
     }
 }
diff --git a/rx-netty-examples/src/test/java/io/reactivex/netty/examples/http/helloworld/HelloWorldTest.java b/rx-netty-examples/src/test/java/io/reactivex/netty/examples/http/helloworld/HelloWorldTest.java
@@ -25,6 +25,9 @@
 import org.junit.Before;
 import org.junit.Test;
 
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+
 import static io.reactivex.netty.examples.http.helloworld.HelloWorldServer.DEFAULT_PORT;
 
 /**
@@ -46,9 +49,9 @@ public void stopServer() throws InterruptedException {
     }
 
     @Test
-    public void testRequestReplySequence() {
+    public void testRequestReplySequence() throws InterruptedException, ExecutionException, TimeoutException {
         HelloWorldClient client = new HelloWorldClient(DEFAULT_PORT);
-        HttpResponseStatus statusCode = client.sendHelloRequest();
+        HttpResponseStatus statusCode = client.sendHelloRequest().getStatus();
         Assert.assertEquals(HttpResponseStatus.OK, statusCode);
     }
 }
diff --git a/rx-netty-examples/src/test/java/io/reactivex/netty/examples/http/plaintext/PlainTextServerTest.java b/rx-netty-examples/src/test/java/io/reactivex/netty/examples/http/plaintext/PlainTextServerTest.java
@@ -26,6 +26,9 @@
 import org.junit.Before;
 import org.junit.Test;
 
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+
 import static io.reactivex.netty.examples.http.plaintext.PlainTextServer.DEFAULT_PORT;
 
 /**
@@ -47,9 +50,9 @@ public void stopServer() throws InterruptedException {
     }
 
     @Test
-    public void testRequestReplySequence() {
+    public void testRequestReplySequence() throws InterruptedException, ExecutionException, TimeoutException {
         HelloWorldClient client = new HelloWorldClient(DEFAULT_PORT); // The client is no different than hello world.
-        HttpResponseStatus statusCode = client.sendHelloRequest();
+        HttpResponseStatus statusCode = client.sendHelloRequest().getStatus();
         Assert.assertEquals(HttpResponseStatus.OK, statusCode);
     }
 }
diff --git a/rx-netty/build.gradle b/rx-netty/build.gradle
@@ -1,10 +1,3 @@
-apply plugin: 'osgi'
-apply plugin: 'groovy'
-
-sourceCompatibility = JavaVersion.VERSION_1_6
-targetCompatibility = JavaVersion.VERSION_1_6
-
-
 /*
  * Copyright 2014 Netflix, Inc.
  *
@@ -20,7 +13,13 @@ targetCompatibility = JavaVersion.VERSION_1_6
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-sourceSets.test.java.srcDir 'src/main/java'
+
+
+apply plugin: 'osgi'
+apply plugin: 'groovy'
+
+sourceCompatibility = JavaVersion.VERSION_1_6
+targetCompatibility = JavaVersion.VERSION_1_6
 
 tasks.withType(Javadoc).each {
     it.classpath = sourceSets.main.compileClasspath
diff --git a/rx-netty/src/main/java/io/reactivex/netty/protocol/http/client/FlatResponseOperator.java b/rx-netty/src/main/java/io/reactivex/netty/protocol/http/client/FlatResponseOperator.java
@@ -0,0 +1,64 @@
+/*
+ * 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.protocol.http.client;
+
+import rx.Observable;
+import rx.Subscriber;
+import rx.functions.Func1;
+
+/**
+ * An operator to be used for a source of {@link HttpClientResponse} containing aggregated responses i.e. which does not
+ * have multiple HTTP chunks. This operator simplifies the handling of such a responses by flattening the content
+ * {@link Observable} into a single element producing a {@link ResponseHolder} object.
+ * See <a href="https://github.com/Netflix/RxNetty/issues/187">related issue</a> for details.
+ *
+ * @author Nitesh Kant
+ */
+public class FlatResponseOperator<T>
+        implements Observable.Operator<ResponseHolder<T>, HttpClientResponse<T>> {
+
+    public static <T> FlatResponseOperator<T> flatResponse() {
+        return new FlatResponseOperator<T>();
+    }
+
+    @Override
+    public Subscriber<? super HttpClientResponse<T>> call(final Subscriber<? super ResponseHolder<T>> child) {
+        return new Subscriber<HttpClientResponse<T>>() {
+            @Override
+            public void onCompleted() {
+                // Content complete propagates to the child subscriber.
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                child.onError(e);
+            }
+
+            @Override
+            public void onNext(final HttpClientResponse<T> response) {
+                response.getContent()
+                        .take(1)
+                        .map(new Func1<T, ResponseHolder<T>>() {
+                            @Override
+                            public ResponseHolder<T> call(T t) {
+                                return new ResponseHolder<T>(response, t);
+                            }
+                        }).subscribe(child);
+            }
+        };
+    }
+}
diff --git a/rx-netty/src/main/java/io/reactivex/netty/protocol/http/client/ResponseHolder.java b/rx-netty/src/main/java/io/reactivex/netty/protocol/http/client/ResponseHolder.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.protocol.http.client;
+
+/**
+ * @author Nitesh Kant
+ */
+public class ResponseHolder<T> {
+
+    private final HttpClientResponse<T> response;
+    private final T content;
+
+    public ResponseHolder(HttpClientResponse<T> response, T content) {
+        this.response = response;
+        this.content = content;
+    }
+
+    public HttpClientResponse<T> getResponse() {
+        return response;
+    }
+
+    public T getContent() {
+        return content;
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -26,6 +26,9 @@`
`26`	`26`	`import org.junit.Before;`
`27`	`27`	`import org.junit.Test;`
`28`	`28`
	`29`	`+import java.util.concurrent.ExecutionException;`
	`30`	`+import java.util.concurrent.TimeoutException;`
	`31`	`+`
`29`	`32`	`import static io.reactivex.netty.examples.http.cpuintensive.CPUIntensiveServer.DEFAULT_PORT;`
`30`	`33`
`31`	`34`	`/**`
`@@ -47,9 +50,9 @@ public void stopServer() throws InterruptedException {`
`47`	`50`	`}`
`48`	`51`
`49`	`52`	`@Test`
`50`		`- public void testRequestReplySequence() {`
	`53`	`+ public void testRequestReplySequence() throws InterruptedException, ExecutionException, TimeoutException {`
`51`	`54`	`HelloWorldClient client = new HelloWorldClient(DEFAULT_PORT); // The client is no different than hello world.`
`52`		`- HttpResponseStatus statusCode = client.sendHelloRequest();`
	`55`	`+ HttpResponseStatus statusCode = client.sendHelloRequest().getStatus();`
`53`	`56`	`Assert.assertEquals(HttpResponseStatus.OK, statusCode);`
`54`	`57`	`}`
`55`	`58`	`}`
Original file line number	Diff line number	Diff line change
`@@ -25,6 +25,9 @@`
`25`	`25`	`import org.junit.Before;`
`26`	`26`	`import org.junit.Test;`
`27`	`27`
	`28`	`+import java.util.concurrent.ExecutionException;`
	`29`	`+import java.util.concurrent.TimeoutException;`
	`30`	`+`
`28`	`31`	`import static io.reactivex.netty.examples.http.helloworld.HelloWorldServer.DEFAULT_PORT;`
`29`	`32`
`30`	`33`	`/**`
`@@ -46,9 +49,9 @@ public void stopServer() throws InterruptedException {`
`46`	`49`	`}`
`47`	`50`
`48`	`51`	`@Test`
`49`		`- public void testRequestReplySequence() {`
	`52`	`+ public void testRequestReplySequence() throws InterruptedException, ExecutionException, TimeoutException {`
`50`	`53`	`HelloWorldClient client = new HelloWorldClient(DEFAULT_PORT);`
`51`		`- HttpResponseStatus statusCode = client.sendHelloRequest();`
	`54`	`+ HttpResponseStatus statusCode = client.sendHelloRequest().getStatus();`
`52`	`55`	`Assert.assertEquals(HttpResponseStatus.OK, statusCode);`
`53`	`56`	`}`
`54`	`57`	`}`