¿Cómo convertir OutputStream a InputStream?

Estoy en la etapa de desarrollo, donde tengo dos módulos y de uno obtuve salida como uno OutputStreamy segundo, que solo acepta InputStream. ¿Sabe usted cómo convertir OutputStreama InputStream(y no al revés, me refiero a realmente de esta manera) que voy a ser capaz de conectar estas dos partes?

Gracias

An OutputStreames aquel en el que escribe datos. Si algún módulo expone un OutputStream, la expectativa es que hay algo leyendo en el otro extremo.

Algo que expone y InputStream, por otro lado, indica que necesitará escuchar esta transmisión, y habrá datos que podrá leer.

Por lo tanto, es posible conectar un InputStreama unOutputStream

InputStream----read---> intermediateBytes[n] ----write----> OutputStream

Como alguien mencionó, esto es lo que el copy()método de IOUtils le permite hacer. No tiene sentido ir hacia otro lado ... espero que esto tenga algún sentido

ACTUALIZAR:

Por supuesto, cuanto más pienso en esto, más puedo ver cómo esto sería realmente un requisito. Sé que algunos de los comentarios mencionaron Pipedflujos de entrada / salida, pero hay otra posibilidad.

Si la secuencia de salida que está expuesta es a ByteArrayOutputStream, entonces siempre puede obtener el contenido completo llamando al toByteArray()método. Luego puede crear un contenedor de flujo de entrada utilizando la ByteArrayInputStreamsubclase. Estos dos son pseudo-flujos, ambos básicamente solo envuelven una matriz de bytes. Usar las transmisiones de esta manera, por lo tanto, es técnicamente posible, pero para mí sigue siendo muy extraño ...

Parece que hay muchos enlaces y otras cosas similares, pero no hay código real que use tuberías. La ventaja de usar java.io.PipedInputStreamy java.io.PipedOutputStreames que no hay consumo adicional de memoria. ByteArrayOutputStream.toByteArray()devuelve una copia del búfer original, lo que significa que, sea lo que sea que tenga en la memoria, ahora tiene dos copias. Luego, escribir en un InputStreammedio significa que ahora tiene tres copias de los datos.

El código:

// take the copy of the stream and re-write it to an InputStream
PipedInputStream in = new PipedInputStream();
final PipedOutputStream out = new PipedOutputStream(in);
new Thread(new Runnable() {
    public void run () {
        try {
            // write the original OutputStream to the PipedOutputStream
            // note that in order for the below method to work, you need
            // to ensure that the data has finished writing to the
            // ByteArrayOutputStream
            originalByteArrayOutputStream.writeTo(out);
        }
        catch (IOException e) {
            // logging and exception handling should go here
        }
        finally {
            // close the PipedOutputStream here because we're done writing data
            // once this thread has completed its run
            if (out != null) {
                // close the PipedOutputStream cleanly
                out.close();
            }
        }   
    }
}).start();

Este código supone que originalByteArrayOutputStreames un, ByteArrayOutputStreamya que generalmente es la única secuencia de salida utilizable, a menos que esté escribiendo en un archivo. ¡Espero que esto ayude! Lo bueno de esto es que, dado que está en un hilo separado, también funciona en paralelo, por lo que lo que sea que esté consumiendo su flujo de entrada también se transmitirá desde su flujo de salida anterior. Eso es beneficioso porque el búfer puede permanecer más pequeño y tendrá menos latencia y menos uso de memoria.

Como los flujos de entrada y salida son solo el punto inicial y final, la solución es almacenar temporalmente los datos en una matriz de bytes. Por lo tanto, debe crear intermedios ByteArrayOutputStream, a partir de los cuales crea byte[]que se utilizan como entrada para nuevos ByteArrayInputStream.

