wrap byte arr/stream in BinaryData input stream wrapper so its not a …

…pain to work with parsing JSON and binary data coming over the wire

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

@@ -18,6 +18,7 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Collections;
 import java.util.Map;
 
 import org.hawkular.bus.common.BasicMessage;
@@ -88,15 +89,15 @@ public <T extends BasicMessage> T deserialize(String nameAndJson) {
      *
      * 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.
+     * data is returned in BinaryData along with the rest of the stream.
      *
      * @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.
+     *         in the stream. The value of the map is the extra binary data that is part of the stream
+     *         but not part of the JSON message.
      */
-    public <T extends BasicMessage> Map<T, byte[]> deserialize(InputStream input) {
+    public <T extends BasicMessage> Map<T, BinaryData> 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();
@@ -127,7 +128,10 @@ public <T extends BasicMessage> Map<T, byte[]> deserialize(InputStream input) {
         // 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);
+            Map<T, byte[]> results = BasicMessage.fromJSON(input, pojo);
+            return Collections.singletonMap(
+                    results.keySet().iterator().next(),
+                    new BinaryData(results.values().iterator().next(), input));
         } catch (Exception e) {
             throw new RuntimeException("Cannot deserialize stream with object [" + name + "]", e);
         }

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

@@ -0,0 +1,104 @@
+/*
+ * Copyright 2015 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * 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 org.hawkular.feedcomm.api;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * This is a "stream" that includes some in-memory data as well as another stream.  The in-memory data (or not null
+ * or empty) will be read first. Once the in-memory data has been exhausted, any additional reads will read data
+ * from the input stream.
+ *
+ * This is used in the use-case that an input stream contained a JSON command (e.g. BasicMessage) followed by
+ * additional binary data. The JSON parser may have read extra data over and beyond the actual JSON message data.
+ * In that case, the extra data the JSON parser read will be found in an in-memory byte array. That additional
+ * in-memory data byte array combined with the input stream that may contain even more data will both be stored
+ * in this object so both pieces of data can be accessed using the normal input stream API.
+ */
+public class BinaryData extends InputStream {
+
+    private byte[] inMemoryData;
+    private final InputStream streamData;
+    private int inMemoryDataPointer;
+
+    public BinaryData(byte[] inMemoryData, InputStream streamData) {
+        this.inMemoryData = (inMemoryData != null) ? inMemoryData : new byte[0];
+        this.streamData = streamData;
+        this.inMemoryDataPointer = 0;
+    }
+
+    public int read() throws IOException {
+        if (unreadInMemoryDataExists()) {
+            return (int) inMemoryData[inMemoryDataPointer++];
+        } else {
+            return streamData.read();
+        }
+    }
+
+    public int read(byte[] b) throws IOException {
+        if (unreadInMemoryDataExists()) {
+            return super.read(b); // the superclass implementation is all we need
+        } else {
+            return this.streamData.read(b); // delegate directly to the stream
+        }
+    }
+
+    public int read(byte[] b, int off, int len) throws IOException {
+        if (unreadInMemoryDataExists()) {
+            return super.read(b, off, len); // the superclass implementation is all we need
+        } else {
+            return this.streamData.read(b, off, len); // delegate directly to the stream
+        }
+    }
+
+    public long skip(long n) throws IOException {
+        if (unreadInMemoryDataExists()) {
+            return super.skip(n); // the superclass implementation is all we need
+        } else {
+            return this.streamData.skip(n); // delegate directly to the stream
+        }
+    }
+
+    public int available() throws IOException {
+        return (inMemoryData.length - inMemoryDataPointer) + streamData.available();
+    }
+
+    public void close() throws IOException {
+        // force nothing more to be read from the in-memory data, let the garbage collected free up its memory,
+        // but avoid NPEs by setting inMemoryData to a small zero-byte array.
+        inMemoryData = new byte[0];
+        inMemoryDataPointer = 0;
+        streamData.close();
+    }
+
+    public void mark(int readlimit) {
+        super.mark(readlimit); // superclass doesn't support this, and neither to we
+    }
+
+    public void reset() throws IOException {
+        super.reset(); // superclass doesn't support this, and neither do we
+    }
+
+    public boolean markSupported() {
+        return super.markSupported(); // superclass doesn't support this, and neither to we
+    }
+
+    private boolean unreadInMemoryDataExists() {
+        return (inMemoryDataPointer < inMemoryData.length);
+    }
+}

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

@@ -120,18 +120,18 @@ public void testReadFromInputStreamWithExtraData() throws IOException {
 
         ByteArrayInputStream in = new UncloseableByteArrayInputStream(nameAndJsonPlusExtra.getBytes());
 
-        Map<BasicMessage, byte[]> map = ad.deserialize(in);
+        Map<BasicMessage, BinaryData> 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);
+        BinaryData leftover = map.values().iterator().next();
+        byte[] leftoverByteArray = new byte[leftover.available()];
+        leftover.read(leftoverByteArray);
 
-        String totalRemaining = new String(leftoverFromJsonParser, "UTF-8") + new String(leftoverFromStream, "UTF-8");
+        String totalRemaining = new String(leftoverByteArray, "UTF-8");
         Assert.assertEquals(extra.length(), totalRemaining.length());
         Assert.assertEquals(extra, totalRemaining);
 
@@ -151,15 +151,15 @@ public void testReadFromInputStreamWithNoExtraData() throws IOException {
         String nameAndJson = EchoRequest.class.getName() + "={\"echoMessage\":\"msg\"}";
         ByteArrayInputStream in = new UncloseableByteArrayInputStream(nameAndJson.getBytes());
 
-        Map<BasicMessage, byte[]> map = ad.deserialize(in);
+        Map<BasicMessage, BinaryData> 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);
+        BinaryData leftover = map.values().iterator().next();
+        Assert.assertEquals(0, leftover.available());
         Assert.assertEquals(0, in.available());
     }
 

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

@@ -0,0 +1,173 @@
+/*
+ * Copyright 2015 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * 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 org.hawkular.feedcomm.api;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Arrays;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+@SuppressWarnings("resource")
+public class BinaryDataTest {
+
+    @Test
+    public void testEmptyBinaryData() throws Exception {
+        BinaryData binaryData;
+
+        binaryData = new BinaryData(null, buildInputStream(""));
+        Assert.assertEquals(0, binaryData.available());
+        Assert.assertEquals(-1, binaryData.read());
+        binaryData = new BinaryData(buildByteArray(""), buildInputStream(""));
+        Assert.assertEquals(0, binaryData.available());
+        Assert.assertEquals(-1, binaryData.read());
+    }
+
+    @Test
+    public void testClose() throws Exception {
+        // because our tests use ByteArrayInputStream, close has no effect on it.
+        // But close does have an effect on our in-memory data - closing the stream closes reads to the in-memory data
+        BinaryData binaryData = new BinaryData(buildByteArray("123"), buildInputStream(""));
+
+        // show we can start reading
+        Assert.assertEquals(3, binaryData.available());
+        Assert.assertEquals('1', binaryData.read());
+        Assert.assertEquals(2, binaryData.available());
+
+        // show that close worked
+        binaryData.close();
+        Assert.assertEquals(0, binaryData.available());
+        Assert.assertEquals(-1, binaryData.read());
+    }
+
+    @Test
+    public void testUnsupportedApis() {
+        BinaryData binaryData = new BinaryData(buildByteArray(""), buildInputStream("123"));
+        Assert.assertFalse(binaryData.markSupported());
+        try {
+            binaryData.reset();
+            Assert.fail("Mark is not supported - an exception should have been thrown");
+        } catch (IOException expected) {
+        }
+    }
+
+    @Test
+    public void testOnlyStreamData() throws Exception {
+        actualTests("", "1234567890");
+    }
+
+    @Test
+    public void testOnlyInMemoryData() throws Exception {
+        actualTests("1234567890", "");
+    }
+
+    @Test
+    public void testBothInMemoryDataAndStreamData() throws Exception {
+        actualTests("12345", "67890");
+    }
+
+    // callers MUST assure that the concatenation of byteArray and inputStream will always be "1234567890"
+    private void actualTests(String byteArrayString, String inputStreamString) throws Exception {
+        BinaryData binaryData;
+        byte[] bytes;
+
+        // available()
+        binaryData = new BinaryData(buildByteArray(byteArrayString), buildInputStream(inputStreamString));
+        Assert.assertEquals(10, binaryData.available());
+        Assert.assertEquals('1', binaryData.read());
+        Assert.assertEquals(9, binaryData.available());
+        Assert.assertEquals('2', binaryData.read());
+        Assert.assertEquals(8, binaryData.available());
+        Assert.assertEquals('3', binaryData.read());
+        Assert.assertEquals(7, binaryData.available());
+        Assert.assertEquals('4', binaryData.read());
+        Assert.assertEquals(6, binaryData.available());
+        Assert.assertEquals('5', binaryData.read());
+        Assert.assertEquals(5, binaryData.available());
+        Assert.assertEquals('6', binaryData.read());
+        Assert.assertEquals(4, binaryData.available());
+        Assert.assertEquals('7', binaryData.read());
+        Assert.assertEquals(3, binaryData.available());
+        Assert.assertEquals('8', binaryData.read());
+        Assert.assertEquals(2, binaryData.available());
+        Assert.assertEquals('9', binaryData.read());
+        Assert.assertEquals(1, binaryData.available());
+        Assert.assertEquals('0', binaryData.read());
+        Assert.assertEquals(0, binaryData.available());
+        Assert.assertEquals(-1, binaryData.read());
+
+        // read(byte[])
+        binaryData = new BinaryData(buildByteArray(byteArrayString), buildInputStream(inputStreamString));
+        bytes = new byte[0]; // according to javadocs, an array of length of zero always returns 0 from read
+        Assert.assertEquals(0, binaryData.read(bytes));
+        Assert.assertEquals("", new String(bytes, "UTF-8"));
+
+        binaryData = new BinaryData(buildByteArray(byteArrayString), buildInputStream(inputStreamString));
+        bytes = new byte[5];
+        Assert.assertEquals(5, binaryData.read(bytes));
+        Assert.assertEquals("12345", new String(bytes, "UTF-8"));
+        Assert.assertEquals(5, binaryData.read(bytes));
+        Assert.assertEquals("67890", new String(bytes, "UTF-8"));
+        Assert.assertEquals(-1, binaryData.read(bytes));
+
+        binaryData = new BinaryData(buildByteArray(byteArrayString), buildInputStream(inputStreamString));
+        bytes = new byte[10];
+        Assert.assertEquals(10, binaryData.read(bytes));
+        Assert.assertEquals("1234567890", new String(bytes, "UTF-8"));
+        Assert.assertEquals(-1, binaryData.read(bytes));
+
+        binaryData = new BinaryData(buildByteArray(byteArrayString), buildInputStream(inputStreamString));
+        bytes = new byte[15];
+        Assert.assertEquals(10, binaryData.read(bytes));
+        Assert.assertEquals("1234567890\0\0\0\0\0", new String(bytes, "UTF-8"));
+        Assert.assertEquals(-1, binaryData.read(bytes));
+
+        // read(byte[], int, int)
+        binaryData = new BinaryData(buildByteArray(byteArrayString), buildInputStream(inputStreamString));
+        bytes = new byte[0]; // according to javadocs, an array of length of zero always returns 0 from read
+        Assert.assertEquals(0, binaryData.read(bytes, 0, 0));
+        Assert.assertEquals("", new String(bytes, "UTF-8"));
+
+        binaryData = new BinaryData(buildByteArray(byteArrayString), buildInputStream(inputStreamString));
+        bytes = new byte[10];
+        Assert.assertEquals(3, binaryData.read(bytes, 0, 3));
+        Assert.assertEquals(4, binaryData.read(bytes, 3, 4));
+        Assert.assertEquals(3, binaryData.read(bytes, 7, 3));
+        Assert.assertEquals("1234567890", new String(bytes, "UTF-8"));
+        Assert.assertEquals(-1, binaryData.read(bytes));
+
+        // skip(int)
+        binaryData = new BinaryData(buildByteArray(byteArrayString), buildInputStream(inputStreamString));
+        Assert.assertEquals(3, binaryData.skip(3));
+        Assert.assertEquals('4', binaryData.read());
+        Assert.assertEquals(4, binaryData.skip(4));
+        Assert.assertEquals('9', binaryData.read());
+        Assert.assertEquals('0', binaryData.read());
+        Assert.assertEquals(0, binaryData.skip(111));
+        Assert.assertEquals(-1, binaryData.read());
+    }
+
+    private byte[] buildByteArray(String str) {
+        return Arrays.copyOf(str.getBytes(), str.length());
+    }
+
+    private InputStream buildInputStream(String str) {
+        return new ByteArrayInputStream(str.getBytes());
+    }
+}