java – Terminate a Stream when there is no incoming Data after a certain Timeout

EDIT: My original answer (below) operated at the readLine() level, but on second thoughts it can be nicer to operate on the Stream level.

Here is an approach that operates on Streams. The core function is takeUntilTime(Stream<T> stream, Instant end), and the rest are small helpers. The idea is to traverse the original stream using its raw Splititerator, which makes it possible to set a timeout.

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import java.util.Optional;
import java.util.Spliterator;
import java.util.stream.Stream;

class Main {
    static long maxDurationNanos = TimeUnit.SECONDS.toNanos(5);

    static <T> Stream<T> generateOrderedStream(Supplier<T> s) {
        // Like Stream.generate(), but the returned stream is ordered.
        return Stream.iterate(s.get(), e -> s.get());
    }

    static <T> Optional<T> advance(Spliterator<T> iter) {
        // Returns an Optional with the next element of the iterator, or an empty Optional if there are no more elements.
        // (This Is much nicer than calling iter.tryAdvance() directly.)
        final var r = new Object() { T elem; };
        return iter.tryAdvance(elem -> r.elem = elem) ? Optional.of(r.elem) : Optional.empty();
    }

    static <T> Optional<T> getFuture(Future<T> future, long timeout, TimeUnit unit) {
        // Returns an Optional with the result of the Future, or an empty Optional on timeout/cancellation (on timeout, also cancels the Future).
        // Checked exceptions (ExecutionException and InterruptedException) are wrapped in an unchecked RuntimeException.
        try {
            return Optional.of(future.get(timeout, unit));
        } catch (TimeoutException e) {
            future.cancel(/* mayInterruptIfRunning: */ true);
            return Optional.empty();
        } catch (CancellationException e) {
            return Optional.empty();
        } catch (ExecutionException|InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    static <T> Optional<T> submitAndGet(ExecutorService executor, Callable<T> task, long timeout, TimeUnit unit) {
        // Convenient wrapper for getFuture(executor.submit(task), ...) that returns an empty Optional if the executor has been shut down.
        try {
            return getFuture(executor.submit(task), timeout, unit);
        } catch (RejectedExecutionException e) { // the executor has probably been shut down
            return Optional.empty();
        }
    }

    static ThreadFactory daemonThreadFactory() {
        return (r) -> {
            Thread thread = new Thread(r);
            thread.setDaemon(true);
            return thread;
        };
    }

    static <T> Stream<T> takeUntilTime(Stream<T> stream, Instant end) {
        // Traverses 'stream' until the specified end time and returns the traversed elements.
        final ExecutorService executor = Executors.newSingleThreadExecutor(daemonThreadFactory());
        final Spliterator<T> iter = stream.spliterator();
        Supplier<Optional<T>> s = () -> {
            long timeoutNanos = ChronoUnit.NANOS.between(Instant.now(), end);
            Optional<T> opt = submitAndGet(executor, () -> advance(iter), timeoutNanos, TimeUnit.NANOSECONDS).flatMap(o -> o);
            if (!opt.isPresent()) { // this will be the end of the stream, so we should clean up
                executor.shutdownNow();
            }
            return opt;
        };
        return generateOrderedStream(s)
            .takeWhile(Optional::isPresent)
            .map(Optional::get);
    }

    static void fillStreamMap(BufferedReader reader) {
        // streaming demo
        final Instant end = Instant.now().plusNanos(maxDurationNanos);
        takeUntilTime(reader.lines(), end)
            .takeWhile(line -> !line.contains("[stop]"))
            .map(line -> "[mapped] " + line)
            .forEachOrdered(System.out::println);
    }

    public static void main(String[] args) {
        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
        fillStreamMap(reader);
    }
}

original answer:

Putting aside the streaming aspect for now, the problem is basically to read with a timeout from a BufferedReader (which presumably wraps System.in). Unfortunately, it’s very hard to do this properly (see eg Set timeout for user’s input, and the article Timeout on Console Input).

One idea from those linked pages is to poll BufferedReader.ready() until it returns trueand then call readLine(). This is ugly (because it uses polling) and unreliable, because readLine() can block even if ready() returned true – for example because an incomplete line is available (on Unix-like systems the user can achieve this by typing some text then pressing Ctrl+D instead of Enter).


Another idea is to create a background thread that repeatedly calls BufferedReader.readLine() and inserts the results into a BlockingQueue (such as ArrayBlockingQueue). Then the main thread can call take() or poll(timeout, unit) on the queue to obtain lines.

A limitation of this approach is that if you later want to read from the BufferedReader directly (as opposed to through the queue), it’s pretty much impossible to avoid losing (at least) one line of input. This is because a thread can’t be interrupted cleanly when it’s blocked on readLine()so if the main thread decides to stop early (eg because of a timeout) it can’t prevent the background thread from reading the line it is currently waiting for.

You could try to “unread” the last line using mark(readAheadLimit) and reset(), but synchronization will be difficult – another thread could try to read from the BufferedReader before the background thread calls reset(). You’d probably have to synchronize using the lock field, however its access level is protected so you’d only be able to access it using reflection or by subclassing BufferedReader. Also, reset() will fail if the line to be unread is longer than readAheadLimit.

Here is an implementation that assumes you only read lines via the queue.

DISCLAIMER: Beware of bugs in these code snippets – multi-threading is tricky. I might try improve the code another time.

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.IOException;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import java.util.stream.Stream;

class InterruptibleLineReader {
    private static final String EOF = new String("<EOF>");
    BufferedReader reader;
    ArrayBlockingQueue<String> lines = new ArrayBlockingQueue<>(/* capacity: */ 2);
    Thread backgroundThread;
    IOException exception;

    public InterruptibleLineReader(BufferedReader reader) {
        this.reader = reader;
        // start a background thread to read lines
        backgroundThread = new Thread(this::backgroundTask);
        backgroundThread.setDaemon(true);
        backgroundThread.start();
    }

    public void close() {
        backgroundThread.interrupt();
        lines.clear();
        lines.add(EOF);
    }

    private void backgroundTask() {
        try {
            try {
                while (true) {
                    String line = reader.readLine();
                    if (Thread.interrupted()) {
                        // nothing to do (close() is responsible for lines.put(EOF) etc. in this case)
                        break;
                    } else if (line == null) {
                        lines.put(EOF);
                        break;
                    }
                    lines.put(line);
                }
            } catch (IOException e) {
                exception = e;
                lines.put(EOF);
            }
        } catch (InterruptedException e) {
            // nothing to do (close() is responsible for lines.put(EOF) etc. in this case)
        }
    }

    public String readLine(long timeout, TimeUnit unit) throws IOException, InterruptedException {
        String line = lines.poll(timeout, unit);
        if (line == EOF) { // EOF or IOException
            lines.put(EOF); // restore the EOF so that any concurrent (and future) calls to this method won't block
            if (exception != null) {
                throw exception;
            } else {
                return null;
            }
        }
        return line;
    }
}

class Main {
    static long maxDurationNanos = TimeUnit.SECONDS.toNanos(5);

    static <T> Stream<T> generateOrderedStream(Supplier<T> s) {
        // like Stream.generate(), but the returned stream is ordered
        return Stream.iterate(s.get(), e -> s.get());
    }

    static Stream<String> readLinesUntilTime(InterruptibleLineReader lineReader, Instant end) {
        // reads lines until the specified end time and returns them as a stream
        Supplier<String> readLine = () -> {
            long remaining = ChronoUnit.NANOS.between(Instant.now(), end);
            try {
                return lineReader.readLine(remaining, TimeUnit.NANOSECONDS);
            } catch (IOException|InterruptedException e) {
                throw new RuntimeException(e);
            }
        };
        return generateOrderedStream(readLine).takeWhile(x -> x != null);
    }

    static void fillStreamMap(InterruptibleLineReader lineReader) {
        // streaming demo
        final Instant end = Instant.now().plusNanos(maxDurationNanos);
        readLinesUntilTime(lineReader, end)
            .takeWhile(line -> !line.contains("[stop]"))
            .map(line -> "[mapped] " + line)
            .forEachOrdered(System.out::println);
    }

    public static void main(String[] args) {
        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));

        // stream lines
        InterruptibleLineReader lineReader = new InterruptibleLineReader(reader);
        System.out.println("--- streaming lines using InterruptibleLineReader for " + maxDurationNanos / 1e9 + " sec  ---");
        fillStreamMap(lineReader);
        lineReader.close();

        /*
        // attempt to use the BufferedReader directly
        // NOTE: several lines may be lost (depending on the capacity of the ArrayBlockingQueue and how quickly the lines are consumed)
        System.out.println("--- reading directly from BufferedReader ---");
        while (true) {
            try {
                String line = reader.readLine();
                if (line == null) { break; }
                System.out.println("[raw] " + line);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
        */
    }
}

