Comment convertir OutputStream en InputStream?

Je suis au stade de développement, où j'ai deux modules et à partir de l'un, j'ai obtenu une sortie en tant que OutputStreamdeuxième et un seul, qui accepte uniquement InputStream. Savez - vous comment convertir OutputStreamà InputStream( et non vice versa, je veux dire vraiment de cette façon) que je serai en mesure de relier ces deux parties?

Merci

Un OutputStreamest celui où vous écrivez des données. Si un module expose un OutputStream, on s'attend à ce qu'il y ait quelque chose à lire à l'autre bout.

Quelque chose qui expose un InputStream, d'autre part, indique que vous devrez écouter ce flux et qu'il y aura des données que vous pourrez lire.

Il est donc possible de connecter un InputStreamà unOutputStream

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

Comme quelqu'un l'a mentionné, c'est ce que la copy()méthode de IOUtils vous permet de faire. Cela n'a pas de sens d'aller dans l'autre sens ... j'espère que cela a du sens

METTRE À JOUR:

Bien sûr, plus j'y pense, plus je peux voir comment ce serait une exigence. Je connais certains des commentaires mentionnés Pipedflux d'entrée / sortie, mais il existe une autre possibilité.

Si le flux de sortie exposé est un ByteArrayOutputStream, vous pouvez toujours obtenir le contenu complet en appelant la toByteArray()méthode. Ensuite, vous pouvez créer un wrapper de flux d'entrée à l'aide de la ByteArrayInputStreamsous-classe. Ces deux sont des pseudo-flux, ils enveloppent simplement un tableau d'octets. L'utilisation des flux de cette façon est donc techniquement possible, mais pour moi, c'est encore très étrange ...

Il semble y avoir de nombreux liens et d'autres choses de ce genre, mais aucun code réel utilisant des canaux. L'avantage d'utiliser java.io.PipedInputStreamet java.io.PipedOutputStreamest qu'il n'y a pas de consommation supplémentaire de mémoire. ByteArrayOutputStream.toByteArray()renvoie une copie du tampon d'origine, ce qui signifie que tout ce que vous avez en mémoire, vous en avez maintenant deux copies. Ensuite, en écrivant à un InputStreammoyen, vous disposez maintenant de trois copies des données.

Le code:

// 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();

Ce code suppose que le originalByteArrayOutputStreamest un ByteArrayOutputStreamcar il est généralement le seul flux de sortie utilisable, sauf si vous écrivez dans un fichier. J'espère que ça aide! La grande chose à ce sujet est que, puisqu'il est dans un thread séparé, il fonctionne également en parallèle, donc tout ce qui consomme votre flux d'entrée sera également diffusé à partir de votre ancien flux de sortie. C'est avantageux car le tampon peut rester plus petit et vous aurez moins de latence et moins d'utilisation de mémoire.

Comme les flux d'entrée et de sortie ne sont que des points de début et de fin, la solution consiste à stocker temporairement les données dans un tableau d'octets. Vous devez donc créer un intermédiaire ByteArrayOutputStream, à partir duquel vous créez byte[]qui est utilisé comme entrée pour new 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);
}

J'espère que ça aide.

Vous aurez besoin d'une classe intermédiaire qui tamponnera entre. À chaque InputStream.read(byte[]...)appel, la classe de mise en mémoire tampon remplira le tableau d'octets transmis avec le bloc suivant transmis depuis OutputStream.write(byte[]...). Étant donné que les tailles des morceaux peuvent ne pas être les mêmes, la classe d'adaptateur devra stocker une certaine quantité jusqu'à ce qu'elle ait suffisamment pour remplir le tampon de lecture et / ou être capable de stocker tout débordement de tampon.

Cet article présente une belle ventilation de quelques approches différentes de ce problème:

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

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

La bibliothèque open source easystream prend directement en charge la conversion d'un OutputStream en un InputStream: http://io-tools.sourceforge.net/easystream/tutorial/tutorial.html

Ils répertorient également d'autres options: http://io-tools.sourceforge.net/easystream/OutputStream_to_InputStream.html

J'ai rencontré le même problème avec la conversion de a ByteArrayOutputStreamen a ByteArrayInputStreamet je l'ai résolu en utilisant une classe dérivée à partir de ByteArrayOutputStreamlaquelle est capable de retourner un ByteArrayInputStreamqui est initialisé avec le tampon interne de ByteArrayOutputStream. De cette façon, aucune mémoire supplémentaire n'est utilisée et la «conversion» est très rapide:

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