public void doTwoThingsWithStream(InputStream inStream, OutputStream outStream){ 
  //create temporary bayte array output stream
  ByteArrayOutputStream baos = new ByteArrayOutputStream();
  doFirstThing(inStream, baos);
  //create input stream from baos
  InputStream isFromFirstData = new ByteArrayInputStream(baos.toByteArray()); 
  doSecondThing(isFromFirstData, outStream);
}

Espero eso ayude.

Necesitará una clase intermedia que amortiguará entre. Cada vez que InputStream.read(byte[]...)se llama, la clase de almacenamiento en búfer llenará la matriz de bytes pasada con el siguiente fragmento pasado OutputStream.write(byte[]...). Dado que los tamaños de los fragmentos pueden no ser los mismos, la clase de adaptador deberá almacenar una cierta cantidad hasta que tenga suficiente para llenar el búfer de lectura y / o poder almacenar cualquier desbordamiento del búfer.

Este artículo tiene un buen desglose de algunos enfoques diferentes para este problema:

http://blog.ostermiller.org/convert-java-outputstream-inputstream

ByteArrayOutputStream buffer = (ByteArrayOutputStream) aOutputStream;
byte[] bytes = buffer.toByteArray();
InputStream inputStream = new ByteArrayInputStream(bytes);

La biblioteca de código abierto easystream tiene soporte directo para convertir un OutputStream en un InputStream: http://io-tools.sourceforge.net/easystream/tutorial/tutorial.html

También enumeran otras opciones: http://io-tools.sourceforge.net/easystream/OutputStream_to_InputStream.html

Encontré el mismo problema con la conversión de a ByteArrayOutputStreama ByteArrayInputStreamy lo resolví usando una clase derivada de la ByteArrayOutputStreamcual es capaz de devolver un ByteArrayInputStreamque se inicializa con el búfer interno del ByteArrayOutputStream. De esta forma no se usa memoria adicional y la 'conversión' es muy rápida:

package info.whitebyte.utils;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;

/**
 * This class extends the ByteArrayOutputStream by 
 * providing a method that returns a new ByteArrayInputStream
 * which uses the internal byte array buffer. This buffer
 * is not copied, so no additional memory is used. After
 * creating the ByteArrayInputStream the instance of the
 * ByteArrayInOutStream can not be used anymore.
 * <p>
 * The ByteArrayInputStream can be retrieved using <code>getInputStream()</code>.
 * @author Nick Russler
 */
public class ByteArrayInOutStream extends ByteArrayOutputStream {
    /**
     * Creates a new ByteArrayInOutStream. The buffer capacity is
     * initially 32 bytes, though its size increases if necessary.
     */
    public ByteArrayInOutStream() {
        super();
    }

    /**
     * Creates a new ByteArrayInOutStream, with a buffer capacity of
     * the specified size, in bytes.
     *
     * @param   size   the initial size.
     * @exception  IllegalArgumentException if size is negative.
     */
    public ByteArrayInOutStream(int size) {
        super(size);
    }

    /**
     * Creates a new ByteArrayInputStream that uses the internal byte array buffer 
     * of this ByteArrayInOutStream instance as its buffer array. The initial value 
     * of pos is set to zero and the initial value of count is the number of bytes 
     * that can be read from the byte array. The buffer array is not copied. This 
     * instance of ByteArrayInOutStream can not be used anymore after calling this
     * method.
     * @return the ByteArrayInputStream instance
     */
    public ByteArrayInputStream getInputStream() {
        // create new ByteArrayInputStream that respects the current count
        ByteArrayInputStream in = new ByteArrayInputStream(this.buf, 0, this.count);

        // set the buffer of the ByteArrayOutputStream 
        // to null so it can't be altered anymore
        this.buf = null;

        return in;
    }
}

Puse las cosas en github: https://github.com/nickrussler/ByteArrayInOutStream

La biblioteca io-extras puede ser útil. Por ejemplo, si desea comprimir un InputStreamuso GZIPOutputStreamy quiere que suceda sincrónicamente (usando el tamaño de búfer predeterminado de 8192):

