New 1.0 API: Tidy up Address outstanding issues. (#204)

components/api/src/main/java/com/hotels/styx/api/HttpRequest.java

@@ -21,14 +21,12 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.CompositeByteBuf;
 import rx.Observable;
-import rx.Subscriber;
 
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
 
 import static com.google.common.base.Objects.toStringHelper;
 import static com.google.common.base.Preconditions.checkArgument;
@@ -378,30 +376,6 @@ public String toString() {
                 .toString();
     }
 
-    // TODO: Mikko: Identical content to the HttpResponse one. Consider moving to base class.
-    public CompletableFuture<Boolean> releaseContentBuffers() {
-        CompletableFuture<Boolean> future = new CompletableFuture<>();
-
-        ((StyxCoreObservable<ByteBuf>) body).delegate().subscribe(new Subscriber<ByteBuf>() {
-            @Override
-            public void onCompleted() {
-                future.complete(true);
-            }
-
-            @Override
-            public void onError(Throwable e) {
-
-            }
-
-            @Override
-            public void onNext(ByteBuf byteBuf) {
-                byteBuf.release();
-            }
-        });
-
-        return future;
-    }
-
     /**
      * Builder.
      */

components/api/src/main/java/com/hotels/styx/api/HttpResponse.java

@@ -24,13 +24,11 @@
 import io.netty.buffer.CompositeByteBuf;
 import io.netty.util.ReferenceCountUtil;
 import rx.Observable;
-import rx.Subscriber;
 
 import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
 
 import static com.google.common.base.Objects.toStringHelper;
 import static com.google.common.base.Preconditions.checkArgument;
@@ -208,30 +206,6 @@ public boolean equals(Object obj) {
                 && Objects.equal(this.cookies, other.cookies);
     }
 
-    public CompletableFuture<Boolean> releaseContentBuffers() {
-        CompletableFuture<Boolean> future = new CompletableFuture<>();
-
-        ((StyxCoreObservable<ByteBuf>) body).delegate()
-                .subscribe(new Subscriber<ByteBuf>() {
-                    @Override
-                    public void onCompleted() {
-                        future.complete(true);
-                    }
-
-                    @Override
-                    public void onError(Throwable e) {
-
-                    }
-
-                    @Override
-                    public void onNext(ByteBuf byteBuf) {
-                        byteBuf.release();
-                    }
-                });
-
-        return future;
-    }
-
     /**
      * Builder.
      */
@@ -293,11 +267,10 @@ public Builder body(StyxObservable<ByteBuf> content) {
         }
 
         /**
-         * Sets the message body. As the content length is known, this header will also be set.
-         * <p>
-         * TODO: Mikko: Styx 2.0 API: Missing test:
+         * Sets the message body by encoding a {@link StyxObservable} of {@link String}s into bytes.
          *
          * @param contentObservable message body content.
+         * @param charset character set
          * @return {@code this}
          */
         public Builder body(StyxObservable<String> contentObservable, Charset charset) {
@@ -416,16 +389,11 @@ public Builder removeHeader(CharSequence name) {
         }
 
         /**
-         * Removes body of the request
-         * <p>
-         * TODO: Mikko: Styx 2.0 API: Ensure that reference counting works well with the new API.
-         * Most importantly it should be safe to use without consumers accidentally using the API
-         * in a dangerous way that might cause buffer leaks.
-         * <p>
-         * Especially when transforming a response to another, etc.
+         * Removes body of the request.
          *
-         * @return
+         * @return {@code this}
          */
+        // TODO: See https://github.com/HotelsDotCom/styx/issues/201
         public Builder removeBody() {
             Observable<ByteBuf> delegate = ((StyxCoreObservable<ByteBuf>) body)
                     .delegate()

components/api/src/main/java/com/hotels/styx/api/StreamingHttpMessage.java

@@ -17,9 +17,11 @@
 
 import com.hotels.styx.api.messages.HttpVersion;
 import io.netty.buffer.ByteBuf;
+import rx.Subscriber;
 
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 
 import static com.hotels.styx.api.HttpHeaderNames.CONTENT_LENGTH;
 import static com.hotels.styx.api.HttpHeaderNames.CONTENT_TYPE;
@@ -113,4 +115,28 @@ default Optional<String> contentType() {
     default boolean chunked() {
         return HttpMessageSupport.chunked(headers());
     }
+
+    default CompletableFuture<Boolean> releaseContentBuffers() {
+        CompletableFuture<Boolean> future = new CompletableFuture<>();
+
+        ((StyxCoreObservable<ByteBuf>) body()).delegate()
+                .subscribe(new Subscriber<ByteBuf>() {
+                    @Override
+                    public void onCompleted() {
+                        future.complete(true);
+                    }
+
+                    @Override
+                    public void onError(Throwable e) {
+                        future.completeExceptionally(e);
+                    }
+
+                    @Override
+                    public void onNext(ByteBuf byteBuf) {
+                        byteBuf.release();
+                    }
+                });
+
+        return future;
+    }
 }

components/api/src/main/java/com/hotels/styx/api/StyxCoreObservable.java

@@ -50,6 +50,10 @@ private static <T> Observable<T> toObservable(CompletionStage<T> future) {
                 }));
     }
 
