How to combine two input streams in Java?

Question

Having two InputStreams in Java, is there any way to merge them, so you end up with one InputStream that gives you the output of both streams? How?

Answer 1

As commented, it is unclear what you mean by merging.

Accepting an available input “randomly” from either complex with InputStream.available does not necessarily give you a useful answer and blocks the behavior of the streams. You will need two streams to read from the streams, and then transfer the data through, say, java.io.Piped(In|Out)putStream (although these classes have problems). Alternatively, for some types of streams it may be possible to use a different interface, for example java.nio non-blocking channels.

If you want the full contents of the first input stream to be followed by the second: new java.io.SequenceInputStream(s1, s2) .

Answer 2

java.io.SequenceInputStream might be what you need. It takes an enumeration of threads and displays the contents of the first thread, then the second, and so on, until all threads are empty.

Answer 3

You can write a custom implementation of InputStream that will do this. Example:

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

This code has not been verified, so your mileage may vary.

Answer 4

Not that I could think. You will probably have to read the contents of the two streams in bytes [], and then create a ByteArrayInputStream from this.

Answer 5

Here is an MVar implementation specific to byte arrays (be sure to add your own package definition). From here it is trivial to write the input stream over the combined streams. I can send this too if required.

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