Comment voulez-vous fusionner les deux flux d'entrée en Java?

Comme le faisait remarquer, il n'est pas clair ce que tu veux dire par fusion.

La prise d'entrée disponibles de façon "aléatoire" de l'un ou l'autre est compliquée par InputStream.available pas nécessairement de vous donner une réponse utile et le blocage du comportement des cours d'eau. Vous auriez besoin de deux fils à la lecture du flux, puis de passer à l'arrière de données via java.io.Piped(In|Out)putStream (bien que ces classes ont des problèmes). Sinon pour certains types de flux, il peut être possible d'utiliser une interface différente, par exemple java.nio non bloquant les canaux.

Si vous souhaitez que le contenu complet du premier flux d'entrée suivie par la seconde: new java.io.SequenceInputStream(s1, s2).

Vous pouvez écrire une coutume InputStream de mise en œuvre qui fait cela. Exemple:

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

Ce code n'est pas testé, donc, votre kilométrage peut varier.

Voici un MVar de mise en œuvre spécifiques à des tableaux d'octets (assurez-vous d'ajouter votre propre définition du package). À partir d'ici, c'est facile à écrire un flux d'entrée sur les flux fusionnés. Je peux poster que si demandé.

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