Here is a more sophisticated implementation that only loses one line of input if you close the queue and read directly from the BufferedReader. It uses a custom “0-capacity” queue to ensure that at most one line will be lost.

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.IOException;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import java.util.NoSuchElementException;
import java.util.stream.Stream;

class InterruptibleLineReader {
    BufferedReader reader;
    ZeroCapacityBlockingQueue<String> lines = new ZeroCapacityBlockingQueue<>(); // a null line indicates EOF or IOException
    Thread backgroundThread;
    IOException exception;
    boolean eof;

    public InterruptibleLineReader(BufferedReader reader) {
        this.reader = reader;
        // start a background thread to read lines
        backgroundThread = new Thread(this::backgroundTask);
        backgroundThread.setDaemon(true);
        backgroundThread.start();
    }

    private void markAsEOF() {
        eof = true;
        if (lines.poll() != null) { // markAsEOF() should not be called when there are unconsumed lines
            throw new IllegalStateException();
        }
        lines.offer(null); // unblock threads that are waiting on the queue
    }

    public void close() {
        backgroundThread.interrupt();
        // warn if there is an unconsumed line, and consume it so we can indicate EOF
        String line = lines.poll();
        if (line != null) {
            System.err.println("InterruptibleLineReader: warning: discarding unconsumed line during close(): '" + line + "'");
        }
        markAsEOF();
    }

