From my point of view, java.io.PipedInputStream/java.io.PipedOutputStream is the best option to considere. In some situations you may want to use ByteArrayInputStream/ByteArrayOutputStream. The problem is that you need to duplicate the buffer to convert a ByteArrayOutputStream to a ByteArrayInputStream. Also ByteArrayOutpuStream/ByteArrayInputStream are limited to 2GB. Here is an OutpuStream/InputStream implementation I wrote to bypass ByteArrayOutputStream/ByteArrayInputStream limitations (Scala code, but easily understandable for java developpers):
import java.io.{IOException, InputStream, OutputStream} import scala.annotation.tailrec /** Acts as a replacement for ByteArrayOutputStream * */ class HugeMemoryOutputStream(capacity: Long) extends OutputStream { private val PAGE_SIZE: Int = 1024000 private val ALLOC_STEP: Int = 1024 /** Pages array * */ private var streamBuffers: Array[Array[Byte]] = Array.empty[Array[Byte]] /** Allocated pages count * */ private var pageCount: Int = 0 /** Allocated bytes count * */ private var allocatedBytes: Long = 0 /** Current position in stream * */ private var position: Long = 0 /** Stream length * */ private var length: Long = 0 allocSpaceIfNeeded(capacity) /** Gets page count based on given length * * @param length Buffer length * @return Page count to hold the specified amount of data */ private def getPageCount(length: Long) = { var pageCount = (length / PAGE_SIZE).toInt + 1 if ((length % PAGE_SIZE) == 0) { pageCount -= 1 } pageCount } /** Extends pages array * */ private def extendPages(): Unit = { if (streamBuffers.isEmpty) { streamBuffers = new Array[Array[Byte]](ALLOC_STEP) } else { val newStreamBuffers = new Array[Array[Byte]](streamBuffers.length + ALLOC_STEP) Array.copy(streamBuffers, 0, newStreamBuffers, 0, streamBuffers.length) streamBuffers = newStreamBuffers } pageCount = streamBuffers.length } /** Ensures buffers are bug enough to hold specified amount of data * * @param value Amount of data */ private def allocSpaceIfNeeded(value: Long): Unit = { @tailrec def allocSpaceIfNeededIter(value: Long): Unit = { val currentPageCount = getPageCount(allocatedBytes) val neededPageCount = getPageCount(value) if (currentPageCount < neededPageCount) { if (currentPageCount == pageCount) extendPages() streamBuffers(currentPageCount) = new Array[Byte](PAGE_SIZE) allocatedBytes = (currentPageCount + 1).toLong * PAGE_SIZE allocSpaceIfNeededIter(value) } } if (value < 0) throw new Error("AllocSpaceIfNeeded < 0") if (value > 0) { allocSpaceIfNeededIter(value) length = Math.max(value, length) if (position > length) position = length } } /** * Writes the specified byte to this output stream. The general * contract for <code>write</code> is that one byte is written * to the output stream. The byte to be written is the eight * low-order bits of the argument <code>b</code>. The 24 * high-order bits of <code>b</code> are ignored. * <p> * Subclasses of <code>OutputStream</code> must provide an * implementation for this method. * * @param b the <code>byte</code>. */ @throws[IOException] override def write(b: Int): Unit = { val buffer: Array[Byte] = new Array[Byte](1) buffer(0) = b.toByte write(buffer) } /** * Writes <code>len</code> bytes from the specified byte array * starting at offset <code>off</code> to this output stream. * The general contract for <code>write(b, off, len)</code> is that * some of the bytes in the array <code>b</code> are written to the * output stream in order; element <code>b[off]</code> is the first * byte written and <code>b[off+len-1]</code> is the last byte written * by this operation. * <p> * The <code>write</code> method of <code>OutputStream</code> calls * the write method of one argument on each of the bytes to be * written out. Subclasses are encouraged to override this method and * provide a more efficient implementation. * <p> * If <code>b</code> is <code>null</code>, a * <code>NullPointerException</code> is thrown. * <p> * If <code>off</code> is negative, or <code>len</code> is negative, or * <code>off+len</code> is greater than the length of the array * <code>b</code>, then an <tt>IndexOutOfBoundsException</tt> is thrown. * * @param b the data. * @param off the start offset in the data. * @param len the number of bytes to write. */ @throws[IOException] override def write(b: Array[Byte], off: Int, len: Int): Unit = { @tailrec def writeIter(b: Array[Byte], off: Int, len: Int): Unit = { val currentPage: Int = (position / PAGE_SIZE).toInt val currentOffset: Int = (position % PAGE_SIZE).toInt if (len != 0) { val currentLength: Int = Math.min(PAGE_SIZE - currentOffset, len) Array.copy(b, off, streamBuffers(currentPage), currentOffset, currentLength) position += currentLength writeIter(b, off + currentLength, len - currentLength) } } allocSpaceIfNeeded(position + len) writeIter(b, off, len) } /** Gets an InputStream that points to HugeMemoryOutputStream buffer * * @return InputStream */ def asInputStream(): InputStream = { new HugeMemoryInputStream(streamBuffers, length) } private class HugeMemoryInputStream(streamBuffers: Array[Array[Byte]], val length: Long) extends InputStream { /** Current position in stream * */ private var position: Long = 0 /** * Reads the next byte of data from the input stream. The value byte is * returned as an <code>int</code> in the range <code>0</code> to * <code>255</code>. If no byte is available because the end of the stream * has been reached, the value <code>-1</code> is returned. This method * blocks until input data is available, the end of the stream is detected, * or an exception is thrown. * * <p> A subclass must provide an implementation of this method. * * @return the next byte of data, or <code>-1</code> if the end of the * stream is reached. */ @throws[IOException] def read: Int = { val buffer: Array[Byte] = new Array[Byte](1) if (read(buffer) == 0) throw new Error("End of stream") else buffer(0) } /** * Reads up to <code>len</code> bytes of data from the input stream into * an array of bytes. An attempt is made to read as many as * <code>len</code> bytes, but a smaller number may be read. * The number of bytes actually read is returned as an integer. * * <p> This method blocks until input data is available, end of file is * detected, or an exception is thrown. * * <p> If <code>len</code> is zero, then no bytes are read and * <code>0</code> is returned; otherwise, there is an attempt to read at * least one byte. If no byte is available because the stream is at end of * file, the value <code>-1</code> is returned; otherwise, at least one * byte is read and stored into <code>b</code>. * * <p> The first byte read is stored into element <code>b[off]</code>, the * next one into <code>b[off+1]</code>, and so on. The number of bytes read * is, at most, equal to <code>len</code>. Let <i>k</i> be the number of * bytes actually read; these bytes will be stored in elements * <code>b[off]</code> through <code>b[off+</code><i>k</i><code>-1]</code>, * leaving elements <code>b[off+</code><i>k</i><code>]</code> through * <code>b[off+len-1]</code> unaffected. * * <p> In every case, elements <code>b[0]</code> through * <code>b[off]</code> and elements <code>b[off+len]</code> through * <code>b[b.length-1]</code> are unaffected. * * <p> The <code>read(b,</code> <code>off,</code> <code>len)</code> method * for class <code>InputStream</code> simply calls the method * <code>read()</code> repeatedly. If the first such call results in an * <code>IOException</code>, that exception is returned from the call to * the <code>read(b,</code> <code>off,</code> <code>len)</code> method. If * any subsequent call to <code>read()</code> results in a * <code>IOException</code>, the exception is caught and treated as if it * were end of file; the bytes read up to that point are stored into * <code>b</code> and the number of bytes read before the exception * occurred is returned. The default implementation of this method blocks * until the requested amount of input data <code>len</code> has been read, * end of file is detected, or an exception is thrown. Subclasses are encouraged * to provide a more efficient implementation of this method. * * @param b the buffer into which the data is read. * @param off the start offset in array <code>b</code> * at which the data is written. * @param len the maximum number of bytes to read. * @return the total number of bytes read into the buffer, or * <code>-1</code> if there is no more data because the end of * the stream has been reached. * @see java.io.InputStream#read() */ @throws[IOException] override def read(b: Array[Byte], off: Int, len: Int): Int = { @tailrec def readIter(acc: Int, b: Array[Byte], off: Int, len: Int): Int = { val currentPage: Int = (position / PAGE_SIZE).toInt val currentOffset: Int = (position % PAGE_SIZE).toInt val count: Int = Math.min(len, length - position).toInt if (count == 0 || position >= length) acc else { val currentLength = Math.min(PAGE_SIZE - currentOffset, count) Array.copy(streamBuffers(currentPage), currentOffset, b, off, currentLength) position += currentLength readIter(acc + currentLength, b, off + currentLength, len - currentLength) } } readIter(0, b, off, len) } /** * Skips over and discards <code>n</code> bytes of data from this input * stream. The <code>skip</code> method may, for a variety of reasons, end * up skipping over some smaller number of bytes, possibly <code>0</code>. * This may result from any of a number of conditions; reaching end of file * before <code>n</code> bytes have been skipped is only one possibility. * The actual number of bytes skipped is returned. If <code>n</code> is * negative, the <code>skip</code> method for class <code>InputStream</code> always * returns 0, and no bytes are skipped. Subclasses may handle the negative * value differently. * * The <code>skip</code> method of this class creates a * byte array and then repeatedly reads into it until <code>n</code> bytes * have been read or the end of the stream has been reached. Subclasses are * encouraged to provide a more efficient implementation of this method. * For instance, the implementation may depend on the ability to seek. * * @param n the number of bytes to be skipped. * @return the actual number of bytes skipped. */ @throws[IOException] override def skip(n: Long): Long = { if (n < 0) 0 else { position = Math.min(position + n, length) length - position } } } }
Easy to use, no buffer duplication, no 2GB memory limit
val out: HugeMemoryOutputStream = new HugeMemoryOutputStream(initialCapacity /*may be 0*/) out.write(...) ... val in1: InputStream = out.asInputStream() in1.read(...) ... val in2: InputStream = out.asInputStream() in2.read(...) ...
new YourOutputStream(thePipedOutputStream)andnew YourInputStream(thePipedInputStream)which probably is not the way your stream works. So I don't think this is the solution.