Odczyt strumienia wejściowego dwa razy

W zależności od tego, skąd dochodzi strumień wejściowy, możesz nie być w stanie go zresetować. Możesz sprawdzić, czy mark() i reset() są obsługiwane za pomocą markSupported().

Jeśli tak, możesz wywołać reset() na strumieniu wejściowym, aby powrócić do początku. Jeśli nie, musisz ponownie odczytać strumień wejściowy ze źródła.

Jeśli twoje InputStream wsparcie za pomocą mark, to możesz mark() swój strumień wejściowy, a następnie reset() to . jeśli twoja InputStrem nie obsługuje mark, możesz użyć klasy java.io.BufferedInputStream, więc możesz osadzić swój strumień wewnątrz BufferedInputStream w ten sposób

    InputStream bufferdInputStream = new BufferedInputStream(yourInputStream);
    bufferdInputStream.mark(some_value);
    //read your bufferdInputStream 
    bufferdInputStream.reset();
    //read it again

Możesz zawinąć strumień wejściowy za pomocą PushbackInputStream. PushbackInputStream pozwala nieprzeczytać ("Odpisz ") bajty, które zostały już przeczytane, więc możesz zrobić tak:

public class StreamTest {
  public static void main(String[] args) throws IOException {
    byte[] bytes = new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9 };

    InputStream originalStream = new ByteArrayInputStream(bytes);

    byte[] readBytes = getBytes(originalStream, 3);
    printBytes(readBytes); // prints: 1 2 3

    readBytes = getBytes(originalStream, 3);
    printBytes(readBytes); // prints: 4 5 6

    // now let's wrap it with PushBackInputStream

    originalStream = new ByteArrayInputStream(bytes);

    InputStream wrappedStream = new PushbackInputStream(originalStream, 10); // 10 means that maximnum 10 characters can be "written back" to the stream

    readBytes = getBytes(wrappedStream, 3);
    printBytes(readBytes); // prints 1 2 3

    ((PushbackInputStream) wrappedStream).unread(readBytes, 0, readBytes.length);

    readBytes = getBytes(wrappedStream, 3);
    printBytes(readBytes); // prints 1 2 3


  }

  private static byte[] getBytes(InputStream is, int howManyBytes) throws IOException {
    System.out.print("Reading stream: ");

    byte[] buf = new byte[howManyBytes];

    int next = 0;
    for (int i = 0; i < howManyBytes; i++) {
      next = is.read();
      if (next > 0) {
        buf[i] = (byte) next;
      }
    }
    return buf;
  }

  private static void printBytes(byte[] buffer) throws IOException {
    System.out.print("Reading stream: ");

    for (int i = 0; i < buffer.length; i++) {
      System.out.print(buffer[i] + " ");
    }
    System.out.println();
  }


}

Należy pamiętać, że PushbackInputStream przechowuje wewnętrzny bufor bajtów, więc tak naprawdę tworzy bufor w pamięci, który przechowuje bajty "zapisane wstecz".

Znając to podejście możemy pójść dalej i połączyć je z FilterInputStream. FilterInputStream przechowuje oryginalny strumień wejściowy jako delegat. Pozwala to na utworzenie nowej definicji klasy, która pozwala na Automatyczne "nieprzeczytane" oryginalne dane. Definicja tej klasy jest następująca:

public class TryReadInputStream extends FilterInputStream {
  private final int maxPushbackBufferSize;

  /**
  * Creates a <code>FilterInputStream</code>
  * by assigning the  argument <code>in</code>
  * to the field <code>this.in</code> so as
  * to remember it for later use.
  *
  * @param in the underlying input stream, or <code>null</code> if
  *           this instance is to be created without an underlying stream.
  */
  public TryReadInputStream(InputStream in, int maxPushbackBufferSize) {
    super(new PushbackInputStream(in, maxPushbackBufferSize));
    this.maxPushbackBufferSize = maxPushbackBufferSize;
  }

  /**
   * Reads from input stream the <code>length</code> of bytes to given buffer. The read bytes are still avilable
   * in the stream
   *
   * @param buffer the destination buffer to which read the data
   * @param offset  the start offset in the destination <code>buffer</code>
   * @aram length how many bytes to read from the stream to buff. Length needs to be less than
   *        <code>maxPushbackBufferSize</code> or IOException will be thrown
   *
   * @return number of bytes read
   * @throws java.io.IOException in case length is
   */
  public int tryRead(byte[] buffer, int offset, int length) throws IOException {
    validateMaxLength(length);

    // NOTE: below reading byte by byte instead of "int bytesRead = is.read(firstBytes, 0, maxBytesOfResponseToLog);"
    // because read() guarantees to read a byte

    int bytesRead = 0;

    int nextByte = 0;

    for (int i = 0; (i < length) && (nextByte >= 0); i++) {
      nextByte = read();
      if (nextByte >= 0) {
        buffer[offset + bytesRead++] = (byte) nextByte;
      }
    }

    if (bytesRead > 0) {
      ((PushbackInputStream) in).unread(buffer, offset, bytesRead);
    }

    return bytesRead;

  }

