deserialize messages of the form "clazzname=JSON...extra..." found in…

… a stream.

this will allow us to stream binary data over the websocket but be able to pass information about the binary data
in the same form of JSON messages as before

hawkular-bus-common/src/test/java/org/hawkular/bus/common/BasicMessageTest.java

@@ -131,6 +131,7 @@ public void testUnmodifiableDetails() {
         // make sure it didn't change and its still the same
         assertEquals("val1", msg.getDetails().get("key1"));
     }
+
     @Test
     public void testReadFromInputStream() throws IOException {
         // tests that this can extract the JSON even if more data follows in the stream

hawkular-feed-comm/feed-comm-api/src/main/java/org/hawkular/feedcomm/api/ApiDeserializer.java

@@ -16,6 +16,10 @@
  */
 package org.hawkular.feedcomm.api;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+
 import org.hawkular.bus.common.BasicMessage;
 
 /**
@@ -51,6 +55,13 @@ private static String[] fromHawkularFormat(String msg) {
     public ApiDeserializer() {
     }
 
+    /**
+     * Deserializes a JSON string in {@link #toHawkularFormat(BasicMessage) Hawkular format}.
+     * The JSON object is returned.
+     *
+     * @param nameAndJson the string to be deserialized
+     * @return the object represented by the JSON
+     */
     public <T extends BasicMessage> T deserialize(String nameAndJson) {
         String[] nameAndJsonArray = fromHawkularFormat(nameAndJson);
         String name = nameAndJsonArray[0];
@@ -69,4 +80,56 @@ public <T extends BasicMessage> T deserialize(String nameAndJson) {
             throw new RuntimeException("Cannot deserialize: [" + nameAndJson + "]", e);
         }
     }
+
+    /**
+     * Reads a JSON string in {@link #toHawkularFormat(BasicMessage) Hawkular format} that
+     * is found in the given input stream and converts the JSON string to a particular message object.
+     * The input stream will remain open so the caller can stream any extra data that might appear after it.
+     *
+     * Because of the way the JSON parser works, some extra data might have been read from the stream
+     * that wasn't part of the JSON message. In that case, a non-empty byte array containing the extra read
+     * data is returned in the map value.
+     *
+     * @param in input stream that has the Hawkular formatted JSON string at the head.
+     *
+     * @return a single-entry map whose key is the message object that was represented by the JSON string found
+     *         in the stream. The value of the map is a byte array containing extra data that was read from
+     *         the stream but not part of the JSON message.
+     */
+    public <T extends BasicMessage> Map<T, byte[]> deserialize(InputStream input) {
+        // We know the format is "name=json" with possible extra data after it.
+        // So first find the "name"
+        StringBuilder nameBuilder = new StringBuilder();
+        boolean foundSeparator = false;
+        while (!foundSeparator) {
+            int currentChar;
+            try {
+                currentChar = input.read();
+            } catch (IOException ioe) {
+                throw new RuntimeException("Cannot deserialize stream due to read error", ioe);
+            }
+            if (currentChar == -1) {
+                throw new RuntimeException("Cannot deserialize stream - doesn't look valid");
+            } else if (currentChar == '=') {
+                foundSeparator = true;
+            } else {
+                nameBuilder.append((char) currentChar);
+            }
+        }
+
+        // The name is the actual name of the POJO that is used to deserialize the JSON.
+        // If not fully qualified with a package then assume it is in our package.
+        String name = nameBuilder.toString();
+        if (name.indexOf(".") == -1) {
+            name = String.format("%s.%s", API_PKG, name);
+        }
+
+        // We now have the name and the input stream is pointing at the JSON
+        try {
+            Class<T> pojo = (Class<T>) Class.forName(name);
+            return BasicMessage.fromJSON(input, pojo);
+        } catch (Exception e) {
+            throw new RuntimeException("Cannot deserialize stream with object [" + name + "]", e);
+        }
+    }
 }

hawkular-feed-comm/feed-comm-api/src/test/java/org/hawkular/feedcomm/api/ApiDeserializerTest.java

