Cache with expiration in stdlib ?

[bobymicroby]

Hello guys,

I was in need of observable( actually a single) that is able to expire cached upstream observable after a period of time.

I was unable to bend the operators from the stdlib to my desire, so I came up with this piece of code :

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);
    }
}

Can someone have a look and tell me if I am doing it right ?

[akarnokd]

There is a Observable.replay() overload that evicts old entries (although doesn't work properly now, see #4023).

[bobymicroby]

@akarnokd My upstream Observable actually acts like a Single , is replay() going to work in my use case where eviction, means renewing the subscription ? I thought replay() only caches/evicts the values from a single stream .

[akarnokd]

No, replay requires you to call connect() manually.

Your implementation works but only if there are no concurrent subscribers to it.

[bobymicroby]

@akarnokd

I think that this is pretty common use case, at least on android devices where one must cache the network response ( Usually abstracted as Single) for some short period of time in order to survive orientation changes.

Usually the subscriptions that will subscribe to this Single TTL cache will be done from one thread,
but it will be best if one can avoid concurrent problems at all.

Can you point me to concurrent problem in the code i have shared?

[akarnokd]

See @dlew 's post about a similar (same?): http://blog.danlew.net/2015/06/22/loading-data-from-multiple-sources-with-rxjava/

if (currentTime > nextExpirationTime)

With the right timing, you'd end up subscribing to two different cached sources. Given two threads, if both pass this if and one gets interrupted, the second goes ahead and subscribes to a new cached value. When the first gets resumed, it overwrites the first cached Observable and triggers a new subscription.

[bobymicroby]

Thanks a lot for the point to the blog post and for the concurrency enlightenment.

[bobymicroby]

@akarnokd I have reviewed @dlew's blogpost, thanks again from pointing that out. Still I think i prefer the internal operators for caching. Do you think that I have fixed the concurrency mistake from the previous comment in the code bellow ?

Regards, B.

package com.zygoid.aura.android.app.util.rx.cache;

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

import java.util.concurrent.atomic.AtomicReference;

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)
        {

            synchronized (this)
            {
                if (currentTime > nextExpirationTime)
                {
                    nextExpirationTime = currentTime + ttl;
                    current = cacheIfNotError(source);
                }
            }

        }

        current.unsafeSubscribe(subscriber);
    }

    public Observable<T> cacheIfNotError(Observable<T> source)
    {
        final AtomicReference<Observable<T>> cached = new AtomicReference<>();
        return Observable.defer(() -> {
            for (; ; )
            {
                Observable<T> conn = cached.get();
                if (conn != null)
                {
                    return conn;
                }
                Observable<T> next = source.doOnError(e -> cached.set(null)).replay().autoConnect();

                if (cached.compareAndSet(null, next))
                {
                    return next;
                }
            }
        });
    }

}

P.S I have updated the code to only cache if there was no error as you suggest on SO