@@ -0,0 +1,816 @@
+package com.github.davidmoten.rx.operators;%0A%0Aimport java.util.concurrent.atomic.AtomicBoolean;%0A%0Aimport rx.Observable;%0Aimport rx.Observable.OnSubscribe;%0Aimport rx.Subscriber;%0A%0Apublic final class OnSubscribeCacheResetable%3CT%3E implements OnSubscribe%3CT%3E %7B%0A%0A    private final AtomicBoolean refresh = new AtomicBoolean(true);%0A    private final Observable%3CT%3E source;%0A    private volatile Observable%3CT%3E current;%0A%0A    public OnSubscribeCacheResetable(Observable%3CT%3E source) %7B%0A        this.source = source;%0A        this.current = source;%0A    %7D%0A%0A    @Override%0A    public void call(Subscriber%3C? super T%3E subscriber) %7B%0A        if (refresh.compareAndSet(true, false)) %7B%0A            current = source.cache();%0A        %7D%0A        current.unsafeSubscribe(subscriber);%0A    %7D%0A%0A    public void reset() %7B%0A        refresh.set(true);%0A    %7D%0A%0A%7D%0A

add cache resetable