Bagaimana cara mengubah OutputStream ke InputStream?

Saya berada pada tahap pengembangan, di mana saya memiliki dua modul dan dari satu saya mendapat output sebagai yang OutputStreamkedua, yang hanya menerima InputStream. Apakah Anda tahu cara mengonversi OutputStreamke InputStream(bukan sebaliknya, maksud saya benar-benar seperti ini) sehingga saya dapat menghubungkan kedua bagian ini?

Terima kasih

An OutputStreamadalah tempat Anda menulis data. Jika beberapa modul memaparkan OutputStream, harapannya adalah ada sesuatu yang membaca di ujung lainnya.

InputStreamDi lain pihak, sesuatu yang memperlihatkan suatu , menunjukkan bahwa Anda perlu mendengarkan aliran ini, dan akan ada data yang dapat Anda baca.

Jadi dimungkinkan untuk menghubungkan InputStreamkeOutputStream

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

Sebagai seseorang yang diceritakan, inilah yang dapat Anda lakukan dengan copy()metode dari IOUtils . Tidak masuk akal untuk pergi ke arah lain ... semoga ini masuk akal

MEMPERBARUI:

Tentu saja semakin saya memikirkan hal ini, semakin saya bisa melihat bagaimana ini sebenarnya menjadi persyaratan. Saya tahu beberapa komentar menyebutkan Pipedinput / ouput stream, tetapi ada kemungkinan lain.

Jika aliran output yang diekspos adalah ByteArrayOutputStream, maka Anda selalu bisa mendapatkan konten lengkap dengan memanggil toByteArray()metode. Kemudian Anda dapat membuat pembungkus aliran input dengan menggunakan ByteArrayInputStreamsub-kelas. Keduanya adalah pseudo-stream, keduanya pada dasarnya hanya membungkus array byte. Menggunakan aliran seperti ini, oleh karena itu, secara teknis mungkin, tetapi bagi saya itu masih sangat aneh ...

Tampaknya ada banyak tautan dan hal-hal lain semacam itu, tetapi tidak ada kode sebenarnya yang menggunakan pipa. Keuntungan menggunakan java.io.PipedInputStreamdan java.io.PipedOutputStreamtidak ada konsumsi memori tambahan. ByteArrayOutputStream.toByteArray()mengembalikan salinan buffer asli, sehingga itu berarti bahwa apa pun yang Anda miliki di memori, kini Anda memiliki dua salinannya. Kemudian menulis ke InputStreamsarana berarti Anda sekarang memiliki tiga salinan data.

Kode:

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

Kode ini mengasumsikan bahwa originalByteArrayOutputStreama adalah ByteArrayOutputStreamkarena biasanya merupakan satu-satunya aliran output yang dapat digunakan, kecuali jika Anda menulis ke file. Saya harap ini membantu! Hal yang hebat tentang ini adalah karena berada di utas terpisah, ia juga bekerja secara paralel, jadi apa pun yang mengonsumsi aliran input Anda akan mengalir keluar dari aliran output lama Anda juga. Itu menguntungkan karena buffer bisa tetap lebih kecil dan Anda akan memiliki latensi lebih sedikit dan penggunaan memori lebih sedikit.

Karena input dan output stream hanya titik awal dan akhir, solusinya adalah menyimpan data sementara dalam array byte. Jadi, Anda harus membuat perantara ByteArrayOutputStream, dari mana Anda membuat byte[]yang digunakan sebagai input untuk yang baru 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);
}

Semoga ini bisa membantu.

Anda akan membutuhkan kelas menengah yang akan buffer antara. Setiap kali InputStream.read(byte[]...)dipanggil, kelas buffering akan mengisi array byte yang dikirimkan dengan potongan berikutnya yang dilewati OutputStream.write(byte[]...). Karena ukuran chunks mungkin tidak sama, kelas adaptor perlu menyimpan jumlah tertentu sampai cukup untuk mengisi buffer baca dan / atau dapat menyimpan buffer overflow.

Artikel ini memiliki uraian bagus dari beberapa pendekatan berbeda untuk masalah ini:

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

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

The easystream perpustakaan open source memiliki dukungan langsung untuk mengkonversi OutputStream untuk InputStream: http://io-tools.sourceforge.net/easystream/tutorial/tutorial.html

Mereka juga mencantumkan opsi lain: http://io-tools.sourceforge.net/easystream/OutputStream_to_InputStream.html

