// Path: src/main/java/com/github/davidmoten/rx2/flowable/CachedFlowable.java
// public final class CachedFlowable<T> extends Flowable<T> {
// 
//     private final OnSubscribeCacheResettable<T> cache;
// 
//     public CachedFlowable(Flowable<T> source) {
//         this(new OnSubscribeCacheResettable<T>(source));
//     }
// 
//     CachedFlowable(OnSubscribeCacheResettable<T> cache) {
//         this.cache = cache;
//     }
// 
//     public CachedFlowable<T> reset() {
//         cache.reset();
//         return this;
//     }
// 
//     @Override
//     protected void subscribeActual(Subscriber<? super T> subscriber) {
//         cache.subscribe(subscriber);
//     }
// }
// 
// Path: src/main/java/com/github/davidmoten/rx2/observable/CachedObservable.java
// public final class CachedObservable<T> extends Observable<T> {
// 
//     private final OnSubscribeCacheResetable<T> cache;
// 
//     public CachedObservable(Observable<T> source) {
//         this(new OnSubscribeCacheResetable<T>(source));
//     }
// 
//     CachedObservable(OnSubscribeCacheResetable<T> cache) {
//         this.cache = cache;
//     }
// 
//     public CachedObservable<T> reset() {
//         cache.reset();
//         return this;
//     }
// 
//     @Override
//     protected void subscribeActual(Observer<? super T> observer) {
//         cache.subscribe(observer);
//     }
// 
// }
// 
// Path: src/main/java/com/github/davidmoten/rx2/observable/CloseableObservableWithReset.java
// public final class CloseableObservableWithReset<T> {
// 
//     private final Observable<T> observable;
//     private final Runnable closeAction;
//     private final Runnable resetAction;
// 
//     public CloseableObservableWithReset(Observable<T> observable, Runnable closeAction, Runnable resetAction) {
//         this.observable = observable;
//         this.closeAction = closeAction;
//         this.resetAction = resetAction;
//     }
// 
//     public Observable<T> observable() {
//         return observable;
//     }
// 
//     public void reset() {
//         resetAction.run();
//     }
// 
//     public void close() {
//         closeAction.run();
//     }
// 
// }


import com.github.davidmoten.guavamini.Optional;
import com.github.davidmoten.rx2.flowable.CachedFlowable;
import com.github.davidmoten.rx2.observable.CachedObservable;
import com.github.davidmoten.rx2.observable.CloseableObservableWithReset;
import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.Scheduler;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
package com.github.davidmoten.rx2;



public final class Observables {


    private Observables() {
        // prevent instantiation
    }

    /**
     * Returns a cached {@link Flowable} like {@link Flowable#cache()}
     * except that the cache can be reset by calling
     * {@link CachedFlowable#reset()}.
     *
     * @param source
     *            the observable to be cached.
     * @param <T>
     *            the generic type of the source
     * @return a cached observable whose cache can be reset.
     */
    public static <T> CachedObservable<T> cache(Observable<T> source) {
        return new CachedObservable<T>(source);
    }

    /**
     * Returns a cached {@link Observable} like {@link Observable#cache()}
     * except that the cache can be reset by calling
     * {@link CachedObservable#reset()} and the cache will be automatically
     * reset an interval after first subscription (or first subscription after
     * reset). The interval is defined by {@code duration} and {@code unit} .
     *
     * @param source
     *            the source observable
     * @param duration
     *            duration till next reset
     * @param unit
     *            units corresponding to the duration
     * @param worker
     *            worker to use for scheduling reset. Don't forget to
     *            unsubscribe the worker when no longer required.
     * @param <T>
     *            the generic type of the source
     * @return cached observable that resets regularly on a time interval
     */
    public static <T> Observable<T> cache(final Observable<T> source, final long duration,
                                        final TimeUnit unit, final Scheduler.Worker worker) {
        final AtomicReference<CachedObservable<T>> cacheRef = new AtomicReference<CachedObservable<T>>();
        CachedObservable<T> cache = new CachedObservable<T>(source);
        cacheRef.set(cache);
        return cache.doOnSubscribe(new Consumer<Disposable>() {
            @Override
            public void accept(Disposable d) {
                Runnable action = new Runnable() {
                    @Override
                    public void run() {
                        cacheRef.get().reset();
                    }
                };
                worker.schedule(action, duration, unit);
            }
        });
    }