InputStream is = ...
InputStream gz = IOUtil.pipe(is, o -> new GZIPOutputStream(o));

Tenga en cuenta que la biblioteca tiene una cobertura de prueba de unidad del 100% (¡por lo que vale, por supuesto!) Y está en Maven Central. La dependencia de Maven es:

<dependency>
  <groupId>com.github.davidmoten</groupId>
  <artifactId>io-extras</artifactId>
  <version>0.1</version>
</dependency>

Asegúrese de buscar una versión posterior.

Desde mi punto de vista, java.io.PipedInputStream / java.io.PipedOutputStream es la mejor opción para considerar. En algunas situaciones, es posible que desee utilizar ByteArrayInputStream / ByteArrayOutputStream. El problema es que necesita duplicar el búfer para convertir un ByteArrayOutputStream en un ByteArrayInputStream. También ByteArrayOutpuStream / ByteArrayInputStream están limitados a 2GB. Aquí hay una implementación OutpuStream / InputStream que escribí para evitar las limitaciones de ByteArrayOutputStream / ByteArrayInputStream (código Scala, pero fácilmente comprensible para los desarrolladores de Java):

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
      }
    }
  }
}

Fácil de usar, sin duplicación de búfer, sin límite de memoria de 2 GB

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(...)
...

Si desea hacer un OutputStream desde un InputStream, hay un problema básico. Un método que escribe en un bloque OutputStream hasta que se hace. Por lo tanto, el resultado está disponible cuando finaliza el método de escritura. Esto tiene 2 consecuencias:

1. Si usa solo un subproceso, debe esperar hasta que todo esté escrito (por lo que debe almacenar los datos de la secuencia en la memoria o el disco).
2. Si desea acceder a los datos antes de que finalice, necesita un segundo hilo.

La variante 1 se puede implementar utilizando matrices de bytes o archivada. La variante 1 se puede implementar usando pipies (ya sea directamente o con abstracción adicional, por ejemplo, RingBuffer o google lib del otro comentario).

De hecho, con Java estándar no hay otra forma de resolver el problema. Cada solución es una implementación de uno de estos.

Hay un concepto llamado "continuación" (ver wikipedia para más detalles). En este caso, básicamente esto significa:

• hay una secuencia de salida especial que espera una cierta cantidad de datos
• si se alcanza la cantidad, la secuencia le da el control a su contraparte, que es una secuencia de entrada especial
• la secuencia de entrada hace que la cantidad de datos esté disponible hasta que se lee, después de eso, devuelve el control a la secuencia de salida

Si bien algunos idiomas tienen este concepto incorporado, para Java necesitas algo de "magia". Por ejemplo, "commons-javaflow" de apache implementa tales como java. La desventaja es que esto requiere algunas modificaciones especiales de código de bytes en el momento de la compilación. Por lo tanto, tendría sentido poner todo el material en una biblioteca adicional con scripts de compilación personalizados.

Publicación anterior, pero podría ayudar a otros. Use esta forma:

OutputStream out = new ByteArrayOutputStream();
...
out.write();
...
ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(out.toString().getBytes()));

Aunque no puede convertir un OutputStream en un InputStream, java proporciona una forma de usar PipedOutputStream y PipedInputStream para que pueda tener datos escritos en un PipedOutputStream para que estén disponibles a través de un PipedInputStream asociado.
En algún momento, enfrenté una situación similar al tratar con bibliotecas de terceros que requerían que se les pasara una instancia de InputStream en lugar de una instancia de OutputStream.
La forma en que solucioné este problema es usar PipedInputStream y PipedOutputStream.
Por cierto, son difíciles de usar y debes usar subprocesos múltiples para lograr lo que deseas. Recientemente publiqué una implementación en github que puedes usar.
Aquí está el enlace . Puedes revisar la wiki para entender cómo usarla.