Multiple readers for InputStream in Java

Input stream work like this: once you read a portion from it, it's gone forever. You can't go back and re-read it. what you could do is something like this:

class InputStreamSplitter {
  InputStreamSplitter(InputStream toReadFrom) {
    this.reader = new InputStreamReader(toReadFrom);
  }
  void addListener(Listener l) {
    this.listeners.add(l);
  }
  void work() {
    String line = this.reader.readLine();
        while(line != null) {
      for(Listener l : this.listeners) {
        l.processLine(line);
      }
    }
  }
}

interface Listener {
  processLine(String line);
}

have all interested parties implement Listener and add them to InputStreamSplitter

Use TeeInputStream to copy all the bytes read from InputStream to secondary OutputStream, e.g. ByteArrayOutputStream.

Note: My other answer is more general (and better in my opinion).

As noted by @dimo414, the answer below requires the first reader to always be ahead of the second reader. If this is indeed the case for you, then this answer might still be preferable since it builds upon standard classes.

To create two readers that read independently from the same source, you'll have to make sure they don't consume data from the same stream.

This can be achieved by combining TeeInputStream from Apache Commons and a PipedInputStream and PipedOutputStream as follows:

import java.io.*;
import org.apache.commons.io.input.TeeInputStream;
class Test {
    public static void main(String[] args) throws IOException {

        // Create the source input stream.
        InputStream is = new FileInputStream("filename.txt");

        // Create a piped input stream for one of the readers.
        PipedInputStream in = new PipedInputStream();

        // Create a tee-splitter for the other reader.
        TeeInputStream tee = new TeeInputStream(is, new PipedOutputStream(in));

        // Create the two buffered readers.
        BufferedReader br1 = new BufferedReader(new InputStreamReader(tee));
        BufferedReader br2 = new BufferedReader(new InputStreamReader(in));

        // Do some interleaved reads from them.
        System.out.println("One line from br1:");
        System.out.println(br1.readLine());
        System.out.println();

        System.out.println("Two lines from br2:");
        System.out.println(br2.readLine());
        System.out.println(br2.readLine());
        System.out.println();

        System.out.println("One line from br1:");
        System.out.println(br1.readLine());
        System.out.println();
    }
}

Output:

One line from br1:
Line1: Lorem ipsum dolor sit amet,      <-- reading from start

Two lines from br2:
Line1: Lorem ipsum dolor sit amet,      <-- reading from start
Line2: consectetur adipisicing elit,

One line from br1:
Line2: consectetur adipisicing elit,    <-- resumes on line 2

As you've probably noted, once you've read a byte from an input stream, it's gone forever (unless you've saved it somewhere yourself).

The solution below does save the bytes until all subscribing input streams have read it.

It works as follows:

// Create a SplittableInputStream from the originalStream
SplittableInputStream is  = new SplittableInputStream(originalStream);

// Fork this to get more input streams reading independently from originalStream
SplittableInputStream is2 = is.split();
SplittableInputStream is3 = is.split();

Each time is is split() it will yield a new InputStream that will read the bytes from the point where is was split.

The SplittableInputStream looks as follows (copy'n'paste away!):

class SplittableInputStream extends InputStream {

    // Almost an input stream: The read-method takes an id.
    static class MultiplexedSource {

        static int MIN_BUF = 4096;

        // Underlying source
        private InputStream source;

        // Read positions of each SplittableInputStream
        private List<Integer> readPositions = new ArrayList<>();

        // Data to be read by the SplittableInputStreams
        int[] buffer = new int[MIN_BUF];

        // Last valid position in buffer
        int writePosition = 0;

        public MultiplexedSource(InputStream source) {
            this.source = source;
        }

        // Add a multiplexed reader. Return new reader id.
        int addSource(int splitId) {
            readPositions.add(splitId == -1 ? 0 : readPositions.get(splitId));
            return readPositions.size() - 1;
        }

        // Make room for more data (and drop data that has been read by
        // all readers)
        private void readjustBuffer() {
            int from = Collections.min(readPositions);
            int to = Collections.max(readPositions);
            int newLength = Math.max((to - from) * 2, MIN_BUF);
            int[] newBuf = new int[newLength];
            System.arraycopy(buffer, from, newBuf, 0, to - from);
            for (int i = 0; i < readPositions.size(); i++)
                readPositions.set(i, readPositions.get(i) - from);
            writePosition -= from;
            buffer = newBuf;
        }

        // Read and advance position for given reader
        public int read(int readerId) throws IOException {

            // Enough data in buffer?
            if (readPositions.get(readerId) >= writePosition) {
                readjustBuffer();
                buffer[writePosition++] = source.read();
            }

            int pos = readPositions.get(readerId);
            int b = buffer[pos];
            if (b != -1)
                readPositions.set(readerId, pos + 1);
            return b;
        }
    }

    // Non-root fields
    MultiplexedSource multiSource;
    int myId;

    // Public constructor: Used for first SplittableInputStream
    public SplittableInputStream(InputStream source) {
        multiSource = new MultiplexedSource(source);
        myId = multiSource.addSource(-1);
    }

    // Private constructor: Used in split()
    private SplittableInputStream(MultiplexedSource multiSource, int splitId) {
        this.multiSource = multiSource;
        myId = multiSource.addSource(splitId);
    }

    // Returns a new InputStream that will read bytes from this position
    // onwards.
    public SplittableInputStream split() {
        return new SplittableInputStream(multiSource, myId);
    }

    @Override
    public int read() throws IOException {
        return multiSource.read(myId);
    }
}

Finally, a demo:

String str = "Lorem ipsum\ndolor sit\namet\n";
InputStream is = new ByteArrayInputStream(str.getBytes("UTF-8"));

// Create the two buffered readers.
SplittableInputStream is1 = new SplittableInputStream(is);
SplittableInputStream is2 = is1.split();

BufferedReader br1 = new BufferedReader(new InputStreamReader(is1));
BufferedReader br2 = new BufferedReader(new InputStreamReader(is2));

// Do some interleaved reads from them.
System.out.println("One line from br1:");
System.out.println(br1.readLine());
System.out.println();

System.out.println("Two lines from br2:");
System.out.println(br2.readLine());
System.out.println(br2.readLine());
System.out.println();

System.out.println("One line from br1:");
System.out.println(br1.readLine());
System.out.println();

Output:

One line from br1:
Lorem ipsum

Two lines from br2:
Lorem ipsum
dolor sit

One line from br1:
dolor sit

Multiple readers for InputStream in Java

Tags:

Java

Io

Inputstream

Related

Recent Posts