    private void backgroundTask() {
        try {
            while (true) {
                String line = reader.readLine();
                if (Thread.interrupted()) {
                    if (line != null) {
                        System.err.println("InterruptibleLineReader: warning: discarding line that was read after close(): '" + line + "'");
                    }
                    // nothing further to do (close() is responsible for calling markAsEOF() in this case)
                    break;
                } else if (line == null) { // EOF
                    markAsEOF();
                    break;
                }
                lines.put(line); // this blocks until the line has been consumed ("0-capacity" behaviour)
                if (Thread.interrupted()) {
                    // nothing to do (close() is responsible for calling markAsEOF() in this case)
                    break;
                }
            }
        } catch (IOException e) {
            exception = e;
            markAsEOF();
        } catch (InterruptedException e) {
            // nothing to do (close() is responsible for calling markAsEOF() in this case)
        }
    }

    public String readLine() throws IOException, InterruptedException {
        String line = lines.take();
        if (line == null) { // EOF or IOException
            markAsEOF(); // restore the null so that any concurrent (and future) calls to this method won't block
            if (exception != null) {
                throw exception;
            } else {
                return null; // EOF
            }
        } else {
            return line;
        }
    }

    public String readLine(long timeout, TimeUnit unit) throws IOException, InterruptedException {
        String line = lines.poll(timeout, unit);
        if (line == null && eof) { // EOF or IOException (not timeout)
            markAsEOF(); // restore the null so that any concurrent (and future) calls to this method won't block
            if (exception != null) {
                throw exception;
            } else {
                return null; // EOF
            }
        } else {
            return line;
        }
    }
}

class ZeroCapacityBlockingQueue<T> {
    int count;
    T item;

    public synchronized boolean add(T x) {
        // does not block (i.e. behaves as if the capacity is actually 1)
        if (count == 1) {
            throw new IllegalStateException("Queue full");
        }
        item = x;
        count++;
        notifyAll();
        return true;
    }

