# Java 9 Flow API: timing out events

DevFeed: [Java 9 Flow API: timing out events](<https://devfeed.tech/articles/java-9-flow-api-timing-out-events-24813.md>)

Original publisher: [Read original article](<https://akarnokd.blogspot.com/2017/09/java-9-flow-api-timing-out-events.html>)

Author: David Karnok (noreply@blogger.com)

Published: 2017-09-25T11:46:00Z

Content type: tutorial

Language: en

Sources: [Akarnokd - Advanced RxJava](<https://devfeed.tech/sources/akarnokd-advanced-rxjava.md>)

Topics: [reactive](<https://devfeed.tech/topics/reactive.md>), [Java 9](<https://devfeed.tech/topics/java-9.md>), [Concurrency](<https://devfeed.tech/topics/concurrency.md>), [Programming](<https://devfeed.tech/topics/programming.md>), [Streams](<https://devfeed.tech/topics/streams.md>)

Tags: [await](<https://devfeed.tech/tags/await.md>), [concurrency](<https://devfeed.tech/tags/concurrency.md>), [flow](<https://devfeed.tech/tags/flow.md>), [flow-api](<https://devfeed.tech/tags/flow-api.md>), [java-9](<https://devfeed.tech/tags/java-9.md>), [jdk](<https://devfeed.tech/tags/jdk.md>), [programming](<https://devfeed.tech/tags/programming.md>), [reactive](<https://devfeed.tech/tags/reactive.md>), [reactive-programming](<https://devfeed.tech/tags/reactive-programming.md>), [streams](<https://devfeed.tech/tags/streams.md>), [subscriber](<https://devfeed.tech/tags/subscriber.md>), [subscription](<https://devfeed.tech/tags/subscription.md>), [timeout](<https://devfeed.tech/tags/timeout.md>)

## AI overview

This tutorial explains how to design a timeout operator for Java 9 Flow API publishers. It models each event as a race between item arrival and a timer, signals a TimeoutException when the timer wins, and addresses serialized downstream signals when timeout and upstream events occur concurrently.

## Source excerpt

Introduction One of the main properties of reactive programming is that the events may arrive over time instead of immediately available to a consumer. In traditional Future-based programming, one could wait for the result in a blocking manner via Future.get(long, TimeUnit). Other data sources, such as network InputStream have either their own built-in timeout facility or one has to use external means to close the stream after certain period of time to unblock the reader to it. Java 8 Streams have also no direct timeout support. In the reactive mindset, one can consider timing out events (items) as requesting an element and racing its arrival against the clock. If the item arrives in time, we should ignore the clock. If the clock fires first, we should stop the sender of the items and somehow notify the consumer of the situation. Perhaps the simplest way is to signal onError with a TimeoutException. Since there could be multiple items from a flow, we have to do this racing for each potential items over and over until the flow terminates. The timeout operator Since there is "time" in timeout, we'll need a source of time that can be started and stopped at will. The first tool that comes into mind is the java.util.Timer class, however, even its Javadoc suggest one uses a ScheduledExecutorService instead. If one has to deal with a lot of timed operations, besides of timing out flows, having the control over such signals via a (set of) ScheduledExecutorServices is desirable. Therefore, let's define our timeout API with it: public static <T> Flow.Publisher<T> timeout( Flow.Publisher<T> source, long timeout, TimeUnit unit, ScheduledExecutorService timer) { return new TimeoutPublisher<>(source, timeout, unit, timer); } (Note that if one uses the Executors.newScheduledExecutorService(), it has to be shutdown at some point, otherwise it's non-daemon thread by default would prevent the JVM from quitting.) One primary responsibility of this type of operator is to make sure the