A couple of more tests

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

@@ -77,7 +77,7 @@ public Server listen(String host, int port, boolean secure) {
 
         try {
             logger.debug("Starting HTTP server on {}:{}", host, port);
-            Channel serverChannel = bootstrap.bind(host, port).sync().channel();
+            serverChannel = bootstrap.bind(host, port).sync().channel();
             logger.debug("HTTP server listening on port {}:{}", host, port);
         } catch (InterruptedException e) {
             logger.debug("Error occurred while HTTP server was listening on {}:{}", host, port, e);

core/src/main/java/org/chodavarapu/datamill/json/JsonObject.java

@@ -189,14 +189,21 @@ public byte asByte() {
 
         @Override
         public byte[] asByteArray() {
-            JSONArray array = object.getJSONArray(name);
-            if (array != null) {
-                byte[] bytes = new byte[array.length()];
-                for (int i = 0; i < bytes.length; i++) {
-                    bytes[i] = (byte) array.getInt(i);
+            try {
+                JSONArray array = object.getJSONArray(name);
+                if (array != null) {
+                    byte[] bytes = new byte[array.length()];
+                    for (int i = 0; i < bytes.length; i++) {
+                        bytes[i] = (byte) array.getInt(i);
+                    }
+
+                    return bytes;
+                }
+            } catch (JSONException e) {
+                String value = asString();
+                if (value != null) {
+                    return value.getBytes();
                 }
-
-                return bytes;
             }
 
             return null;

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

@@ -11,14 +11,14 @@
 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;
-import java.util.concurrent.TimeUnit;
 
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -30,6 +30,9 @@ public class ClientToServerChannelHandlerTest {
     @Mock
     private ChannelHandlerContext context;
 
+    @Captor
+    private ArgumentCaptor<LastHttpContent> lastContentCaptor;
+
     @Mock
     private Route route;
 
@@ -42,6 +45,18 @@ public class ClientToServerChannelHandlerTest {
     @Captor
     private ArgumentCaptor<FullHttpResponse> responseCaptor;
 
+    @Captor
+    private ArgumentCaptor<HttpObject> responseFragmentsCaptor;
+
+    private void waitForExecutorToFinishAllTasks(ExecutorService executor) throws Exception {
+        for (int i = 0; i < 5; i++) {
+            // We submit an empty task and wait for it so that other tasks submitted ahead of this get executed first
+            // and we wait for their completion. But each of those may have submitted others which we need to wait for
+            // as well which is why this is done in a loop.
+            executor.submit(() -> {}).get();
+        }
+    }
+
     @Test
     public void sendContinueResponseIfRequested() {
         ClientToServerChannelHandler handler = new ClientToServerChannelHandler(threadPool, route, null);
@@ -71,8 +86,7 @@ public void readEntitySentWithFullRequest() throws Exception {
 
         handler.channelRead(context, request);
 
-        service.shutdown();
-        service.awaitTermination(1, TimeUnit.SECONDS);
+        waitForExecutorToFinishAllTasks(service);
 
         verify(route).apply(requestCaptor.capture());
 
@@ -100,12 +114,51 @@ public void readEntitySentWithMultipleChunks() throws Exception {
         DefaultLastHttpContent last = new DefaultLastHttpContent();
         handler.channelRead(context, last);
 
-        service.shutdown();
-        service.awaitTermination(1, TimeUnit.SECONDS);
+        waitForExecutorToFinishAllTasks(service);
 
         verify(route).apply(requestCaptor.capture());
 
         ServerRequest appliedRequest = requestCaptor.getValue();
         assertEquals("Test Content Additional Content", appliedRequest.entity().asString().toBlocking().last());
     }
+
+    @Test
+    public void singleChunkResponseSent() throws Exception {
+        ExecutorService service = Executors.newSingleThreadExecutor();
+        ClientToServerChannelHandler handler = new ClientToServerChannelHandler(service, route, null);
+
+        DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "localhost");
+
+        when(route.apply(any())).thenReturn(new ResponseBuilderImpl().ok());
+
+        handler.channelRead(context, request);
+
+        waitForExecutorToFinishAllTasks(service);
+
+        verify(context).writeAndFlush(responseCaptor.capture());
+
+        assertEquals(HttpResponseStatus.OK, responseCaptor.getValue().status());
+    }
+
+    @Test
+    public void multipeResponseChunksSent() throws Exception {
+        ExecutorService service = Executors.newSingleThreadExecutor();
+        ClientToServerChannelHandler handler = new ClientToServerChannelHandler(service, route, null);
+
+        DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "localhost");
+
+        when(route.apply(any())).thenReturn(new ResponseBuilderImpl().ok("Test Content"));
+
+        handler.channelRead(context, request);
+
+        waitForExecutorToFinishAllTasks(service);
+
+        verify(context, times(2)).write(responseFragmentsCaptor.capture());
+        verify(context).writeAndFlush(lastContentCaptor.capture());
+
+        assertEquals(HttpResponseStatus.OK, ((HttpResponse) responseFragmentsCaptor.getAllValues().get(0)).status());
+        byte[] bytes = new byte[((HttpContent) responseFragmentsCaptor.getAllValues().get(1)).content().readableBytes()];
+        ((HttpContent) responseFragmentsCaptor.getAllValues().get(1)).content().readBytes(bytes);
+        assertArrayEquals("Test Content".getBytes(), bytes);
+    }
 }

core/src/test/java/org/chodavarapu/datamill/json/JsonObjectTest.java

@@ -6,9 +6,7 @@
 
 import java.util.function.Consumer;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.fail;
+import static org.junit.Assert.*;
 
 /**
  * @author Ravi Chodavarapu (rchodava@gmail.com)
@@ -36,6 +34,7 @@ public void unsupportedConversions() {
         testThrowsConversionException(j -> j.asDouble());
         testThrowsConversionException(j -> j.asFloat());
         testThrowsConversionException(j -> j.asInteger());
+        testThrowsConversionException(j -> j.asLocalDateTime());
         testThrowsConversionException(j -> j.asLong());
         testThrowsConversionException(j -> j.asShort());
 
@@ -46,6 +45,7 @@ public void unsupportedConversions() {
         assertFalse(testObject.isFloat());
         assertFalse(testObject.isInteger());
         assertFalse(testObject.isLong());
+        assertFalse(testObject.isNumeric());
         assertFalse(testObject.isShort());
         assertFalse(testObject.isString());
     }
@@ -64,7 +64,7 @@ public void stringConversion() throws Exception {
 
     @Test
     public void propertyConversions() {
-        JsonObject json = new JsonObject("{\"character\": \"v\", \"numeric\": 2, \"boolean\": true, \"string\": \"value\"}");
+        JsonObject json = new JsonObject("{\"character\": \"v\", \"numeric\": 2, \"boolean\": true, \"string\": \"value\", \"bytes\": [1, 2]}");
         assertEquals(true, json.get("boolean").asBoolean());
         assertEquals(2, json.get("numeric").asByte());
         assertEquals('v', json.get("character").asCharacter());
@@ -75,6 +75,8 @@ public void propertyConversions() {
         assertEquals(2l, json.get("numeric").asLong());
         assertEquals(2, json.get("numeric").asShort());
         assertEquals("value", json.get("string").asString());
+        assertArrayEquals("value".getBytes(), json.get("string").asByteArray());
+        assertArrayEquals(new byte[] {(byte) 1, (byte) 2}, json.get("bytes").asByteArray());
 
         testThrowsConversionException(json, j -> j.get("string").asCharacter());
     }