+    public static <T> StyxObservable<T> empty() {
+        return new StyxCoreObservable<>(Observable.empty());
+    }
+
     public static <T> StyxObservable<T> of(T item) {
         return new StyxCoreObservable<T>(Observable.just(item));
     }

components/api/src/main/java/com/hotels/styx/api/StyxObservable.java

@@ -36,8 +36,6 @@ public interface StyxObservable<T> {
 
     <U> StyxObservable<U> reduce(BiFunction<T, U, U> accumulator, U initialValue);
 
-    // TODO: Mikko: Styx 2.0 Api: `onError`: is more flexible type signature possible? Such as:
-    //       <U> StyxObservable<U> onError(Function<Throwable, StyxObservable<U>> errorHandler);
     StyxObservable<T> onError(Function<Throwable, StyxObservable<T>> errorHandler);
 
     /**
@@ -64,11 +62,6 @@ static <T> StyxObservable<T> of(T value) {
         return new StyxCoreObservable<>(Observable.just(value));
     }
 
-    // TODO: Mikko: Only required for testing?
-    static <T> StyxObservable<T> empty() {
-        return new StyxCoreObservable<T>(Observable.empty());
-    }
-
     static <T> StyxObservable<T> from(Iterable<T> values) {
         return new StyxCoreObservable<>(Observable.from(values));
     }

components/api/src/main/java/com/hotels/styx/api/netty/CustomHttpResponseStatus.java → components/api/src/main/java/com/hotels/styx/api/messages/CustomHttpResponseStatus.java

@@ -13,9 +13,7 @@
   See the License for the specific language governing permissions and
   limitations under the License.
  */
-package com.hotels.styx.api.netty;
-
-import com.hotels.styx.api.messages.HttpResponseStatus;
+package com.hotels.styx.api.messages;
 
 /**
  * Custom HTTP response status codes.

components/api/src/main/java/com/hotels/styx/api/messages/HttpResponseStatus.java

@@ -117,9 +117,7 @@ public static HttpResponseStatus statusWithCode(int code) {
         return STATUSES[code];
     }
 
-    // TODO: Mikko: Styx 2.0 API: Had to relax visibility due to CustomHttpResponseStatus class.
-    // Check with Kyle.
-    public HttpResponseStatus(int code, String description) {
+    HttpResponseStatus(int code, String description) {
         this.code = code;
         this.description = description;
     }

components/api/src/test/java/com/hotels/styx/api/FullHttpRequestTest.java

@@ -249,24 +249,6 @@ public void requestBodyCannotBeChangedViaStreamingRequest() {
         assertThat(original.bodyAs(UTF_8), is("original"));
     }
 
-    // TODO: Mikko: Styx 2.0 API: Ought to move to HttpRequest class?
-    @Test(expectedExceptions = io.netty.util.IllegalReferenceCountException.class)
-    public void toFullReqestReleasesOriginalRefCountedBuffers() throws ExecutionException, InterruptedException {
-        ByteBuf content = Unpooled.copiedBuffer("original", UTF_8);
-
-        HttpRequest original = HttpRequest.get("/foo")
-                .body(StyxObservable.of(content))
-                .build();
-
-        FullHttpRequest fullRequest = original.toFullRequest(100)
-                .asCompletableFuture()
-                .get();
-
-        content.array()[0] = 'A';
-
-        assertThat(fullRequest.bodyAs(UTF_8), is("original"));
-    }
-
     @Test
     public void transformedBodyIsNewCopy() {
         FullHttpRequest request = get("/foo")

components/api/src/test/java/com/hotels/styx/api/HttpRequestTest.java

@@ -17,6 +17,7 @@
 
 import com.google.common.collect.ImmutableMap;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
@@ -81,6 +82,23 @@ public void decodesToFullHttpRequest() throws Exception {
         assertThat(full.body(), is(bytes("foobar")));
     }
 
+    @Test(expectedExceptions = io.netty.util.IllegalReferenceCountException.class)
+    public void toFullRequestReleasesOriginalReferenceCountedBuffers() throws ExecutionException, InterruptedException {
+        ByteBuf content = Unpooled.copiedBuffer("original", UTF_8);
+
+        HttpRequest original = HttpRequest.get("/foo")
+                .body(StyxObservable.of(content))
+                .build();
+
+        FullHttpRequest fullRequest = original.toFullRequest(100)
+                .asCompletableFuture()
+                .get();
+
+        content.array()[0] = 'A';
+
+        assertThat(fullRequest.bodyAs(UTF_8), is("original"));
+    }
+
     @Test(dataProvider = "emptyBodyRequests")
     public void encodesToStreamingHttpRequestWithEmptyBody(HttpRequest streamingRequest) throws Exception {
         FullHttpRequest full = streamingRequest.toFullRequest(0x1000)
@@ -95,7 +113,7 @@ public void encodesToStreamingHttpRequestWithEmptyBody(HttpRequest streamingRequ
     private Object[][] emptyBodyRequests() {
         return new Object[][]{
                 {get("/foo/bar").build()},
-                {post("/foo/bar", StyxObservable.empty()).build()},
+                {post("/foo/bar", StyxCoreObservable.empty()).build()},
         };
     }
 

components/api/src/test/java/com/hotels/styx/api/HttpResponseTest.java

@@ -22,6 +22,8 @@
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
 import java.util.stream.Stream;
 
 import static com.hotels.styx.api.HttpCookie.cookie;
@@ -45,7 +47,9 @@
 import static com.hotels.styx.api.messages.HttpVersion.HTTP_1_0;
 import static com.hotels.styx.api.messages.HttpVersion.HTTP_1_1;
 import static com.hotels.styx.support.matchers.IsOptional.isValue;
+import static java.nio.charset.StandardCharsets.UTF_16;
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.concurrent.TimeUnit.SECONDS;
 import static java.util.stream.Collectors.toList;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.contains;
@@ -91,7 +95,7 @@ public void encodesToFullHttpResponseWithEmptyBody(HttpResponse response) throws
     private Object[][] emptyBodyResponses() {
         return new Object[][]{
                 {response().build()},
-                {response().body(StyxObservable.empty()).build()},
+                {response().body(StyxCoreObservable.empty()).build()},
         };
     }
 
@@ -310,6 +314,28 @@ public void rejectsInvalidContentLength() {
                 .build();
     }
 
+    @Test
+    public void encodesBodyWithCharset() throws InterruptedException, ExecutionException, TimeoutException {
+        StyxObservable<String> o = StyxObservable.of("Hello, World!");
+
+        FullHttpResponse responseUtf8 = response()
+                .body(o, UTF_8)
+                .build()
+                .toFullResponse(1_000_000)
+                .asCompletableFuture()
+                .get(1, SECONDS);
+
+        FullHttpResponse responseUtf16 = response()
+                .body(o, UTF_16)
+                .build()
+                .toFullResponse(1_000_000)
+                .asCompletableFuture()
+                .get(1, SECONDS);
+
+        assertThat(responseUtf8.body(), is("Hello, World!".getBytes(UTF_8)));
+        assertThat(responseUtf16.body(), is("Hello, World!".getBytes(UTF_16)));
+    }
+
     private static HttpResponse.Builder response() {
         return HttpResponse.response();
     }

components/api/src/test/java/com/hotels/styx/api/MockContext.java

@@ -15,10 +15,6 @@
  */
 package com.hotels.styx.api;
 
-import java.util.concurrent.CompletableFuture;
-import java.util.function.BiFunction;
-import java.util.function.Function;
-
 public class MockContext implements HttpInterceptor.Context {
 
     public static final HttpInterceptor.Context MOCK_CONTEXT = new MockContext();
@@ -32,42 +28,4 @@ public void add(String key, Object value) {
     public <T> T get(String key, Class<T> clazz) {
         return null;
     }
-
-    // TODO: Mikko: Styx 2.0 API: MockObservable support for `onError`.
-    static class MockObservable<T> implements StyxObservable<T> {
-        private final T value;
-
-        <U> MockObservable(T value) {
-            this.value = value;
-        }
-
-        @Override
-        public <U> StyxObservable<U> map(Function<T, U> transformation) {
-            return new MockObservable<>(transformation.apply(value));
-        }
-
-        @Override
-        public <U> StyxObservable<U> flatMap(Function<T, StyxObservable<U>> transformation) {
-            return transformation.apply(value);
-        }
-
-        @Override
-        public <U> StyxObservable<U> reduce(BiFunction<T, U, U> accumulator, U initialValue) {
-            throw new UnsupportedOperationException();
-        }
-
-        @Override
-        public StyxObservable<T> onError(Function<Throwable, StyxObservable<T>> errorHandler) {
-            return new MockObservable<>(value);
-        }
-
-        @Override
-        public CompletableFuture<T> asCompletableFuture() {
-            return CompletableFuture.completedFuture(this.value);
-        }
-
-        public T value() {
-            return value;
-        }
-    }
 }