  public byte[] tryRead(int maxBytesToRead) throws IOException {
    validateMaxLength(maxBytesToRead);

    ByteArrayOutputStream baos = new ByteArrayOutputStream(); // as ByteArrayOutputStream to dynamically allocate internal bytes array instead of allocating possibly large buffer (if maxBytesToRead is large)

    // NOTE: below reading byte by byte instead of "int bytesRead = is.read(firstBytes, 0, maxBytesOfResponseToLog);"
    // because read() guarantees to read a byte

    int nextByte = 0;

    for (int i = 0; (i < maxBytesToRead) && (nextByte >= 0); i++) {
      nextByte = read();
      if (nextByte >= 0) {
        baos.write((byte) nextByte);
      }
    }

    byte[] buffer = baos.toByteArray();

    if (buffer.length > 0) {
      ((PushbackInputStream) in).unread(buffer, 0, buffer.length);
    }

    return buffer;

  }

  private void validateMaxLength(int length) throws IOException {
    if (length > maxPushbackBufferSize) {
      throw new IOException(
        "Trying to read more bytes than maxBytesToRead. Max bytes: " + maxPushbackBufferSize + ". Trying to read: " +
        length);
    }
  }

}

Ta klasa ma dwie metody. Jeden do wczytania do istniejącego bufora (definicja jest analogiczna do wywołania public int read(byte b[], int off, int len) klasy InputStream). Drugi, który zwraca nowy bufor (może to być bardziej efektywne, jeśli rozmiar bufora do odczytu jest nieznany).

Teraz zobaczmy naszą klasę w działaniu:

public class StreamTest2 {
  public static void main(String[] args) throws IOException {
    byte[] bytes = new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9 };

    InputStream originalStream = new ByteArrayInputStream(bytes);

    byte[] readBytes = getBytes(originalStream, 3);
    printBytes(readBytes); // prints: 1 2 3

    readBytes = getBytes(originalStream, 3);
    printBytes(readBytes); // prints: 4 5 6

    // now let's use our TryReadInputStream

    originalStream = new ByteArrayInputStream(bytes);

    InputStream wrappedStream = new TryReadInputStream(originalStream, 10);

    readBytes = ((TryReadInputStream) wrappedStream).tryRead(3); // NOTE: no manual call to "unread"(!) because TryReadInputStream handles this internally
    printBytes(readBytes); // prints 1 2 3

    readBytes = ((TryReadInputStream) wrappedStream).tryRead(3); 
    printBytes(readBytes); // prints 1 2 3

    readBytes = ((TryReadInputStream) wrappedStream).tryRead(3);
    printBytes(readBytes); // prints 1 2 3

    // we can also call normal read which will actually read the bytes without "writing them back"
    readBytes = getBytes(wrappedStream, 3);
    printBytes(readBytes); // prints 1 2 3

    readBytes = getBytes(wrappedStream, 3);
    printBytes(readBytes); // prints 4 5 6

    readBytes = ((TryReadInputStream) wrappedStream).tryRead(3); // now we can try read next bytes
    printBytes(readBytes); // prints 7 8 9

    readBytes = ((TryReadInputStream) wrappedStream).tryRead(3); 
    printBytes(readBytes); // prints 7 8 9


  }



}

Do dzielenia InputStream na dwa, unikając ładowania wszystkich danych w pamięci , a następnie przetwarzania ich niezależnie:

1. Stwórz parę OutputStream, dokładnie: PipedOutputStream
2. Połącz każdy strumień PipedOutputStream z pipedinputstream, te PipedInputStream są zwracane InputStream.
3. Połącz strumień wejściowy sourcingu z właśnie utworzonym OutputStream. Tak więc wszystko, co czyta się z źródła InputStream, byłoby napisane w obu OutputStream. Nie trzeba tego wdrażać, bo to jest już w TeeInputStream (commons.io).

4. W oddzielonym wątku odczytuje cały strumień wejściowy sourcingu i domyślnie dane wejściowe są przesyłane do docelowych strumieni wejściowych.

   public static final List<InputStream> splitInputStream(InputStream input) 
       throws IOException 
   { 
       Objects.requireNonNull(input);      
   
       PipedOutputStream pipedOut01 = new PipedOutputStream();
       PipedOutputStream pipedOut02 = new PipedOutputStream();
   
       List<InputStream> inputStreamList = new ArrayList<>();
       inputStreamList.add(new PipedInputStream(pipedOut01));
       inputStreamList.add(new PipedInputStream(pipedOut02));
   
       TeeOutputStream tout = new TeeOutputStream(pipedOut01, pipedOut02);
   
       TeeInputStream tin = new TeeInputStream(input, tout, true);
   
       Executors.newSingleThreadExecutor().submit(tin::readAllBytes);  
   
       return Collections.unmodifiableList(inputStreamList);
   }
   

