#442 - Add KlvParser.parseStream(InputStream)

api/src/main/java/org/jmisb/api/common/KlvParseException.java

@@ -1,7 +1,11 @@
 package org.jmisb.api.common;
 
+import java.util.Arrays;
+
 /** Indicates an error occurred during metadata parsing. */
 public class KlvParseException extends Exception {
+    private final byte[] buffer;
+
     /**
      * Constructor.
      *
@@ -10,5 +14,22 @@ public class KlvParseException extends Exception {
      */
     public KlvParseException(String message) {
         super(message);
+        this.buffer = null;
+    }
+
+    /**
+     * Constructor.
+     *
+     * @param exception the inner exception object
+     * @param buffer the buffer corresponding to the inner exception
+     */
+    public KlvParseException(KlvParseException exception, byte[] buffer) {
+        super(exception);
+        this.buffer = Arrays.copyOf(buffer, buffer.length);
+    }
+
+    /** @return a copy of the {@code byte[]} corresponding to the wrapped exception. */
+    public byte[] getBuffer() {
+        return Arrays.copyOf(buffer, buffer.length);
     }
 }

api/src/main/java/org/jmisb/api/klv/BerDecoder.java

@@ -1,7 +1,11 @@
 package org.jmisb.api.klv;
 
+import java.io.IOException;
+import java.io.InputStream;
+
 /** Decode data using Basic Encoding Rules (BER). */
 public class BerDecoder {
+
     private BerDecoder() {}
 
     /**
@@ -10,7 +14,7 @@ private BerDecoder() {}
      * @param data Array holding the BER-encoded data
      * @param offset Index of the first byte of the array to decode
      * @param isOid true if the data is encoded using BER-OID
-     * @return decoded The decoded field
+     * @return the decoded field
      * @throws IllegalArgumentException if the encoded data is invalid
      */
     public static BerField decode(byte[] data, int offset, boolean isOid)
@@ -77,4 +81,62 @@ public static BerField decode(byte[] data, int offset, boolean isOid)
 
         return new BerField(length, value);
     }
+
+    /**
+     * Decode a field (length and value) from an InputStream.
+     *
+     * @param is InputStream with BER-encoded data at the tip
+     * @param isOid true if the data is encoded using BER-OID
+     * @return the decoded field
+     * @throws IllegalArgumentException if the encoded data is invalid
+     */
+    public static BerField decode(InputStream is, boolean isOid) throws IOException {
+        if (!isOid) {
+            return decodeBer(is);
+        }
+
+        return decodeBerOid(is);
+    }
+
+    private static BerField decodeBer(InputStream is) throws IOException {
+        int length = is.read();
+
+        if ((length & 0x80) == 0) {
+            // BER Short Form.  If the first bit of the BER is 0 then the BER is 1-byte.
+            return new BerField(1, length);
+        }
+
+        // BER Long Form (variable length)
+        int berLength = length & 0x7f;
+        if (berLength > 4) {
+            throw new IllegalArgumentException(
+                    "BER long form: BER length is >5 bytes; data is probably corrupt");
+        }
+        int fullBerSize = berLength + 1;
+        byte[] data = new byte[berLength];
+        int read = is.read(data, 0, data.length);
+        if (read != data.length) {
+            throw new IllegalArgumentException("BER parsing ran out of bytes");
+        }
+        int len = 0;
+        for (int i = 0; i < berLength; ++i) {
+            int b = 0x00FF & data[i];
+            len = (len << 8) | b;
+        }
+        return new BerField(fullBerSize, len);
+    }
+
+    private static BerField decodeBerOid(InputStream is) throws IOException {
+        int read;
+        int value = 0;
+        int length = 0;
+        do {
+            read = is.read();
+            int highbits = (value << 7);
+            int lowbits = (read & 0x7F);
+            value = highbits + lowbits;
+            length++;
+        } while ((read & 0x80) == 0x80);
+        return new BerField(length, value);
+    }
 }

api/src/main/java/org/jmisb/api/klv/KlvParser.java

