OnSubscribeTTLCache.java

import rx.Observable;
import rx.Subscriber;
import rx.functions.Func0;

public class OnSubscribeTTLCache<T> implements Observable.OnSubscribe<T>
{

    private final Observable<T> source;
    private final long ttl;
    private final Func0<Long> clock;

    private volatile Observable<T> current;
    private volatile long nextExpirationTime;

    public static <T> Observable<T> cache(Observable<T> source, long ttl, Func0<Long> clock)
    {
        return Observable.create(new OnSubscribeTTLCache<>(source, ttl, clock));
    }

    public OnSubscribeTTLCache(Observable<T> source, long ttl, Func0<Long> clock)
    {
        this.ttl = ttl;
        this.source = source;
        this.clock = clock;
        this.current = source;
        this.nextExpirationTime = Long.MIN_VALUE;
    }

    @Override
    public void call(Subscriber<? super T> subscriber)
    {
        final long currentTime = clock.call();

        if (currentTime > nextExpirationTime)
        {
            nextExpirationTime = currentTime + ttl;
            current = source.cacheWithInitialCapacity(1);
        }
        current.unsafeSubscribe(subscriber);
    }
}