Response builder should just return responses built instead of Observables

core/src/main/java/org/chodavarapu/datamill/http/ResponseBuilder.java

@@ -1,21 +1,19 @@
 package org.chodavarapu.datamill.http;
 
-import rx.Observable;
-
 /**
  * @author Ravi Chodavarapu (rchodava@gmail.com)
  */
 public interface ResponseBuilder {
-    Observable<Response> badRequest();
-    Observable<Response> badRequest(String content);
+    Response badRequest();
+    Response badRequest(String content);
     <T> ResponseBuilder header(String name, T value);
-    Observable<Response> internalServerError();
-    Observable<Response> internalServerError(String content);
-    Observable<Response> noContent();
-    Observable<Response> notFound();
-    Observable<Response> ok();
-    Observable<Response> ok(String content);
-    Observable<Response> ok(byte[] content);
-    Observable<Response> unauthorized();
-    Observable<Response> unauthorized(String content);
+    Response internalServerError();
+    Response internalServerError(String content);
+    Response noContent();
+    Response notFound();
+    Response ok();
+    Response ok(String content);
+    Response ok(byte[] content);
+    Response unauthorized();
+    Response unauthorized(String content);
 }

core/src/main/java/org/chodavarapu/datamill/http/ServerRequest.java

@@ -2,13 +2,16 @@
 
 import com.google.common.collect.Multimap;
 import org.chodavarapu.datamill.values.Value;
+import rx.Observable;
+
+import java.util.function.Function;
 
 /**
  * @author Ravi Chodavarapu (rchodava@gmail.com)
  */
 public interface ServerRequest extends Request {
     Value firstTrailingHeader(String header);
     Value firstTrailingHeader(RequestHeader header);
-    ResponseBuilder respond();
+    Observable<Response> respond(Function<ResponseBuilder, Response> responseBuilder);
     Multimap<String, String> trailingHeaders();
 }

core/src/main/java/org/chodavarapu/datamill/http/impl/MatcherBasedRoute.java

@@ -33,11 +33,11 @@ public Observable<Response> apply(ServerRequest request) {
             }
 
             if (availableMethods.size() > 0) {
-                return request.respond()
-                        .header("Allow", Joiner.on(',').join(availableMethods))
+                return request.respond(b ->
+                        b.header("Allow", Joiner.on(',').join(availableMethods))
                         .header("Access-Control-Allow-Headers", "Authorization")
                         .header("Access-Control-Allow-Origin", "*")
-                        .ok();
+                        .ok());
             }
         }
 

core/src/main/java/org/chodavarapu/datamill/http/impl/ResponseBuilderImpl.java

@@ -6,7 +6,6 @@
 import org.chodavarapu.datamill.http.ResponseBuilder;
 import org.chodavarapu.datamill.http.Status;
 import org.chodavarapu.datamill.values.StringValue;