@@ -1,8 +1,12 @@
 package org.jmisb.api.klv;
 
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.function.Consumer;
 import org.jmisb.api.common.KlvParseException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -13,6 +17,84 @@ public class KlvParser {
 
     private KlvParser() {}
 
+    /**
+     * Parse an InputStream containing one or more {@link IMisbMessage}s.
+     *
+     * <p>This differs from {@link #parseBytes(byte[]) parseBytes(byte[])} by parsing the
+     * UniversalLabel, BER-length, and value, and sending the {@link IMisbMessage} object in
+     * realtime to {@code handler}.
+     *
+     * <p>This is an additional interface for parsing KLV metadata. It assumes that {@code is}
+     * contains one or more top-level messages, i.e., byte sequences starting with a Universal Label
+     * (UL). If a particular UL is unsupported it will be send to {@code handler} as a {@link
+     * RawMisbMessage}.
+     *
+     * <p>If parsing errors occur with a valid length message, with an invalid value, the
+     * corresponding {@code byte[]} and {@link KlvParseException} will be sent to {@code
+     * exceptionHandler}.
+     *
+     * <p>The supported UL are determined by the {@link MisbMessageFactory} singleton.
+     *
+     * @param is The input stream
+     * @param handler The resultant {@link IMisbMessage} objects streamed
+     * @param exceptionHandler The {@link KlvParseException} errors detected in the stream.
+     * @throws KlvParseException if an unrecoverable parsing exception occurs when splitting
+     *     messages
+     */
+    public static void parseStream(
+            InputStream is,
+            Consumer<IMisbMessage> handler,
+            Consumer<KlvParseException> exceptionHandler)
+            throws KlvParseException {
+
+        // reusable key array to minimize garbage
+        byte[] key = new byte[UniversalLabel.LENGTH];
+
+        try {
+            while (true) {
+                ByteArrayOutputStream out = new ByteArrayOutputStream();
+
+                // Read the UniversalLabel
+                int read = is.read(key, 0, key.length);
+                if (read < 0) {
+                    break;
+                }
+                if (read != key.length) {
+                    throw new KlvParseException(
+                            "Read " + read + " bytes when expected " + key.length);
+                }
+                out.write(key);
+
+                // Read the payload length
+                BerField length = BerDecoder.decode(is, false);
+                out.write(BerEncoder.encode(length.getValue()));
+
+                // Read the payload
+                byte[] payload = new byte[length.getValue()];
+                read = is.read(payload, 0, payload.length);
+                if (read == 0) {
+                    break;
+                }
+                if (read != payload.length) {
+                    throw new KlvParseException(
+                            "Read " + read + " bytes when expected " + key.length);
+                }
+                out.write(payload);
+
+                // hand off the IMisbMessage
+                byte[] buf = out.toByteArray();
+                try {
+                    IMisbMessage msg = MisbMessageFactory.getInstance().handleMessage(buf);
+                    handler.accept(msg);
+                } catch (KlvParseException e) {
+                    exceptionHandler.accept(new KlvParseException(e, buf));
+                }
+            }
+        } catch (IOException e) {
+            throw new KlvParseException("IOException during stream parsing");
+        }
+    }
+
     /**
      * Parse a byte array containing one or more {@link IMisbMessage}s.
      *

api/src/test/java/org/jmisb/api/klv/BerDecoderTest.java

@@ -1,5 +1,7 @@
 package org.jmisb.api.klv;
 
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
@@ -21,6 +23,24 @@ public void testShortFormLengthField() {
         Assert.assertEquals(l3.getLength(), 1);
     }
 
+    @Test
+    public void testShortFormLengthFieldInputStream() throws IOException {
+        // BER Short Form is always encoded in a single byte, and has its high order bit set to 0
+        byte[] data = {0x00, 0x05, 0x7f}; // 1, 5, 127
+        ByteArrayInputStream bais = new ByteArrayInputStream(data);
+        BerField l1 = BerDecoder.decode(bais, false);
+        BerField l2 = BerDecoder.decode(bais, false);
+        BerField l3 = BerDecoder.decode(bais, false);
+
+        Assert.assertEquals(l1.getValue(), 0);
+        Assert.assertEquals(l2.getValue(), 5);
+        Assert.assertEquals(l3.getValue(), 127);
+
+        Assert.assertEquals(l1.getLength(), 1);
+        Assert.assertEquals(l2.getLength(), 1);
+        Assert.assertEquals(l3.getLength(), 1);
+    }
+
     @Test
     public void testLongFormLengthField() {
         byte[] data = {
@@ -40,6 +60,26 @@ public void testLongFormLengthField() {
         Assert.assertEquals(l3.getLength(), 5);
     }
 
+    @Test
+    public void testLongFormLengthFieldInputStream() throws IOException {
+        byte[] data = {
+            (byte) 0x81, 0x05, (byte) 0x82, 0x01, (byte) 0x80, (byte) 0x84, 0x01, 0x01, 0x01, 0x01
+        };
+
+        ByteArrayInputStream bais = new ByteArrayInputStream(data);
+        BerField l1 = BerDecoder.decode(bais, false);
+        BerField l2 = BerDecoder.decode(bais, false);
+        BerField l3 = BerDecoder.decode(bais, false);
+
+        Assert.assertEquals(l1.getValue(), 5);
+        Assert.assertEquals(l2.getValue(), 384);
+        Assert.assertEquals(l3.getValue(), 16_843_009);
+
+        Assert.assertEquals(l1.getLength(), 2);
+        Assert.assertEquals(l2.getLength(), 3);
+        Assert.assertEquals(l3.getLength(), 5);
+    }
+
     @Test(expectedExceptions = IllegalArgumentException.class)
     public void testParseBufferOverrun() {
         byte[] data = {0x00, 0x05, 0x7f};

api/src/test/java/org/jmisb/api/klv/KlvParserTest.java

@@ -4,6 +4,8 @@
 
 import com.github.valfirst.slf4jtest.TestLogger;
 import com.github.valfirst.slf4jtest.TestLoggerFactory;
+import java.io.ByteArrayInputStream;
+import java.util.ArrayList;
 import java.util.List;
 import org.jmisb.api.common.KlvParseException;
 import org.testng.annotations.AfterMethod;
@@ -62,6 +64,54 @@ private void doParse() throws KlvParseException {
         assertTrue(rawMessage.getIdentifiers().isEmpty());
     }
 
+    @Test
+    public void checkParseStream() throws KlvParseException {
+        doParseStream();
+    }
+
+    private void doParseStream() throws KlvParseException {
+        byte[] bytes =
+                new byte[] {
+                    0x06, 0x0E, 0x2B, 0x34, 0x02, 0x0B, 0x01, 0x01, 0x0E, 0x01, 0x03, 0x01, 0x02,
+                    0x00, 0x00, 0x00, 0x09, 0x04, 0x02, 0x00, 0x4f, 0x01, 0x03, 0x1e, 0x2f, 0x3a,
+                    0x06, 0x0E, 0x2B, 0x34, 0x02, 0x0B, 0x01, 0x01, 0x0E, 0x01, 0x03, 0x01, 0x02,
+                    0x00, 0x00, 0x00, 0x09, 0x04, 0x02, 0x00, 0x4f, 0x01, 0x03, 0x1e, 0x2f, 0x3b
+                };
+        ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
+        List<IMisbMessage> messages = new ArrayList<>();
+        List<KlvParseException> errors = new ArrayList<>();
+        KlvParser.parseStream(bais, messages::add, errors::add);
+        assertEquals(0, errors.size());
+        if (!errors.isEmpty()) {
+            throw errors.get(0);
+        }
+
+        assertNotNull(messages);
+        assertEquals(messages.size(), 2);
+        for (int i = 0; i < messages.size(); i++) {
+            IMisbMessage message = messages.get(i);
+            assertNotNull(message);
+            assertEquals(
+                    message.getUniversalLabel(),
+                    new UniversalLabel(
+                            new byte[] {
+                                0x06, 0x0E, 0x2B, 0x34, 0x02, 0x0B, 0x01, 0x01, 0x0E, 0x01, 0x03,
+                                0x01, 0x02, 0x00, 0x00, 0x00
+                            }));
+            assertEquals(message.displayHeader(), "Unknown");
+            assertTrue(message instanceof RawMisbMessage);
+            RawMisbMessage rawMessage = (RawMisbMessage) message;
+            assertEquals(
+                    rawMessage.getBytes(),
+                    new byte[] {
+                        0x06, 0x0E, 0x2B, 0x34, 0x02, 0x0B, 0x01, 0x01, 0x0E, 0x01, 0x03, 0x01,
+                        0x02, 0x00, 0x00, 0x00, 0x09, 0x04, 0x02, 0x00, 0x4f, 0x01, 0x03, 0x1e,
+                        0x2f, (byte) (0x3a + i)
+                    });
+            assertTrue(rawMessage.getIdentifiers().isEmpty());
+        }
+    }
+
     @Test(expectedExceptions = KlvParseException.class)
     public void checkBadLength() throws KlvParseException {
         KlvParser.parseBytes(