Wie verschmelzen Sie zwei Eingänge in Java-Streams?

Question

Mit zwei Inputstreams in Java, gibt es eine Möglichkeit, sie zu verschmelzen, so dass Sie mit einem Input beenden, die Ihnen die Ausgabe beiden Ströme gibt? Wie?

Answer 1

Wie kommentierte, es ist nicht klar, was Sie von merge bedeuten.

Unter verfügbarer Eingabe „zufällig“ von entweder kompliziert durch InputStream.available nicht unbedingt Ihnen eine nützliche Antwort zu geben und das Verhalten der Ströme zu blockieren. Sie müssten zwei Fäden aus den Bächen zu lesen und dann die Daten der Durchreise zurück, sagen wir, java.io.Piped(In|Out)putStream (obwohl diese Klassen Probleme haben). Alternativ für einige Arten von Strom kann es möglich sein, eine andere Schnittstelle zu verwenden, zum Beispiel java.nio nicht blockierende Kanäle.

Wenn Sie den vollständigen Inhalt des ersten Eingangsstroms durch die zweite gefolgt werden soll:. new java.io.SequenceInputStream(s1, s2)

Answer 2

java.io.SequenceInputStream könnte sein, was Sie brauchen. Er nimmt eine Aufzählung von Streams und ausgeben wird der Inhalt des ersten Stroms, dann die zweite, und so weiter, bis alle Ströme sind leer.

Answer 3

Sie können eine benutzerdefinierte InputStream Implementierung schreiben, der dies tut. Beispiel:

import java.io.IOException;
import java.io.InputStream;
import java.util.Collections;
import java.util.Deque;
import java.util.LinkedList;

public class CatInputStream extends InputStream {
    private final Deque<InputStream> streams;

    public CatInputStream(InputStream... streams) {
        this.streams = new LinkedList<InputStream>();
        Collections.addAll(this.streams, streams);
    }

    private void nextStream() throws IOException {
        streams.removeFirst().close();
    }

    @Override
    public int read() throws IOException {
        int result = -1;
        while (!streams.isEmpty()
                && (result = streams.getFirst().read()) == -1) {
            nextStream();
        }
        return result;
    }

    @Override
    public int read(byte b[], int off, int len) throws IOException {
        int result = -1;
        while (!streams.isEmpty()
                && (result = streams.getFirst().read(b, off, len)) == -1) {
            nextStream();
        }
        return result;
    }

    @Override
    public long skip(long n) throws IOException {
        long skipped = 0L;
        while (skipped < n && !streams.isEmpty()) {
            int thisSkip = streams.getFirst().skip(n - skipped);
            if (thisSkip > 0)
                skipped += thisSkip;
            else
                nextStream();
        }
        return skipped;
    }

    @Override
    public int available() throws IOException {
        return streams.isEmpty() ? 0 : streams.getFirst().available();
    }

    @Override
    public void close() throws IOException {
        while (!streams.isEmpty())
            nextStream();
    }
}

Dieser Code ist nicht getestet, so die Leistung kann variieren.

Answer 4

Nicht, dass ich denken kann. Sie würden wahrscheinlich gelesen haben, um den Inhalt der beiden Strom in ein byte [] und dann eine ByteArrayInputStream aus, dass erstellen.

Answer 5

Hier ist ein MVar Implementierung speziell für Byte-Arrays (stellen Sie sicher, dass Ihre eigene Paketdefinition hinzufügen). Von hier aus ist es trivial einen Eingangsstrom auf fusionierte Ströme zu schreiben. Das kann ich auch schreiben, falls gewünscht.

import java.nio.ByteBuffer;

public final class MVar {

  private static enum State {
    EMPTY, ONE, MANY
  }

  private final Object lock;

  private State state;

  private byte b;

  private ByteBuffer bytes;
  private int length;

  public MVar() {
    lock = new Object();
    state = State.EMPTY;
  }

  public final void put(byte b) {
    synchronized (lock) {
      while (state != State.EMPTY) {
        try {
          lock.wait();
        } catch (InterruptedException e) {}
      }
      this.b = b;
      state = State.ONE;
      lock.notifyAll();
    }
  }

  public final void put(byte[] bytes, int offset, int length) {
    if (length == 0) {
      return;
    }
    synchronized (lock) {
      while (state != State.EMPTY) {
        try {
          lock.wait();
        } catch (InterruptedException e) {}
      }
      this.bytes = ByteBuffer.allocateDirect(length);
      this.bytes.put(bytes, offset, length);
      this.bytes.position(0);
      this.length = length;
      state = State.MANY;
      lock.notifyAll();
    }
  }

  public final byte take() {
    synchronized (lock) {
      while (state == State.EMPTY) {
        try {
          lock.wait();
        } catch (InterruptedException e) {}
      }
      switch (state) {
      case ONE: {
        state = State.EMPTY;
        byte b = this.b;
        lock.notifyAll();
        return b;
      }
      case MANY: {
        byte b = bytes.get();
        state = --length <= 0 ? State.EMPTY : State.MANY;
        lock.notifyAll();
        return b;
      }
      default:
        throw new AssertionError();
      }
    }
  }

  public final int take(byte[] bytes, int offset, int length) {
    if (length == 0) {
      return 0;
    }
    synchronized (lock) {
      while (state == State.EMPTY) {
        try {
          lock.wait();
        } catch (InterruptedException e) {}
      }
      switch (state) {
      case ONE:
        bytes[offset] = b;
        state = State.EMPTY;
        lock.notifyAll();
        return 1;
      case MANY:
        if (this.length > length) {
          this.bytes.get(bytes, offset, length);
          this.length = this.length - length;
          synchronized (lock) {
            lock.notifyAll();
          }
          return length;
        }
        this.bytes.get(bytes, offset, this.length);
        this.bytes = null;
        state = State.EMPTY;
        length = this.length;
        lock.notifyAll();
        return length;
      default:
        throw new AssertionError();
      }
    }
  }
}