442 - WIP attempt at processing InputStream

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() {}
 
     /**
@@ -77,4 +81,60 @@ 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 decoded 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;
+        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;
+        }
+        length = len;
+
+        return new BerField(fullBerSize, length);
+    }
+
+    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,73 @@ public class KlvParser {
 
     private KlvParser() {}
 
+    /**
+     * Parse an InputStream containing one or more {@link IMisbMessage}s.
+     *
+     * <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 returned as a {@link RawMisbMessage}.
+     *
+     * <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.
+     */
+    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(e);
+                }
+            }
+        } catch (IOException e) {
+            throw new KlvParseException("IOException during stream parsing");
+        }
+    }
+
     /**
      * Parse a byte array containing one or more {@link IMisbMessage}s.
      *