Saya mengalami masalah yang sama dengan mengkonversi ByteArrayOutputStreamke ByteArrayInputStreamdan menyelesaikannya dengan menggunakan kelas turunan dari ByteArrayOutputStreamyang mampu mengembalikan ByteArrayInputStreamyang diinisialisasi dengan buffer internal ByteArrayOutputStream. Dengan cara ini tidak ada memori tambahan yang digunakan dan 'konversi' sangat cepat:

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

Saya meletakkan barang-barang di github: https://github.com/nickrussler/ByteArrayInOutStream

Perpustakaan io-ekstra mungkin berguna. Misalnya jika Anda ingin gzip suatu InputStreampenggunaan GZIPOutputStreamdan Anda ingin itu terjadi secara serempak (menggunakan ukuran buffer default 8192):

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

Perhatikan bahwa perpustakaan memiliki cakupan uji unit 100% (untuk apa yang tentunya layak!) Dan ada di Maven Central. Ketergantungan Maven adalah:

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

Pastikan untuk memeriksa versi yang lebih baru.

Dari sudut pandang saya, java.io.PipedInputStream / java.io.PipedOutputStream adalah pilihan terbaik untuk dipertimbangkan. Dalam beberapa situasi Anda mungkin ingin menggunakan ByteArrayInputStream / ByteArrayOutputStream. Masalahnya adalah Anda perlu menduplikasi buffer untuk mengkonversi ByteArrayOutputStream ke ByteArrayInputStream. ByteArrayOutpuStream / ByteArrayInputStream juga terbatas hingga 2GB. Berikut ini adalah implementasi OutpuStream / InputStream yang saya tulis untuk mem-bypass keterbatasan ByteArrayOutputStream / ByteArrayInputStream (kode Scala, tetapi mudah dimengerti untuk 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
      }
    }
  }
}

Mudah digunakan, tidak ada duplikasi buffer, tidak ada batas memori 2GB

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

Jika Anda ingin membuat OutputStream dari InputStream ada satu masalah mendasar. Metode menulis ke blok OutputStream sampai selesai. Jadi hasilnya tersedia ketika metode penulisan selesai. Ini memiliki 2 konsekuensi:

1. Jika Anda hanya menggunakan satu utas, Anda harus menunggu sampai semuanya ditulis (jadi Anda perlu menyimpan data aliran dalam memori atau disk).
2. Jika Anda ingin mengakses data sebelum selesai, Anda perlu utas kedua.

Varian 1 dapat diimplementasikan menggunakan byte array atau diarsipkan. Varian 1 dapat diimplementasikan menggunakan pipies (baik secara langsung atau dengan abstraksi ekstra - mis. RingBuffer atau google lib dari komentar lain).

Memang dengan java standar tidak ada cara lain untuk menyelesaikan masalah. Setiap solusi adalah implementasi dari salah satunya.

Ada satu konsep yang disebut "kelanjutan" (lihat wikipedia untuk detail). Dalam hal ini pada dasarnya ini berarti:

• ada aliran output khusus yang mengharapkan sejumlah data tertentu
• jika jumlah tercapai, aliran memberikan kontrol ke lawannya yang merupakan aliran input khusus
• aliran input membuat jumlah data tersedia sampai terbaca, setelah itu, ia mengembalikan kontrol ke aliran output

Sementara beberapa bahasa memiliki konsep ini, untuk java Anda memerlukan "sihir". Misalnya "commons-javaflow" dari implementasi apache seperti untuk java. Kerugiannya adalah ini memerlukan beberapa modifikasi bytecode khusus pada waktu build. Jadi masuk akal untuk meletakkan semua barang di perpustakaan tambahan dengan skrip pembuatan kustom.

Posting lama tetapi mungkin membantu orang lain, Gunakan cara ini:

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

Meskipun Anda tidak dapat mengonversi OutputStream ke InputStream, java menyediakan cara menggunakan PipedOutputStream dan PipedInputStream sehingga Anda dapat memiliki data yang ditulis ke PipedOutputStream agar tersedia melalui PipedInputStream yang terkait.
Beberapa waktu yang lalu saya menghadapi situasi yang sama ketika berhadapan dengan perpustakaan pihak ketiga yang membutuhkan instance InputStream untuk diteruskan kepada mereka alih-alih instance OutputStream.
Cara saya memperbaiki masalah ini adalah dengan menggunakan PipedInputStream dan PipedOutputStream.
Omong-omong mereka sulit digunakan dan Anda harus menggunakan multithreading untuk mencapai apa yang Anda inginkan. Baru-baru ini saya menerbitkan implementasi di github yang dapat Anda gunakan.
Inilah tautannya . Anda dapat membaca wiki untuk memahami cara menggunakannya.