Należy pamiętać, aby zamknąć strumienie wejściowe po zużyciu i zamknąć wątek, który działa: TeeInputStream.readAllBytes()

W przypadku, trzeba podzielić go na wiele InputStream, zamiast dwóch. Zamień w poprzednim fragmencie kodu klasę TeeOutputStream na własną implementacja, która hermetyzowałaby List<OutputStream> i nadpisywałaby OutputStream interfejs:

public final class TeeListOutputStream extends OutputStream {
    private final List<? extends OutputStream> branchList;

    public TeeListOutputStream(final List<? extends OutputStream> branchList) {
        Objects.requireNonNull(branchList);
        this.branchList = branchList;
    }

    @Override
    public synchronized void write(final int b) throws IOException {
        for (OutputStream branch : branchList) {
            branch.write(b);
        }
    }

    @Override
    public void flush() throws IOException {
        for (OutputStream branch : branchList) {
            branch.flush();
        }
    }

    @Override
    public void close() throws IOException {
        for (OutputStream branch : branchList) {
            branch.close();
        }
    }
}

Jeśli korzystasz z implementacji InputStream, Możesz sprawdzić wynik InputStream#markSupported() to mówi ci, czy możesz użyć metody mark() / reset().

Jeśli możesz oznaczyć strumień podczas czytania, zadzwoń reset(), aby wrócić do początku.

Innym rozwiązaniem byłoby przekonwertowanie strumienia wejściowego na tablicę bajtów, a następnie iterację nad tablicą tyle razy, ile potrzebujesz. Można znaleźć kilka rozwiązań w tym poście konwertuje strumień wejściowy do tablicy bajtów w Javie za pomocą bibliotek stron trzecich lub nie. Uwaga, jeśli czytana zawartość jest zbyt duża, możesz mieć problemy z pamięcią.

Na koniec, jeśli potrzebujesz odczytać obrazek, użyj :

BufferedImage image = ImageIO.read(new URL("http://www.example.com/images/toto.jpg"));

Za pomocą ImageIO#read(java.net.URL) umożliwia również korzystanie z pamięci podręcznej.

Konwertuj strumień wejściowy na bajty, a następnie przekaż go do funkcji savefile, gdzie montujesz to samo do strumienia wejściowego. Również w oryginalnej funkcji Użyj bajtów do użycia w innych zadaniach

A może:

if (stream.markSupported() == false) {

        // lets replace the stream object
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        IOUtils.copy(stream, baos);
        stream.close();
        stream = new ByteArrayInputStream(baos.toByteArray());
        // now the stream should support 'mark' and 'reset'

    }

W przypadku, gdy ktoś działa w aplikacji Spring Boot i chcesz przeczytać ciało odpowiedzi RestTemplate (dlatego chcę przeczytać strumień dwa razy), istnieje czysty (er)sposób na to.

Po pierwsze, musisz użyć sprężyny StreamUtils, Aby skopiować strumień do ciągu znaków:

String text = StreamUtils.copyToString(response.getBody(), Charset.defaultCharset()))

ClientHttpRequestFactory factory = new BufferingClientHttpRequestFactory(new SimpleClientHttpRequestFactory());
RestTemplate restTemplate = new RestTemplate(factory);

Lub, jeśli używasz factory bean, to (to jest Kotlin, ale niemniej jednak): {]}

@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
fun createRestTemplate(): RestTemplate = RestTemplateBuilder()
  .requestFactory { BufferingClientHttpRequestFactory(SimpleClientHttpRequestFactory()) }
  .additionalInterceptors(loggingInterceptor)
  .build()

Źródło: https://objectpartners.com/2018/03/01/log-your-resttemplate-request-and-response-without-destroying-the-body/

Jeśli używasz RestTemplate do wykonywania połączeń http, po prostu dodaj interceptor. Ciało odpowiedzi jest buforowane przez implementację ClientHttpResponse. Teraz inputstream może być pobierany z respose tyle razy ile potrzebujemy

ClientHttpRequestInterceptor interceptor =  new ClientHttpRequestInterceptor() {

            @Override
            public ClientHttpResponse intercept(HttpRequest request, byte[] body,
                    ClientHttpRequestExecution execution) throws IOException {
                ClientHttpResponse  response = execution.execute(request, body);

                  // additional work before returning response
                  return response 
            }
        };

    // Add the interceptor to RestTemplate Instance 

         restTemplate.getInterceptors().add(interceptor); 

ByteArrayInputStream ins = new ByteArrayInputStream("Hello".getBytes());
System.out.println("ins.available() at begining:: " + ins.available());
ins.mark(0);
// Read input stream for some operations
System.out.println("ins.available() after reading :: " + ins.available());
    ins.reset();
    System.out.println("ins.available() after resetting :: " + ins.available());
    // ins is ready for reading once again.