    /**
     * Returns a cached {@link Observable} like {@link Observable#cache()}
     * except that the cache may be reset by the user calling
     * {@link CloseableObservableWithReset#reset}.
     *
     * @param source
     *            the source observable
     * @param duration
     *            duration till next reset
     * @param unit
     *            units corresponding to the duration
     * @param scheduler
     *            scheduler to use for scheduling reset.
     * @param <T>
     *            generic type of source observable
     * @return {@link CloseableObservableWithReset} that should be closed once
     *         finished to prevent worker memory leak.
     */
// Path: src/main/java/com/github/davidmoten/rx2/flowable/CachedFlowable.java
// public final class CachedFlowable<T> extends Flowable<T> {
// 
//     private final OnSubscribeCacheResettable<T> cache;
// 
//     public CachedFlowable(Flowable<T> source) {
//         this(new OnSubscribeCacheResettable<T>(source));
//     }
// 
//     CachedFlowable(OnSubscribeCacheResettable<T> cache) {
//         this.cache = cache;
//     }
// 
//     public CachedFlowable<T> reset() {
//         cache.reset();
//         return this;
//     }
// 
//     @Override
//     protected void subscribeActual(Subscriber<? super T> subscriber) {
//         cache.subscribe(subscriber);
//     }
// }
// 
// Path: src/main/java/com/github/davidmoten/rx2/observable/CachedObservable.java
// public final class CachedObservable<T> extends Observable<T> {
// 
//     private final OnSubscribeCacheResetable<T> cache;
// 
//     public CachedObservable(Observable<T> source) {
//         this(new OnSubscribeCacheResetable<T>(source));
//     }
// 
//     CachedObservable(OnSubscribeCacheResetable<T> cache) {
//         this.cache = cache;
//     }
// 
//     public CachedObservable<T> reset() {
//         cache.reset();
//         return this;
//     }
// 
//     @Override
//     protected void subscribeActual(Observer<? super T> observer) {
//         cache.subscribe(observer);
//     }
// 
// }
// 
// Path: src/main/java/com/github/davidmoten/rx2/observable/CloseableObservableWithReset.java
// public final class CloseableObservableWithReset<T> {
// 
//     private final Observable<T> observable;
//     private final Runnable closeAction;
//     private final Runnable resetAction;
// 
//     public CloseableObservableWithReset(Observable<T> observable, Runnable closeAction, Runnable resetAction) {
//         this.observable = observable;
//         this.closeAction = closeAction;
//         this.resetAction = resetAction;
//     }
// 
//     public Observable<T> observable() {
//         return observable;
//     }
// 
//     public void reset() {
//         resetAction.run();
//     }
// 
//     public void close() {
//         closeAction.run();
//     }
// 
// }


// Path: src/main/java/com/github/davidmoten/rx2/Observables.java
import com.github.davidmoten.guavamini.Optional;
import com.github.davidmoten.rx2.flowable.CachedFlowable;
import com.github.davidmoten.rx2.observable.CachedObservable;
import com.github.davidmoten.rx2.observable.CloseableObservableWithReset;
import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.Scheduler;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

package com.github.davidmoten.rx2;



public final class Observables {


    private Observables() {
        // prevent instantiation
    }

    /**
     * Returns a cached {@link Flowable} like {@link Flowable#cache()}
     * except that the cache can be reset by calling
     * {@link CachedFlowable#reset()}.
     *
     * @param source
     *            the observable to be cached.
     * @param <T>
     *            the generic type of the source
     * @return a cached observable whose cache can be reset.
     */
    public static <T> CachedObservable<T> cache(Observable<T> source) {
        return new CachedObservable<T>(source);
    }

    /**
     * Returns a cached {@link Observable} like {@link Observable#cache()}
     * except that the cache can be reset by calling
     * {@link CachedObservable#reset()} and the cache will be automatically
     * reset an interval after first subscription (or first subscription after
     * reset). The interval is defined by {@code duration} and {@code unit} .
     *
     * @param source
     *            the source observable
     * @param duration
     *            duration till next reset
     * @param unit
     *            units corresponding to the duration
     * @param worker
     *            worker to use for scheduling reset. Don't forget to
     *            unsubscribe the worker when no longer required.
     * @param <T>
     *            the generic type of the source
     * @return cached observable that resets regularly on a time interval
     */
    public static <T> Observable<T> cache(final Observable<T> source, final long duration,
                                        final TimeUnit unit, final Scheduler.Worker worker) {
        final AtomicReference<CachedObservable<T>> cacheRef = new AtomicReference<CachedObservable<T>>();
        CachedObservable<T> cache = new CachedObservable<T>(source);
        cacheRef.set(cache);
        return cache.doOnSubscribe(new Consumer<Disposable>() {
            @Override
            public void accept(Disposable d) {
                Runnable action = new Runnable() {
                    @Override
                    public void run() {
                        cacheRef.get().reset();
                    }
                };
                worker.schedule(action, duration, unit);
            }
        });
    }

    /**
     * Returns a cached {@link Observable} like {@link Observable#cache()}
     * except that the cache may be reset by the user calling
     * {@link CloseableObservableWithReset#reset}.
     *
     * @param source
     *            the source observable
     * @param duration
     *            duration till next reset
     * @param unit
     *            units corresponding to the duration
     * @param scheduler
     *            scheduler to use for scheduling reset.
     * @param <T>
     *            generic type of source observable
     * @return {@link CloseableObservableWithReset} that should be closed once
     *         finished to prevent worker memory leak.
     */
    public static <T> CloseableObservableWithReset<T> cache(final Observable<T> source,