J'ai mis les trucs sur github: https://github.com/nickrussler/ByteArrayInOutStream

La bibliothèque io-extras peut être utile. Par exemple, si vous voulez gziper un InputStreamusing GZIPOutputStreamet que cela se produise de manière synchrone (en utilisant la taille de tampon par défaut de 8192):

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

Notez que la bibliothèque a une couverture de test unitaire à 100% (pour ce que ça vaut bien sûr!) Et est sur Maven Central. La dépendance Maven est:

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

Assurez-vous de vérifier une version ultérieure.

De mon point de vue, java.io.PipedInputStream / java.io.PipedOutputStream est la meilleure option à considérer. Dans certaines situations, vous souhaiterez peut-être utiliser ByteArrayInputStream / ByteArrayOutputStream. Le problème est que vous devez dupliquer le tampon pour convertir un ByteArrayOutputStream en ByteArrayInputStream. De même, ByteArrayOutpuStream / ByteArrayInputStream sont limités à 2 Go. Voici une implémentation OutpuStream / InputStream que j'ai écrite pour contourner les limitations de ByteArrayOutputStream / ByteArrayInputStream (code Scala, mais facilement compréhensible pour les développeurs 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
      }
    }
  }
}

Facile à utiliser, pas de duplication de tampon, pas de limite de mémoire de 2 Go

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 vous voulez créer un OutputStream à partir d'un InputStream, il y a un problème de base. Une méthode qui écrit dans un bloc OutputStream jusqu'à ce qu'elle soit terminée. Le résultat est donc disponible lorsque la méthode d'écriture est terminée. Cela a 2 conséquences:

1. Si vous n'utilisez qu'un seul thread, vous devez attendre que tout soit écrit (vous devez donc stocker les données du flux dans la mémoire ou sur le disque).
2. Si vous souhaitez accéder aux données avant qu'elles ne soient terminées, vous avez besoin d'un deuxième thread.

La variante 1 peut être implémentée à l'aide de tableaux d'octets ou déposée. La variante 1 peut être implémentée à l'aide de pipies (directement ou avec une abstraction supplémentaire - par exemple RingBuffer ou la bibliothèque google de l'autre commentaire).

En effet, avec java standard, il n'y a pas d'autre moyen de résoudre le problème. Chaque solution est une implémentation de l'une d'entre elles.

Il existe un concept appelé "continuation" (voir wikipedia pour plus de détails). Dans ce cas, cela signifie essentiellement:

• il existe un flux de sortie spécial qui attend une certaine quantité de données
• si le montant est atteint, le flux donne le contrôle à son homologue qui est un flux d'entrée spécial
• le flux d'entrée rend la quantité de données disponibles jusqu'à ce qu'il soit lu, après quoi, il renvoie le contrôle au flux de sortie

Alors que certaines langues ont ce concept intégré, pour java, vous avez besoin de "magie". Par exemple "commons-javaflow" d'apache implémente un tel pour java. L'inconvénient est que cela nécessite des modifications de bytecode spéciales au moment de la construction. Il serait donc logique de mettre toutes les choses dans une bibliothèque supplémentaire avec des scripts de construction personnalisés.

Ancien message mais pourrait aider les autres, utilisez cette méthode:

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

Bien que vous ne puissiez pas convertir un OutputStream en un InputStream, java fournit un moyen d'utiliser PipedOutputStream et PipedInputStream que vous pouvez avoir des données écrites dans un PipedOutputStream pour devenir disponibles via un PipedInputStream associé.
Il y a quelque temps, j'ai fait face à une situation similaire lorsque je traitais avec des bibliothèques tierces qui exigeaient qu'une instance InputStream leur soit transmise au lieu d'une instance OutputStream.
La façon dont j'ai résolu ce problème consiste à utiliser PipedInputStream et PipedOutputStream.
Au fait, ils sont difficiles à utiliser et vous devez utiliser le multithreading pour obtenir ce que vous voulez. J'ai récemment publié une implémentation sur github que vous pouvez utiliser.
Voici le lien . Vous pouvez parcourir le wiki pour comprendre comment l'utiliser.