    public synchronized boolean offer(T x) {
        // does not block (i.e. behaves as if the capacity is actually 1)
        if (count == 1) {
            return false;
        }
        return add(x);
    }

    public synchronized void put(T x) throws InterruptedException {
        // blocks until the item has been removed ("0-capacity" behaviour)
        while (count == 1) {
            wait();
        }
        add(x);
        while (count == 1 && item == x) {
            wait();
        }
    }

    public synchronized T remove() {
        if (count == 0) {
            throw new NoSuchElementException();
        }
        T x = item;
        item = null;
        count--;
        notifyAll();
        return x;
    }

    public synchronized T poll() {
        if (count == 0) {
            return null;
        }
        return remove();
    }

    public synchronized T take() throws InterruptedException {
        while (count == 0) {
            wait();
        }
        return remove();
    }

    public synchronized T poll(long timeout, TimeUnit unit) throws InterruptedException {
        long deadline = System.nanoTime() + unit.toNanos(timeout);
        while (count == 0) {
            long remaining = deadline - System.nanoTime();
            if (remaining <= 0) {
                return null;
            }
            TimeUnit.NANOSECONDS.timedWait(this, remaining);
        }
        return remove();
    }
}

class Main {
    static long maxDurationNanos = TimeUnit.SECONDS.toNanos(5);

    static <T> Stream<T> generateOrderedStream(Supplier<T> s) {
        // like Stream.generate(), but the returned stream is ordered
        return Stream.iterate(s.get(), e -> s.get());
    }

    static Stream<String> readLinesUntilTime(InterruptibleLineReader lineReader, Instant end) {
        // reads lines until the specified end time and returns them as a stream
        Supplier<String> readLine = () -> {
            long remaining = ChronoUnit.NANOS.between(Instant.now(), end);
            try {
                return lineReader.readLine(remaining, TimeUnit.NANOSECONDS);
            } catch (IOException|InterruptedException e) {
                throw new RuntimeException(e);
            }
        };
        return generateOrderedStream(readLine).takeWhile(x -> x != null);
    }

    static void fillStreamMap(InterruptibleLineReader lineReader) {
        // streaming demo
        final Instant end = Instant.now().plusNanos(maxDurationNanos);
        readLinesUntilTime(lineReader, end)
            .takeWhile(line -> !line.contains("[stop]"))
            .map(line -> "[mapped] " + line)
            .forEachOrdered(System.out::println);
    }

    public static void main(String[] args) {
        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));

        // stream lines
        InterruptibleLineReader lineReader = new InterruptibleLineReader(reader);
        System.out.println("--- streaming lines using InterruptibleLineReader for " + maxDurationNanos / 1e9 + " sec  ---");
        fillStreamMap(lineReader);
        lineReader.close();

        /*
        // attempt to use the BufferedReader directly
        // NOTE: a line will be lost
        System.out.println("--- reading directly from BufferedReader ---");
        while (true) {
            try {
                String line = reader.readLine();
                if (line == null) { break; }
                System.out.println("[raw] " + line);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
        */
    }
}

Here is an example run of the second implementation (with the last part of main() uncommented). The timestamps are in seconds and “>” denotes input.

0.06 --- streaming lines using InterruptibleLineReader for 5.0 sec  ---
0.82 > one
0.83 [mapped] one
1.76 > two
1.76 [mapped] two
2.73 > three
2.73 [mapped] three
5.06 --- reading directly from BufferedReader ---
6.93 > four
6.94 InterruptibleLineReader: warning: discarding line that was read after close(): 'four'
7.76 > five
7.76 [raw] five
8.60 > six
8.60 [raw] six

Note how the line “four” was lost. To avoid losing lines, don’t use the underlying BufferedReader after the InterruptibleLineReader instance is created.

(If you really need a BufferedReader after that point, you could write a dummy subclass of BufferedReader that wraps InterruptibleLineReader and forwards readLine() calls to it. The other BufferedReader methods, such as read() and mark()can’t be implemented easily.)

Leave a Comment