Add a newInputStream() to the IBaseDataObject. (#617)

src/main/java/emissary/core/BaseDataObject.java

@@ -15,10 +15,12 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.Serializable;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
 import java.nio.channels.SeekableByteChannel;
 import java.rmi.Remote;
 import java.util.ArrayList;
@@ -369,6 +371,16 @@ public SeekableByteChannelFactory getChannelFactory() {
         }
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public InputStream newInputStream() {
+        final SeekableByteChannelFactory sbcf = getChannelFactory();
+
+        return sbcf == null ? null : Channels.newInputStream(sbcf.create());
+    }
+
     /**
      * <p>
      * Return BaseDataObjects byte array OR as much as we can from the reference to the data up to MAX_BYTE_ARRAY_SIZE.

src/main/java/emissary/core/IBaseDataObject.java

@@ -4,6 +4,7 @@
 import emissary.directory.DirectoryEntry;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.util.Collection;
 import java.util.Date;
@@ -66,6 +67,18 @@ enum MergePolicy {
      */
     void setChannelFactory(final SeekableByteChannelFactory sbcf);
 
+    /**
+     * Returns a new InputStream to the data that this BaseDataObject contains.
+     * <p>
+     * NOTE 1: Mutating the data elements of this IBaseDataObject while reading from the InputStream will have indeterminate
+     * results.
+     * <p>
+     * NOTE 2: The calling code is responsible for closing the returned InputStream.
+     * 
+     * @return a new stream that reads the data that this object contains, or null if this object has no data.
+     */
+    InputStream newInputStream();
+
     /**
      * Returns the seekable byte channel factory containing a reference to the data
      * 

src/test/java/emissary/core/BaseDataObjectTest.java

@@ -12,6 +12,7 @@
 import ch.qos.logback.classic.Level;
 import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.Multimap;
+import org.apache.commons.io.IOUtils;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -20,7 +21,9 @@
 import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.Mockito;
 
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.lang.reflect.Field;
 import java.nio.ByteBuffer;
 import java.nio.channels.SeekableByteChannel;
@@ -1406,4 +1409,34 @@ void testArrayInArrayOutNoSet() throws IOException {
             logbackTester.checkLogList(LEVELS_ONE_WARN, ONE_UNSAFE_MODIFICATION_DETECTED, NO_THROWABLES);
         }
     }
+
+    @Test
+    void testNewInputStream() throws IOException {
+        final IBaseDataObject ibdo = new BaseDataObject();
+
+        assertNull(ibdo.newInputStream());
+
+        final byte[] bytes1 = new byte[] {0, 1, 2, 3};
+
+        ibdo.setData(bytes1);
+
+        try (final InputStream bytesInputStream = ibdo.newInputStream();
+                final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();) {
+            IOUtils.copy(bytesInputStream, byteArrayOutputStream);
+
+            assertArrayEquals(bytes1, byteArrayOutputStream.toByteArray());
+        }
+
+        final byte[] bytes2 = new byte[] {4, 5, 6, 7};
+        final SeekableByteChannelFactory sbcf = SeekableByteChannelHelper.memory(bytes2);
+
+        ibdo.setChannelFactory(sbcf);
+
+        try (final InputStream sbcfInputStream = ibdo.newInputStream();
+                final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) {
+            IOUtils.copy(sbcfInputStream, byteArrayOutputStream);
+
+            assertArrayEquals(bytes2, byteArrayOutputStream.toByteArray());
+        }
+    }
 }