@@ -16,7 +16,10 @@
  */
 package org.hawkular.feedcomm.api;
 
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
 import java.util.HashMap;
+import java.util.Map;
 
 import org.hawkular.bus.common.BasicMessage;
 import org.junit.Assert;
@@ -106,6 +109,60 @@ public void testEchoResponse() {
         Assert.assertEquals(pojo.getReply(), newpojo.getReply());
     }
 
+    @Test
+    public void testReadFromInputStreamWithExtraData() throws IOException {
+        // tests that this can extract the JSON even if more data follows in the stream
+        ApiDeserializer ad = new ApiDeserializer();
+
+        String nameAndJson = EchoRequest.class.getName() + "={\"echoMessage\":\"msg\"}";
+        String extra = "This is some extra data";
+        String nameAndJsonPlusExtra = nameAndJson + extra;
+
+        ByteArrayInputStream in = new UncloseableByteArrayInputStream(nameAndJsonPlusExtra.getBytes());
+
+        Map<BasicMessage, byte[]> map = ad.deserialize(in);
+        BasicMessage request = map.keySet().iterator().next();
+        Assert.assertTrue(request instanceof EchoRequest);
+        EchoRequest echoRequest = (EchoRequest) request;
+        Assert.assertEquals("msg", echoRequest.getEchoMessage());
+
+        // now make sure the stream still has our extra data that we can read now
+        byte[] leftoverFromJsonParser = map.values().iterator().next();
+        byte[] leftoverFromStream = new byte[in.available()];
+        in.read(leftoverFromStream);
+
+        String totalRemaining = new String(leftoverFromJsonParser, "UTF-8") + new String(leftoverFromStream, "UTF-8");
+        Assert.assertEquals(extra.length(), totalRemaining.length());
+        Assert.assertEquals(extra, totalRemaining);
+
+        // as a quick test, show that an exception results if we give bogus data in the input stream
+        in = new UncloseableByteArrayInputStream("this is not valid data".getBytes());
+        try {
+            ad.deserialize(in);
+            Assert.fail("Should have thrown an exception - the stream had invalid data");
+        } catch (Exception expected) {
+        }
+    }
+
+    @Test
+    public void testReadFromInputStreamWithNoExtraData() throws IOException {
+        ApiDeserializer ad = new ApiDeserializer();
+
+        String nameAndJson = EchoRequest.class.getName() + "={\"echoMessage\":\"msg\"}";
+        ByteArrayInputStream in = new UncloseableByteArrayInputStream(nameAndJson.getBytes());
+
+        Map<BasicMessage, byte[]> map = ad.deserialize(in);
+        BasicMessage request = map.keySet().iterator().next();
+        Assert.assertTrue(request instanceof EchoRequest);
+        EchoRequest echoRequest = (EchoRequest) request;
+        Assert.assertEquals("msg", echoRequest.getEchoMessage());
+
+        // now make sure the stream is empty
+        byte[] leftoverFromJsonParser = map.values().iterator().next();
+        Assert.assertEquals(0, leftoverFromJsonParser.length);
+        Assert.assertEquals(0, in.available());
+    }
+
     // takes a POJO, gets its JSON, then deserializes that JSON back into a POJO.
     private <T extends BasicMessage> T testSpecificPojo(T pojo) {
         String nameAndJson = String.format("%s=%s", pojo.getClass().getSimpleName(), pojo.toJSON());
@@ -115,4 +172,17 @@ private <T extends BasicMessage> T testSpecificPojo(T pojo) {
         Assert.assertNotSame(pojo, results); // just sanity check
         return results;
     }
+
+    // This is just to test that our JsonParser does NOT close the stream.
+    // If close is called, that is bad and should fail the test
+    class UncloseableByteArrayInputStream extends ByteArrayInputStream {
+        public UncloseableByteArrayInputStream(byte[] buf) {
+            super(buf);
+        }
+
+        @Override
+        public void close() throws IOException {
+            Assert.fail("The input stream should NOT have been closed");
+        }
+    }
 }