Jak przekonwertować OutputStream na InputStream?

Jestem na etapie rozwoju, gdzie mam dwa moduły i z jednego dostałem wyjście jako OutputStreamdrugi i drugi, który akceptuje tylko InputStream. Czy wiesz, jak przekonwertować OutputStreamna InputStream(nie odwrotnie, mam na myśli naprawdę w ten sposób), że będę w stanie połączyć te dwie części?

Dzięki

Jest OutputStreamto miejsce, w którym zapisujesz dane. Jeśli jakiś moduł ujawnia OutputStream, oczekuje się, że na drugim końcu jest coś do odczytania.

InputStreamZ drugiej strony coś, co ujawnia , wskazuje, że będziesz musiał wysłuchać tego strumienia i będą dane, które możesz odczytać.

Możliwe jest więc podłączenie InputStreamdoOutputStream

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

Jak ktoś wspomniał, na to pozwala copy()metoda IOUtils . Nie ma sensu iść w drugą stronę ... mam nadzieję, że ma to jakiś sens

AKTUALIZACJA:

Oczywiście im więcej o tym myślę, tym bardziej widzę, jak to właściwie byłoby wymaganiem. Znam niektóre komentarze wspomniane Pipedstrumienie wejścia / wyjścia, ale jest inna możliwość.

Jeśli ujawniony strumień wyjściowy to a ByteArrayOutputStream, zawsze można uzyskać pełną zawartość, wywołując toByteArray()metodę. Następnie możesz utworzyć opakowanie strumienia wejściowego za pomocą ByteArrayInputStreampodklasy. Te dwa są pseudo-strumieniami, oba po prostu owijają tablicę bajtów. Dlatego korzystanie ze strumieni w ten sposób jest technicznie możliwe, ale dla mnie nadal jest bardzo dziwne ...

Wydaje się, że istnieje wiele linków i innych podobnych rzeczy, ale nie ma rzeczywistego kodu używającego potoków. Zaletą używania java.io.PipedInputStreami java.io.PipedOutputStreamjest to, że nie ma dodatkowego zużycia pamięci. ByteArrayOutputStream.toByteArray()zwraca kopię oryginalnego bufora, co oznacza, że ​​cokolwiek masz w pamięci, masz teraz dwie kopie. Pisząc do InputStreamoznacza, że ​​masz teraz trzy kopie danych.

Kod:

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

Ten kod zakłada, że originalByteArrayOutputStreamjest to ByteArrayOutputStreamjako, ponieważ jest to zwykle jedyny możliwy strumień wyjściowy, chyba że piszesz do pliku. Mam nadzieję, że to pomoże! Wspaniałą rzeczą jest to, że ponieważ znajduje się w osobnym wątku, działa również równolegle, więc cokolwiek zużywa strumień wejściowy, będzie również przesyłane strumieniowo ze starego strumienia wyjściowego. Jest to korzystne, ponieważ bufor może pozostać mniejszy, a Ty będziesz mieć mniejsze opóźnienia i mniejsze zużycie pamięci.

Ponieważ strumienie wejściowe i wyjściowe są tylko punktem początkowym i końcowym, rozwiązaniem jest tymczasowe przechowywanie danych w tablicy bajtów. Musisz więc utworzyć pośredni ByteArrayOutputStream, z którego utworzysz, byte[]który będzie używany jako dane wejściowe dla nowego 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);
}

Mam nadzieję, że to pomoże.

Będziesz potrzebował klasy pośredniej, która będzie buforować między. Przy każdym InputStream.read(byte[]...)wywołaniu klasa buforująca wypełnia przekazaną tablicę bajtów kolejną porcją przekazywaną z OutputStream.write(byte[]...). Ponieważ rozmiary porcji mogą nie być takie same, klasa adaptera będzie musiała przechowywać pewną ilość, dopóki nie będzie wystarczająca do zapełnienia bufora odczytu i / lub będzie w stanie zapisać wszelkie przepełnienia bufora.

W tym artykule opisano kilka różnych podejść do tego problemu:

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

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

Easystream biblioteka open source ma bezpośredniego wsparcia przekonwertować OutputStream do InputStream: http://io-tools.sourceforge.net/easystream/tutorial/tutorial.html

Wymieniają także inne opcje: http://io-tools.sourceforge.net/easystream/OutputStream_to_InputStream.html

