wake-up-neo.com

Wie konvertiere ich OutputStream in InputStream?

Ich befinde mich in der Entwicklungsphase, wo ich zwei Module habe und von einem habe ich eine OutputStream ausgegeben und das zweite, das nur InputStream akzeptiert. Wissen Sie, wie Sie OutputStream in InputStream konvertieren (nicht umgekehrt, ich meine wirklich so), dass ich diese beiden Teile verbinden kann?

Vielen Dank

270
Waypoint

Eine OutputStream ist eine, in die Sie Daten schreiben. Wenn ein Modul eine OutputStream verfügbar macht, wird erwartet, dass am anderen Ende etwas gelesen wird.

Andererseits zeigt eine InputStream etwas an, dass Sie diesen Stream abhören müssen, und es gibt Daten, die Sie lesen können. 

So ist es möglich, eine InputStream mit einer OutputStream zu verbinden.

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

Wie jemand anderes bemerkte, können Sie dies mit der copy()-Methode aus IOUtils tun. Es macht keinen Sinn, den anderen Weg zu gehen ... hoffentlich macht das Sinn

UPDATE:

Je mehr ich darüber nachdenke, desto mehr kann ich erkennen, dass dies tatsächlich eine Anforderung ist. Ich kenne einige der erwähnten Kommentare Piped input/ouput, aber es gibt auch eine andere Möglichkeit.

Wenn der freigelegte Ausgabestrom eine ByteArrayOutputStream ist, können Sie immer den vollständigen Inhalt abrufen, indem Sie die toByteArray()-Methode aufrufen. Dann können Sie mithilfe der Unterklasse ByteArrayInputStream einen Eingabestrom-Wrapper erstellen. Bei diesen beiden handelt es sich um Pseudo-Streams, die im Grunde nur ein Byte-Array umschließen. Die Verwendung der Streams auf diese Weise ist daher technisch möglich, aber für mich ist es immer noch sehr seltsam ...

94
Java Drinker

Es scheint viele Links und ähnliches zu geben, aber keinen eigentlichen Code, der Pipes verwendet. Der Vorteil der Verwendung von Java.io.PipedInputStream und Java.io.PipedOutputStream / besteht darin, dass kein zusätzlicher Speicherplatz verbraucht wird. ByteArrayOutputStream.toByteArray() gibt eine Kopie des ursprünglichen Puffers zurück, was bedeutet, dass Sie jetzt zwei Kopien davon haben, was immer Sie im Speicher haben. Wenn Sie dann in ein InputStream schreiben, haben Sie jetzt drei Kopien der Daten.

Der 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
            originalByteArrayOutputStream.writeTo(out);
        } catch (IOException e) {
            // logging and exception handling should go here
        }
    }
}).start();

In diesem Code wird davon ausgegangen, dass originalByteArrayOutputStream ein ByteArrayOutputStream ist, da es normalerweise der einzige verwendbare Ausgabestrom ist, sofern Sie nicht in eine Datei schreiben. Ich hoffe das hilft! Das Tolle daran ist, dass es sich in einem separaten Thread befindet und auch parallel arbeitet. Was auch immer Ihren Eingabestream verbraucht, wird auch aus Ihrem alten Ausgabestream gestreamt. Dies ist vorteilhaft, da der Puffer kleiner bleiben kann und Sie weniger Latenz und weniger Speicher benötigen.

153
mikeho

Da Eingabe- und Ausgabeströme nur Start- und Endpunkt sind, können Sie die Daten temporär in einem Byte-Array speichern. Sie müssen also die Zwischenvariable ByteArrayOutputStream erstellen, aus der Sie byte[] erstellen, die als Eingabe für die neue ByteArrayInputStream verwendet wird. 

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

Ich hoffe es hilft.

22
BorutT

Sie benötigen eine Zwischenklasse, die zwischenspeichert. Bei jedem Aufruf von InputStream.read(byte[]...) füllt die Pufferungsklasse das übergebene Byte-Array mit dem nächsten von OutputStream.write(byte[]...) übergebenen Block. Da die Größen der Blöcke möglicherweise nicht gleich sind, muss die Adapterklasse eine bestimmte Menge speichern, bis der Lesepuffer gefüllt ist und/oder ein Pufferüberlauf gespeichert werden kann.

Dieser Artikel enthält eine nette Aufschlüsselung verschiedener Ansätze für dieses Problem:

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

20
mckamey
ByteArrayOutputStream buffer = (ByteArrayOutputStream) aOutputStream;
byte[] bytes = buffer.toByteArray();
InputStream inputStream = new ByteArrayInputStream(bytes);
16
Vijay Gupta

Die Open Source-Bibliothek easystream unterstützt direkt die Konvertierung eines OutputStream in einen InputStream: http://io-tools.sourceforge.net/easystream/tutorial/tutorial.html

Sie listen auch andere Optionen auf: http://io-tools.sourceforge.net/easystream/OutputStream_to_InputStream.html

16
koppor

Das gleiche Problem trat bei der Konvertierung einer ByteArrayOutputStream in eine ByteArrayInputStream auf und löste dieses Problem durch Verwendung einer abgeleiteten Klasse von ByteArrayOutputStream, die eine ByteArrayInputStream zurückgeben kann, die mit dem internen Puffer der ByteArrayOutputStream initialisiert wird. Auf diese Weise wird kein zusätzlicher Speicher verwendet und die "Konvertierung" ist sehr schnell:

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

Ich habe das Zeug auf github gelegt: https://github.com/nickrussler/ByteArrayInOutStream

9
Nick Russler

Die Bibliothek io-extras kann nützlich sein. Wenn Sie beispielsweise eine InputStream mit GZIPOutputStream gzipen möchten und dies synchron (unter Verwendung der Standardpuffergröße 8192) geschehen soll:

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

Beachten Sie, dass die Bibliothek zu 100% einen Unit-Test umfasst (was natürlich wert ist!) Und sich auf Maven Central befindet. Die Maven-Abhängigkeit ist:

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

Stellen Sie sicher, dass Sie nach einer neueren Version suchen.

2
Dave Moten

Aus meiner Sicht ist Java.io.PipedInputStream/Java.io.PipedOutputStream die beste Option. In einigen Situationen möchten Sie möglicherweise ByteArrayInputStream/ByteArrayOutputStream verwenden. Das Problem ist, dass Sie den Puffer duplizieren müssen, um einen ByteArrayOutputStream in einen ByteArrayInputStream zu konvertieren. Auch ByteArrayOutpuStream/ByteArrayInputStream sind auf 2 GB beschränkt. Hier ist eine OutpuStream/InputStream-Implementierung, die ich geschrieben habe, um die Einschränkungen von ByteArrayOutputStream/ByteArrayInputStream (Scala-Code, aber für Java-Entwickler leicht verständlich) zu umgehen:

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

Einfach zu verwenden, keine Pufferduplizierung, keine 2 GB Speichergrenze

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(...)
...
0
Luc Vaillant