-import rx.Observable;
 
 /**
  * @author Ravi Chodavarapu (rchodava@gmail.com)
@@ -15,13 +14,13 @@ public class ResponseBuilderImpl implements ResponseBuilder {
     private final Multimap<String, String> headers = LinkedListMultimap.create();
 
     @Override
-    public Observable<Response> badRequest() {
-        return Observable.just(new ResponseImpl(Status.BAD_REQUEST, headers));
+    public Response badRequest() {
+        return new ResponseImpl(Status.BAD_REQUEST, headers);
     }
 
     @Override
-    public Observable<Response> badRequest(String content) {
-        return Observable.just(new ResponseImpl(Status.BAD_REQUEST, headers, new ValueEntity(new StringValue(content))));
+    public Response badRequest(String content) {
+        return new ResponseImpl(Status.BAD_REQUEST, headers, new ValueEntity(new StringValue(content)));
     }
 
     @Override
@@ -31,47 +30,47 @@ public <T> ResponseBuilder header(String name, T value) {
     }
 
     @Override
-    public Observable<Response> internalServerError() {
-        return Observable.just(new ResponseImpl(Status.INTERNAL_SERVER_ERROR, headers));
+    public Response internalServerError() {
+        return new ResponseImpl(Status.INTERNAL_SERVER_ERROR, headers);
     }
 
     @Override
-    public Observable<Response> internalServerError(String content) {
-        return Observable.just(new ResponseImpl(Status.INTERNAL_SERVER_ERROR, headers, new ValueEntity(new StringValue(content))));
+    public Response internalServerError(String content) {
+        return new ResponseImpl(Status.INTERNAL_SERVER_ERROR, headers, new ValueEntity(new StringValue(content)));
     }
 
     @Override
-    public Observable<Response> noContent() {
-        return Observable.just(new ResponseImpl(Status.NO_CONTENT, headers));
+    public Response noContent() {
+        return new ResponseImpl(Status.NO_CONTENT, headers);
     }
 
     @Override
-    public Observable<Response> notFound() {
-        return Observable.just(new ResponseImpl(Status.NOT_FOUND, headers));
+    public Response notFound() {
+        return new ResponseImpl(Status.NOT_FOUND, headers);
     }
 
     @Override
-    public Observable<Response> ok() {
-        return Observable.just(new ResponseImpl(Status.OK, headers));
+    public Response ok() {
+        return new ResponseImpl(Status.OK, headers);
     }
 
     @Override
-    public Observable<Response> ok(String content) {
-        return Observable.just(new ResponseImpl(Status.OK, headers, new ValueEntity(new StringValue(content))));
+    public Response ok(String content) {
+        return new ResponseImpl(Status.OK, headers, new ValueEntity(new StringValue(content)));
     }
 
     @Override
-    public Observable<Response> ok(byte[] content) {
-        return Observable.just(new ResponseImpl(Status.OK, headers, new BytesEntity(content)));
+    public Response ok(byte[] content) {
+        return new ResponseImpl(Status.OK, headers, new BytesEntity(content));
     }
 
     @Override
-    public Observable<Response> unauthorized() {
-        return Observable.just(new ResponseImpl(Status.UNAUTHORIZED, headers));
+    public Response unauthorized() {
+        return new ResponseImpl(Status.UNAUTHORIZED, headers);
     }
 
     @Override
-    public Observable<Response> unauthorized(String content) {
-        return Observable.just(new ResponseImpl(Status.UNAUTHORIZED, headers, new ValueEntity(new StringValue(content))));
+    public Response unauthorized(String content) {
+        return new ResponseImpl(Status.UNAUTHORIZED, headers, new ValueEntity(new StringValue(content)));
     }
 }

core/src/main/java/org/chodavarapu/datamill/http/impl/ServerRequestImpl.java

@@ -5,9 +5,12 @@
 import io.netty.handler.codec.http.QueryStringDecoder;
 import org.chodavarapu.datamill.http.*;
 import org.chodavarapu.datamill.values.Value;
+import rx.*;
+import rx.Observable;
 
 import java.nio.charset.Charset;
 import java.util.*;
+import java.util.function.Function;
 
 /**
  * @author Ravi Chodavarapu (rchodava@gmail.com)
@@ -73,8 +76,8 @@ public Map<String, Object> options() {
     }
 
     @Override
-    public ResponseBuilder respond() {
-        return new ResponseBuilderImpl();
+    public rx.Observable<Response> respond(Function<ResponseBuilder, Response> responseBuilder) {
+        return Observable.just(responseBuilder.apply(new ResponseBuilderImpl()));
     }
 
     public void setTrailingHeaders(Multimap<String, String> trailingHeaders) {

core/src/test/java/org/chodavarapu/datamill/http/impl/BeanMethodMatcherTest.java

@@ -22,37 +22,37 @@ public class BeanMethodMatcherTest {
     private static class TestBean {
         @DELETE
         public Observable<Response> delete() {
-            return new ResponseBuilderImpl().ok("DELETE");
+            return Observable.just(new ResponseBuilderImpl().ok("DELETE"));
         }
 
         @GET
         public Observable<Response> get() {
-            return new ResponseBuilderImpl().ok("GET");
+            return Observable.just(new ResponseBuilderImpl().ok("GET"));
         }
 
         @HEAD
         public Observable<Response> head() {
-            return new ResponseBuilderImpl().ok("HEAD");
+            return Observable.just(new ResponseBuilderImpl().ok("HEAD"));
         }
 
         @OPTIONS
         public Observable<Response> options() {
-            return new ResponseBuilderImpl().ok("OPTIONS");
+            return Observable.just(new ResponseBuilderImpl().ok("OPTIONS"));
         }
 
         @PATCH
         public Observable<Response> patch() {
-            return new ResponseBuilderImpl().ok("PATCH");
+            return Observable.just(new ResponseBuilderImpl().ok("PATCH"));
         }
 
         @POST
         public Observable<Response> post() {
-            return new ResponseBuilderImpl().ok("POST");
+            return Observable.just(new ResponseBuilderImpl().ok("POST"));
         }
 
         @PUT
         public Observable<Response> put() {
-            return new ResponseBuilderImpl().ok("PUT");
+            return Observable.just(new ResponseBuilderImpl().ok("PUT"));
         }
     }
 
@@ -61,58 +61,58 @@ private static class TestBeanWithPaths {
         @Path("delete")
         @DELETE
         public Observable<Response> delete() {
-            return new ResponseBuilderImpl().ok("DELETE");
+            return Observable.just(new ResponseBuilderImpl().ok("DELETE"));
         }
 
         @Path("get/")
         @GET
         public Observable<Response> get() {
-            return new ResponseBuilderImpl().ok("GET");
+            return Observable.just(new ResponseBuilderImpl().ok("GET"));
         }
 
         @Path("/head/")
         @HEAD
         public Observable<Response> head() {
-            return new ResponseBuilderImpl().ok("HEAD");
+            return Observable.just(new ResponseBuilderImpl().ok("HEAD"));
         }
 
         @Path("/options")
         @OPTIONS
         public Observable<Response> options() {
-            return new ResponseBuilderImpl().ok("OPTIONS");
+            return Observable.just(new ResponseBuilderImpl().ok("OPTIONS"));
         }
 
         @Path("")
         @PATCH
         public Observable<Response> patch() {
-            return new ResponseBuilderImpl().ok("PATCH");
+            return Observable.just(new ResponseBuilderImpl().ok("PATCH"));
         }
 
         @Path("/")
         @POST
         public Observable<Response> post() {
-            return new ResponseBuilderImpl().ok("POST");
+            return Observable.just(new ResponseBuilderImpl().ok("POST"));
         }
     }
 
     private static class TestBeanWithOnlyMethodPaths {
         @POST
         @Path("/post")
         public Observable<Response> post() {
-            return new ResponseBuilderImpl().ok("POST");
+            return Observable.just(new ResponseBuilderImpl().ok("POST"));
         }
     }
 
     @Path("/test")
     private static class TestBeanWithOnlyBeanPath {
         @POST
         public Observable<Response> post() {
-            return new ResponseBuilderImpl().ok("POST");
+            return Observable.just(new ResponseBuilderImpl().ok("POST"));
         }
 
         @GET
         public Observable<Response> get() {
-            return new ResponseBuilderImpl().ok("GET");
+            return Observable.just(new ResponseBuilderImpl().ok("GET"));
         }
     }
 

core/src/test/java/org/chodavarapu/datamill/http/impl/ClientToServerChannelHandlerTest.java

@@ -11,6 +11,7 @@
 import org.mockito.Captor;
 import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
+import rx.Observable;
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -82,7 +83,7 @@ public void readEntitySentWithFullRequest() throws Exception {
         DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "localhost");
         request.content().writeBytes("Test Content".getBytes());
 
-        when(route.apply(any())).thenReturn(new ResponseBuilderImpl().ok());
+        when(route.apply(any())).thenReturn(Observable.just(new ResponseBuilderImpl().ok()));
 
         handler.channelRead(context, request);
 
@@ -96,7 +97,7 @@ public void readEntitySentWithFullRequest() throws Exception {
 
     @Test
     public void readEntitySentWithMultipleChunks() throws Exception {
-        when(route.apply(any())).thenReturn(new ResponseBuilderImpl().ok());
+        when(route.apply(any())).thenReturn(Observable.just(new ResponseBuilderImpl().ok()));
 
         ExecutorService service = Executors.newSingleThreadExecutor();
 
@@ -129,7 +130,7 @@ public void singleChunkResponseSent() throws Exception {
 
         DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "localhost");
 
-        when(route.apply(any())).thenReturn(new ResponseBuilderImpl().ok());
+        when(route.apply(any())).thenReturn(Observable.just(new ResponseBuilderImpl().ok()));
 
         handler.channelRead(context, request);
 
@@ -147,7 +148,7 @@ public void multipeResponseChunksSent() throws Exception {
 
         DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "localhost");
 
-        when(route.apply(any())).thenReturn(new ResponseBuilderImpl().ok("Test Content"));
+        when(route.apply(any())).thenReturn(Observable.just(new ResponseBuilderImpl().ok("Test Content")));
 
         handler.channelRead(context, request);
 

core/src/test/java/org/chodavarapu/datamill/http/impl/ResponseBuilderTest.java

@@ -13,19 +13,19 @@ public class ResponseBuilderTest {
     @Test
     public void responseCodes() {
         ResponseBuilder builder = new ResponseBuilderImpl();
-        assertEquals(Status.BAD_REQUEST, builder.badRequest().toBlocking().last().status());
-        assertEquals(Status.BAD_REQUEST, builder.badRequest("Content").toBlocking().last().status());
-        assertEquals("Content", builder.badRequest("Content").toBlocking().last().entity().asString().toBlocking().last());
-        assertEquals(Status.INTERNAL_SERVER_ERROR, builder.internalServerError().toBlocking().last().status());
-        assertEquals(Status.INTERNAL_SERVER_ERROR, builder.internalServerError("Content").toBlocking().last().status());
-        assertEquals("Content", builder.internalServerError("Content").toBlocking().last().entity().asString().toBlocking().last());
-        assertEquals(Status.NOT_FOUND, builder.notFound().toBlocking().last().status());
-        assertEquals(Status.OK, builder.ok().toBlocking().last().status());
-        assertEquals(Status.OK, builder.ok("Content").toBlocking().last().status());
-        assertEquals("Content", builder.ok("Content").toBlocking().last().entity().asString().toBlocking().last());
-        assertEquals(Status.UNAUTHORIZED, builder.unauthorized().toBlocking().last().status());
-        assertEquals(Status.UNAUTHORIZED, builder.unauthorized("Content").toBlocking().last().status());
-        assertEquals("Content", builder.unauthorized("Content").toBlocking().last().entity().asString().toBlocking().last());
-        assertEquals(Status.NO_CONTENT, builder.noContent().toBlocking().last().status());
+        assertEquals(Status.BAD_REQUEST, builder.badRequest().status());
+        assertEquals(Status.BAD_REQUEST, builder.badRequest("Content").status());
+        assertEquals("Content", builder.badRequest("Content").entity().asString().toBlocking().last());
+        assertEquals(Status.INTERNAL_SERVER_ERROR, builder.internalServerError().status());
+        assertEquals(Status.INTERNAL_SERVER_ERROR, builder.internalServerError("Content").status());
+        assertEquals("Content", builder.internalServerError("Content").entity().asString().toBlocking().last());
+        assertEquals(Status.NOT_FOUND, builder.notFound().status());
+        assertEquals(Status.OK, builder.ok().status());
+        assertEquals(Status.OK, builder.ok("Content").status());
+        assertEquals("Content", builder.ok("Content").entity().asString().toBlocking().last());
+        assertEquals(Status.UNAUTHORIZED, builder.unauthorized().status());
+        assertEquals(Status.UNAUTHORIZED, builder.unauthorized("Content").status());
+        assertEquals("Content", builder.unauthorized("Content").entity().asString().toBlocking().last());
+        assertEquals(Status.NO_CONTENT, builder.noContent().status());
     }
 }