Napotkałem ten sam problem z konwersją a ByteArrayOutputStreamna a ByteArrayInputStreami rozwiązałem go za pomocą klasy pochodnej, z ByteArrayOutputStreamktórej jest w stanie zwrócić a ByteArrayInputStreamzainicjowany wewnętrznym buforem ByteArrayOutputStream. W ten sposób nie jest wykorzystywana dodatkowa pamięć, a „konwersja” jest bardzo szybka:

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

Kładę rzeczy na github: https://github.com/nickrussler/ByteArrayInOutStream

Biblioteka io-dodatków może być przydatna. Na przykład, jeśli chcesz gzip InputStreamużyć GZIPOutputStreami chcesz, aby stało się to synchronicznie (przy użyciu domyślnego rozmiaru bufora 8192):

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

Zauważ, że biblioteka ma 100% pokrycia testem jednostkowym (oczywiście, co jest tego warte!) I jest dostępna w Maven Central. Zależność Maven to:

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

Koniecznie sprawdź późniejszą wersję.

Z mojego punktu widzenia java.io.PipedInputStream / java.io.PipedOutputStream to najlepsza opcja do rozważenia. W niektórych sytuacjach możesz chcieć użyć ByteArrayInputStream / ByteArrayOutputStream. Problem polega na tym, że musisz zduplikować bufor, aby przekonwertować ByteArrayOutputStream na ByteArrayInputStream. Również ByteArrayOutpuStream / ByteArrayInputStream są ograniczone do 2 GB. Oto implementacja OutpuStream / InputStream, którą napisałem, aby ominąć ograniczenia ByteArrayOutputStream / ByteArrayInputStream (kod Scala, ale łatwo zrozumiały dla programistów 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
      }
    }
  }
}

Łatwy w użyciu, bez powielania buforów, bez limitu pamięci 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(...)
...

Jeśli chcesz utworzyć OutputStream z InputStream, istnieje jeden podstawowy problem. Metoda zapisująca do bloków OutputStream, dopóki nie zostanie wykonana. Tak więc wynik jest dostępny po zakończeniu metody pisania. Ma to 2 konsekwencje:

1. Jeśli używasz tylko jednego wątku, musisz poczekać, aż wszystko zostanie zapisane (więc musisz zapisać dane strumienia w pamięci lub na dysku).
2. Jeśli chcesz uzyskać dostęp do danych przed ich zakończeniem, potrzebujesz drugiego wątku.

Wariant 1 może być zaimplementowany przy użyciu tablic bajtowych lub złożony. Wariant 1 można zaimplementować za pomocą pipetów (bezpośrednio lub z dodatkową abstrakcją - np. RingBuffer lub biblioteka Google z innego komentarza).

Rzeczywiście w standardowej Javie nie ma innego sposobu rozwiązania problemu. Każde rozwiązanie jest implementacją jednego z nich.

Istnieje jedna koncepcja zwana „kontynuacją” (szczegóły w Wikipedii ). W tym przypadku w zasadzie oznacza to:

• istnieje specjalny strumień wyjściowy, który oczekuje pewnej ilości danych
• jeśli kwota zostanie osiągnięta, strumień daje kontrolę nad swoim odpowiednikiem, który jest specjalnym strumieniem wejściowym
• strumień wejściowy udostępnia ilość danych, dopóki nie zostanie odczytany, a następnie przekaże sterowanie z powrotem do strumienia wyjściowego

Podczas gdy niektóre języki mają tę wbudowaną koncepcję, java potrzebuje trochę „magii”. Na przykład „commons-javaflow” z apache implementuje takie dla java. Wadą jest to, że wymaga to specjalnych modyfikacji kodu bajtowego podczas kompilacji. Sensowne byłoby więc umieszczenie wszystkich rzeczy w dodatkowej bibliotece z niestandardowymi skryptami kompilacji.

Stary post, ale może pomóc innym, użyj tego sposobu:

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

Chociaż nie można przekonwertować OutputStream na InputStream, java zapewnia sposób za pomocą PipedOutputStream i PipedInputStream, że można zapisywać dane w PipedOutputStream, aby stały się dostępne za pośrednictwem powiązanego PipedInputStream.
Kiedyś miałem do czynienia z podobną sytuacją, gdy miałem do czynienia z bibliotekami stron trzecich, które wymagały przekazania do nich instancji InputStream zamiast instancji OutputStream.
Rozwiązałem ten problem, używając PipedInputStream i PipedOutputStream.
Nawiasem mówiąc, są one trudne w użyciu i musisz używać wielowątkowości, aby osiągnąć to, czego chcesz. Niedawno opublikowałem implementację na github, z której możesz korzystać.
Oto link . Możesz przejść przez wiki, aby zrozumieć, jak z niej korzystać.