How to simplify a rxjava stream with timeouts
<p>I have the following snippet:</p>

<pre class="lang-java prettyprint-override"><code>Observable.just(1)
    .flatMap(-&gt; doSomething1)
    .timeout(10, SECONDS)
    .flatMap(-&gt; doSomething2)
    .timeout(10, SECONDS)
    .flatMap(-&gt; doSomething3)
    .timeout(10, SECONDS)
    .flatMap(-&gt; doSomething4)
    .timeout(10, SECONDS)
    .subscribe();
</code></pre>

<p>I don't want to repeat myself after every <code>flatMap</code> adding a <code>timeout</code>. My first thought was to only apply <code>timeout</code> to the beginning or end of the stream, but that is not the behaviour I intent as it only applies timeout to the closer observable.</p>

<pre class="lang-java prettyprint-override"><code>Observable.just(1)
    .flatMap(-&gt; doSomething1)
    .flatMap(-&gt; doSomething2)
    .flatMap(-&gt; doSomething3)
    .flatMap(-&gt; doSomething4)
    .timeout(10, SECONDS)
    .subscribe();

Observable.just(1)
    .timeout(10, SECONDS)
    .flatMap(-&gt; doSomething1)
    .flatMap(-&gt; doSomething2)
    .flatMap(-&gt; doSomething3)
    .flatMap(-&gt; doSomething4)
    .subscribe();
</code></pre>

<p>The <code>doSomethingX</code> functions execute a little code on call that can take a while before returning the next observable which itself does not need to be wrapped into a timeout.</p>

<p>How can this be improved?</p>

<p>UPDATE:</p>

<p>O more practical example below. The ideia is to compose a stream that I can retry in case of failure or timeout. I'm simulating the scenario where one of the operators times out once, but works on a retry.</p>

<pre class="lang-java prettyprint-override"><code>@Test
public void streamToBeSimplified() throws Exception {
    final AtomicBoolean retry = new AtomicBoolean(true);

    Action1&lt;Object&gt; print = new Action1&lt;Object&gt;() {
        @Override
        public void call(Object o) {
            System.out.println(" &gt;&gt;&gt;" + o);
        }
    };

    Observable.just(1)
            .doOnNext(print)
            .flatMap(new Func1&lt;Integer, Observable&lt;Integer&gt;&gt;() {
                @Override
                public Observable&lt;Integer&gt; call(Integer integer) {
                    return Observable.just(2);
                }
            })
            .timeout(1, TimeUnit.SECONDS)
            .doOnNext(print)
            .flatMap(new Func1&lt;Object, Observable&lt;Integer&gt;&gt;() {
                @Override
                public Observable&lt;Integer&gt; call(Object o) {

                    if(retry.getAndSet(false)) {
                        try {
                            Thread.sleep(2000L);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    return Observable.just(3);
                }
            })
            .timeout(1, TimeUnit.SECONDS)
            .doOnNext(print)
            .retry(2)
            .subscribe();

}
</code></pre>