public final class BiFunctions {

    private BiFunctions() {
        // prevent instantiation
    }

    @SuppressWarnings("unchecked")
    public static <A, B, C> BiFunction<A, B, C> throwing() {
        return (BiFunction<A, B, C>) ThrowingHolder.INSTANCE;
    }

    private static final class ThrowingHolder {
        static BiFunction<Object, Object, Object> INSTANCE = new BiFunction<Object, Object, Object>() {

            @Override
            public Object apply(Object t1, Object t2) throws Exception {
                throw new ThrowingException();
            }
        };
    }

    public static <T extends Number> BiFunction<Statistics, T, Statistics> collectStats() {
        return new BiFunction<Statistics, T, Statistics>() {

            @Override
            public Statistics apply(Statistics s, T t) {
                return s.add(t);
            }
        };
    }

    public static <T, R, S> BiFunction<T, R, S> constant(final S value) {
        // TODO make holder
        return new BiFunction<T, R, S>() {

            @Override
            public S apply(T t1, R t2) throws Exception {
                return value;
            }
        };
    }

    public static <T, R, S> BiFunction<T, R, S> toNull() {
        // TODO make holder
        return new BiFunction<T, R, S>() {

            @Override
            public S apply(T t1, R t2) throws Exception {
                return null;
            }
        };

    }

} public final class Flowables {

    private static final int DEFAULT_MATCH_BATCH_SIZE = 128;

    private Flowables() {
        // prevent instantiation
    }

    public static <A, B, K, C> Flowable<C> match(Flowable<A> a, Flowable<B> b,
            Function<? super A, K> aKey, Function<? super B, K> bKey,
            BiFunction<? super A, ? super B, C> combiner, int requestSize) {
        return new FlowableMatch<A, B, K, C>(a, b, aKey, bKey, combiner, requestSize);
    }

    public static <A, B, K, C> Flowable<C> match(Flowable<A> a, Flowable<B> b,
            Function<? super A, K> aKey, Function<? super B, K> bKey,
            BiFunction<? super A, ? super B, C> combiner) {
        return match(a, b, aKey, bKey, combiner, DEFAULT_MATCH_BATCH_SIZE);
    }

    public static <T> Flowable<T> repeat(T t) {
        return new FlowableRepeat<T>(t, -1);
    }

    public static <T> Flowable<T> repeat(T t, long count) {
        return new FlowableRepeat<T>(t, count);
    }

    /**
     * <p>
     * Creates a Flowable that is aimed at supporting calls to a service that
     * provides data in pages where the page sizes are determined by requests from
     * downstream (requests are a part of the backpressure machinery of RxJava).
     * 
     * <p>
     * <img src=
     * "https://raw.githubusercontent.com/davidmoten/rxjava2-extras/master/src/docs/fetchPagesByRequest.png"
     * alt="image">
     * 
     * <p>
     * Here's an example.
     * 
     * <p>
     * Suppose you have a stateless web service, say a rest service that returns
     * JSON/XML and supplies you with
     * 
     * <ul>
     * <li>the most popular movies of the last 24 hours sorted by descending
     * popularity</li>
     * </ul>
     * 
     * <p>
     * The service supports paging in that you can pass it a start number and a page
     * size and it will return just that slice from the list.
     * 
     * <p>
     * Now I want to give a library with a Flowable definition of this service to my
     * colleagues that they can call in their applications whatever they may be. For
     * example,
     * 
     * <ul>
     * <li>Fred may just want to know the most popular movie each day,</li>
     * <li>Greta wants to get the top 20 and then have the ability to keep scrolling
     * down the list in her UI.</li>
     * </ul>
     * 
     * <p>
     * Let's see how we can efficiently support those use cases. I'm going to assume
     * that the movie data returned by the service are mapped conveniently to
     * objects by whatever framework I'm using (JAXB, Jersey, etc.). The fetch
     * method looks like this:
     * 
     * <pre>
     * {@code
     * // note that start is 0-based
     * List<Movie> mostPopularMovies(int start, int size);
     * }
     * </pre>
     * 
     * <p>
     * Now I'm going to wrap this synchronous call as a Flowable to give to my
     * colleagues:
     * 
     * <pre>
     * {@code
     * Flowable<Movie> mostPopularMovies(int start) {
     *     return Flowables.fetchPagesByRequest(
     *           (position, n) -> Flowable.fromIterable(mostPopular(position, n)),
     *           start)
     *         // rebatch requests so that they are always between 
     *         // 5 and 100 except for the first request
     *       .compose(Transformers.rebatchRequests(5, 100, false));
     * }
     * 
     * Flowable<Movie> mostPopularMovies() {
     *     return mostPopularMovies(0);
     * }
     * }
     * </pre>
     * <p>
     * Note particularly that the method above uses a variant of rebatchRequests to
     * limit both minimum and maximum requests. We particularly don't want to allow
     * a single call requesting the top 100,000 popular movies because of the memory
     * and network pressures that arise from that call.
     * 
     * <p>
     * Righto, Fred now uses the new API like this:
     * 
     * <pre>
     * {
     *     &#64;code
     *     Movie top = mostPopularMovies().compose(Transformers.maxRequest(1)).first()
     *             .blockingFirst();
     * }
     * </pre>
     * <p>
     * The use of maxRequest above may seem unnecessary but strangely enough the
     * first operator requests Long.MAX_VALUE of upstream and cancels as soon as one
     * arrives. The take, elemnentAt and firstXXX operators all have this
     * counter-intuitive characteristic.
     * 
     * <p>
     * Greta uses the new API like this:
     * 
     * <pre>
     * {@code
     * mostPopularMovies()
     *     .rebatchRequests(20)
     *     .doOnNext(movie -> addToUI(movie))
     *     .subscribe(subscriber);
     * }
     * </pre>
     * <p>
     * A bit more detail about fetchPagesByRequest:
     * 
     * <p>
     * If the fetch function returns a Flowable that delivers fewer than the
     * requested number of items then the overall stream completes.
     * 
     * @param fetch
     *            a function that takes a position index and a length and returns a
     *            Flowable
     * @param start
     *            the start index
     * @param maxConcurrent
     *            how many pages to request concurrently
     * @param <T>
     *            item type
     * @return Flowable that fetches pages based on request amounts
     */
    public static <T> Flowable<T> fetchPagesByRequest(
            final BiFunction<? super Long, ? super Long, ? extends Flowable<T>> fetch, long start,
            int maxConcurrent) {
        return FlowableFetchPagesByRequest.create(fetch, start, maxConcurrent);
    }

    public static <T> Flowable<T> fetchPagesByRequest(
            final BiFunction<? super Long, ? super Long, ? extends Flowable<T>> fetch, long start) {
        return fetchPagesByRequest(fetch, start, 2);
    }

    public static <T> Flowable<T> fetchPagesByRequest(
            final BiFunction<? super Long, ? super Long, ? extends Flowable<T>> fetch) {
        return fetchPagesByRequest(fetch, 0, 2);
    }

    /**
     * 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> CachedFlowable<T> cache(Flowable<T> source) {
        return new CachedFlowable<T>(source);
    }

    /**
     * Returns a cached {@link Flowable} like {@link Flowable#cache()} except that
     * the cache can be reset by calling {@link CachedFlowable#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> Flowable<T> cache(final Flowable<T> source, final long duration,
            final TimeUnit unit, final Scheduler.Worker worker) {
        final AtomicReference<CachedFlowable<T>> cacheRef = new AtomicReference<CachedFlowable<T>>();
        CachedFlowable<T> cache = new CachedFlowable<T>(source);
        cacheRef.set(cache);
        return cache.doOnSubscribe(new Consumer<Subscription>() {
            @Override
            public void accept(Subscription d) {
                Runnable action = new Runnable() {
                    @Override
                    public void run() {
                        cacheRef.get().reset();
                    }
                };
                worker.schedule(action, duration, unit);
            }
        });
    }

    /**
     * Returns a cached {@link Flowable} like {@link Flowable#cache()} except that
     * the cache may be reset by the user calling
     * {@link CloseableFlowableWithReset#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 CloseableFlowableWithReset} that should be closed once
     *         finished to prevent worker memory leak.
     */
    public static <T> CloseableFlowableWithReset<T> cache(final Flowable<T> source,
            final long duration, final TimeUnit unit, final Scheduler scheduler) {
        final AtomicReference<CachedFlowable<T>> cacheRef = new AtomicReference<CachedFlowable<T>>();
        final AtomicReference<Optional<Scheduler.Worker>> workerRef = new AtomicReference<Optional<Scheduler.Worker>>(
                Optional.<Scheduler.Worker>absent());
        CachedFlowable<T> cache = new CachedFlowable<T>(source);
        cacheRef.set(cache);
        Runnable closeAction = new Runnable() {
            @Override
            public void run() {
                while (true) {
                    Optional<Scheduler.Worker> w = workerRef.get();
                    if (w == null) {
                        // we are finished
                        break;
                    } else {
                        if (workerRef.compareAndSet(w, null)) {
                            if (w.isPresent()) {
                                w.get().dispose();
                            }
                            // we are finished
                            workerRef.set(null);
                            break;
                        }
                    }
                    // if not finished then try again
                }
            }
        };
        Runnable resetAction = new Runnable() {

            @Override
            public void run() {
                startScheduledResetAgain(duration, unit, scheduler, cacheRef, workerRef);
            }
        };
        return new CloseableFlowableWithReset<T>(cache, closeAction, resetAction);
    }

    private static <T> void startScheduledResetAgain(final long duration, final TimeUnit unit,
            final Scheduler scheduler, final AtomicReference<CachedFlowable<T>> cacheRef,
            final AtomicReference<Optional<Scheduler.Worker>> workerRef) {

        Runnable action = new Runnable() {
            @Override
            public void run() {
                cacheRef.get().reset();
            }
        };
        // CAS loop to cancel the current worker and create a new one
        while (true) {
            Optional<Scheduler.Worker> wOld = workerRef.get();
            if (wOld == null) {
                // we are finished
                return;
            }
            Optional<Scheduler.Worker> w = Optional.of(scheduler.createWorker());
            if (workerRef.compareAndSet(wOld, w)) {
                if (wOld.isPresent())
                    wOld.get().dispose();
                w.get().schedule(action, duration, unit);
                break;
            }
        }
    }

    private static final int DEFAULT_RING_BUFFER_SIZE = 128;

    public static <T> Flowable<T> mergeInterleaved(Publisher<? extends Publisher<? extends T>> publishers,
            int maxConcurrency, int batchSize, boolean delayErrors) {
        return new FlowableMergeInterleave<T>(publishers, maxConcurrency, batchSize, delayErrors);
    }

    public static <T> Flowable<T> mergeInterleaved(Publisher<? extends Publisher<? extends T>> publishers,
            int maxConcurrency) {
        return mergeInterleaved(publishers, maxConcurrency, 128, false);
    }

    public static <T> MergeInterleaveBuilder<T> mergeInterleaved(Publisher<? extends Publisher<? extends T>> publishers) {
        return new MergeInterleaveBuilder<T>(publishers);
    }

    public static final class MergeInterleaveBuilder<T> {

        private final Publisher<? extends Publisher<? extends T>> publishers;
        private int maxConcurrency = 4;
        private int batchSize = DEFAULT_RING_BUFFER_SIZE;
        private boolean delayErrors = false;

        MergeInterleaveBuilder(Publisher<? extends Publisher<? extends T>> publishers) {
            this.publishers = publishers;
        }

        public MergeInterleaveBuilder<T> maxConcurrency(int maxConcurrency) {
            this.maxConcurrency = maxConcurrency;
            return this;
        }

        public MergeInterleaveBuilder<T> batchSize(int batchSize) {
            this.batchSize = batchSize;
            return this;
        }

        public MergeInterleaveBuilder<T> delayErrors(boolean delayErrors) {
            this.delayErrors = delayErrors;
            return this;
        }

        public Flowable<T> build() {
            return mergeInterleaved(publishers, maxConcurrency, batchSize, delayErrors);
        }
    }
} public final class Functions {

    private Functions() {
        // prevent instantiation
    }

    public static <T> Function<Object, T> constant(final T value) {
        return new Function<Object, T>() {

            @Override
            public T apply(Object t) throws Exception {
                return value;
            }
        };
    }

    @SuppressWarnings("unchecked")
    public static <T> Function<T, T> identity() {
        return (Function<T, T>) IdentityHolder.INSTANCE;
    }

    private static final class IdentityHolder {
        static final Function<Object, Object> INSTANCE = new Function<Object, Object>() {

            @Override
            public Object apply(Object t) throws Exception {
                return t;
            }
        };
    }

    public static <T, R> Function<T, R> throwing() {
        // TODO make holder
        return new Function<T, R>() {

            @Override
            public R apply(T t) {
                throw new ThrowingException();
            }
        };
    }
    
    public static <T> Function<T, String> toStringFunction() {
        // TODO make holder
        return new Function<T,String> () {

            @Override
            public String apply(T t) throws Exception {
                return String.valueOf(t);
            }
        };
    }

} public final class StateMachine {

    private StateMachine() {
        // prevent instantiation
    }

    public interface Transition<State, In, Out> extends Function3<State, In, FlowableEmitter<Out>, State> {

        // override so IDEs have better suggestions for parameters
        @Override
        State apply(State state, In value, FlowableEmitter<Out> FlowableEmitter);

    }

    public interface Transition2<State, In, Out> extends Function3<State, In, Emitter<Out>, State> {

        // override so IDEs have better suggestions for parameters
        @Override
        State apply(State state, In value, Emitter<Out> emitter);

    }
    
    public interface Completion<State, Out> extends BiPredicate<State, FlowableEmitter<Out>> {

        // override so IDEs have better suggestions for parameters
        @Override
        boolean test(State state, FlowableEmitter<Out> emitter);

    }

    public interface Completion2<State, Out> extends BiConsumer<State, Emitter<Out>> {

        // override so IDEs have better suggestions for parameters
        @Override
        void accept(State state, Emitter<Out> emitter);

    }

    
    public interface Errored<State, Out> extends Consumer3<State, Throwable, Emitter<Out>> {

        // override so IDEs have better suggestions for parameters
        @Override
        void accept(State state, Throwable error, Emitter<Out> emitter);

    }

    public static Builder builder() {
        return new Builder();
    }

    public static final class Builder {

        private Builder() {
            // prevent instantiation from other packages
        }

        public <State> Builder2<State> initialStateFactory(Callable<State> initialState) {
            return new Builder2<State>(initialState);
        }

        public <State> Builder2<State> initialState(final State initialState) {
            return initialStateFactory(Callables.constant(initialState));
        }

    }

    public static final class Builder2<State> {

        private final Callable<State> initialState;

        private Builder2(Callable<State> initialState) {
            this.initialState = initialState;
        }

        public <In, Out> Builder3<State, In, Out> transition(Transition<State, In, Out> transition) {
            return new Builder3<State, In, Out>(initialState, transition);
        }

    }

    public static final class Builder3<State, In, Out> {

        private static final int DEFAULT_REQUEST_SIZE = 1;

        private final Callable<State> initialState;
        private final Transition<State, In, Out> transition;
        private Completion<State, Out> completion = CompletionAlwaysTrueHolder.instance();
        private BackpressureStrategy backpressureStrategy = BackpressureStrategy.BUFFER;
        private int requestBatchSize = DEFAULT_REQUEST_SIZE;

        private Builder3(Callable<State> initialState, Transition<State, In, Out> transition) {
            this.initialState = initialState;
            this.transition = transition;
        }

        public Builder3<State, In, Out> completion(Completion<State, Out> completion) {
            this.completion = completion;
            return this;
        }

        public Builder3<State, In, Out> backpressureStrategy(BackpressureStrategy backpressureStrategy) {
            this.backpressureStrategy = backpressureStrategy;
            return this;
        }

        public Builder3<State, In, Out> requestBatchSize(int value) {
            this.requestBatchSize = value;
            return this;
        }

        public FlowableTransformer<In, Out> build() {
            return Transformers.stateMachine(initialState, transition, completion, backpressureStrategy,
                    requestBatchSize);
        }

    }

    @VisibleForTesting
    static final class CompletionAlwaysTrueHolder {

        private CompletionAlwaysTrueHolder() {
            // prevent instantiation
        }

        private static final Completion<Object, Object> INSTANCE = new Completion<Object, Object>() {
            @Override
            public boolean test(Object t1, FlowableEmitter<Object> t2) {
                return true;
            }
        };

        @SuppressWarnings("unchecked")
        static <State, Out> Completion<State, Out> instance() {
            return (Completion<State, Out>) INSTANCE;
        }
    }

    public static interface Emitter<T> {
        void onNext_(T t);

        void onError_(Throwable e);

        void onComplete_();

        void cancel_();
    }

} public class StateMachine2 {

    public static Builder builder() {
        return new Builder();
    }

    public static final class Builder {

        private Builder() {
            // prevent instantiation from other packages
        }

        public <State> Builder2<State> initialStateFactory(Callable<State> initialState) {
            return new Builder2<State>(initialState);
        }

        public <State> Builder2<State> initialState(final State initialState) {
            return initialStateFactory(Callables.constant(initialState));
        }

    }

    public static final class Builder2<State> {

        private final Callable<State> initialState;

        private Builder2(Callable<State> initialState) {
            this.initialState = initialState;
        }

        public <In, Out> Builder3<State, In, Out> transition(Transition2<State, In, Out> transition) {
            return new Builder3<State, In, Out>(initialState, transition);
        }

    }

    public static final class Builder3<State, In, Out> {

        private static final int DEFAULT_REQUEST_SIZE = 1;

        private final Callable<State> initialState;
        private final Transition2<State, In, Out> transition;
        private Completion2<State, Out> completion = null;
        private Errored<State, Out> errored = null;
        private BackpressureStrategy backpressureStrategy = BackpressureStrategy.BUFFER;
        private int requestBatchSize = DEFAULT_REQUEST_SIZE;

        private Builder3(Callable<State> initialState, Transition2<State, In, Out> transition) {
            this.initialState = initialState;
            this.transition = transition;
        }

        public Builder3<State, In, Out> completion(Completion2<State, Out> completion) {
            this.completion = completion;
            return this;
        }

        public Builder3<State, In, Out> errored(Errored<State, Out> errored) {
            this.errored = errored;
            return this;
        }

        public Builder3<State, In, Out> backpressureStrategy(BackpressureStrategy backpressureStrategy) {
            this.backpressureStrategy = backpressureStrategy;
            return this;
        }

        public Builder3<State, In, Out> requestBatchSize(int value) {
            this.requestBatchSize = value;
            return this;
        }

        public FlowableTransformer<In, Out> build() {
            return new FlowableTransformer<In, Out>() {

                @Override
                public Publisher<Out> apply(Flowable<In> source) {
                    return new FlowableStateMachine<State, In, Out>(source, initialState, transition, completion,
                            errored, backpressureStrategy, requestBatchSize);
                }
            };
        }

    }

} public final class Statistics {

    private final long count;
    private final double sumX;
    private final double sumX2;

    private Statistics(long count, double sumX, double sumX2) {
        this.count = count;
        this.sumX = sumX;
        this.sumX2 = sumX2;
    }

    public static Statistics create() {
        return new Statistics(0, 0, 0);
    }

    public Statistics add(Number number) {
        double x = number.doubleValue();
        return new Statistics(count + 1, sumX + x, sumX2 + x * x);
    }

    public long count() {
        return count;
    }

    public double sum() {
        return sumX;
    }

    public double sumSquares() {
        return sumX2;
    }

    public double mean() {
        return sumX / count;
    }

    public double sd() {
        double m = mean();
        return Math.sqrt(sumX2 / count - m * m);
    }

    @Override
    public String toString() {
        StringBuilder builder = new StringBuilder();
        builder.append("Statistics [count=");
        builder.append(count);
        builder.append(", sum=");
        builder.append(sum());
        builder.append(", sumSquares=");
        builder.append(sumSquares());
        builder.append(", mean=");
        builder.append(mean());
        builder.append(", sd=");
        builder.append(sd());
        builder.append("]");
        return builder.toString();
    }
} public final class Options {

    public static final String DEFAULT_FILE_PREFIX = "bufferToFile_";

    private final Callable<File> fileFactory;
    private final int pageSizeBytes;
    private final Scheduler scheduler;

    @VisibleForTesting
    Options(Callable<File> filefactory, int pageSizeBytes, Scheduler scheduler) {
        Preconditions.checkNotNull(filefactory);
        Preconditions.checkArgument(pageSizeBytes > 0, "bufferSizeBytes must be greater than 0");
        Preconditions.checkNotNull(scheduler);
        this.fileFactory = filefactory;
        this.pageSizeBytes = pageSizeBytes;
        this.scheduler = scheduler;
    }

    public Callable<File> fileFactory() {
        return fileFactory;
    }

    public int pageSizeBytes() {
        return pageSizeBytes;
    }

    public Scheduler scheduler() {
        return scheduler;
    }

    public static BuilderFlowable builderFlowable() {
        return new BuilderFlowable();
    }

    public static BuilderObservable builderObservable() {
        return new BuilderObservable();
    }

    public static final class BuilderFlowable {

        private Callable<File> fileFactory = FileFactoryHolder.INSTANCE;
        private int pageSizeBytes = 1024 * 1024;
        private Scheduler scheduler = Schedulers.io();

        BuilderFlowable() {
        }

        /**
         * Sets the page size in bytes. A page corresponds to a single memory
         * mapped file. If this method is not called the default value is 1MB
         * (1024*1024 bytes).
         * 
         * @param pageSizeBytes
         *            the page size in bytes.
         * @return this
         */
        public BuilderFlowable pageSizeBytes(int pageSizeBytes) {
            this.pageSizeBytes = pageSizeBytes;
            return this;
        }

        /**
         * Sets the scheduler to use for reading items from files and emitting
         * them.
         * 
         * @param scheduler
         *            for emitting items
         * @return this
         */
        public BuilderFlowable scheduler(Scheduler scheduler) {
            this.scheduler = scheduler;
            return this;
        }

        /**
         * Sets the file factory to be used by the queue storage mechanism.
         * Defaults to using {@code File.createTempFile("bufferToFileDb","")} if
         * this method is not called.
         * 
         * @param fileFactory
         *            the factory
         * @return the current builder
         */
        public BuilderFlowable fileFactory(Callable<File> fileFactory) {
            this.fileFactory = fileFactory;
            return this;
        }

        /**
         * Returns a Flowable Transformer to buffer stream items to disk
         * (files). The serialization method uses
         * {@code java.io.ObjectOutputStream} for writing values to disk and
         * {@code java.io.ObjectInputStream} for reading values from disk and
         * emitting them. Note that this is a pretty verbose serialization
         * method for small items in that a 4 byte Integer is serialized to an
         * 81 byte array!
         * 
         * @param <T>
         *            stream item type
         * @return FlowableTransformer using the java io serializer
         */
        public <T extends Serializable> FlowableTransformer<T, T> serializerJavaIO() {
            return serializer(Serializers.<T>javaIO());
        }

        /**
         * Returns a Flowable Transformer that buffers items to disk (files).
         * The serialization method passes through byte arrays directly for both
         * writing and reading.
         * 
         * @return FlowableTransformer using the bytes serializer
         */
        public FlowableTransformer<byte[], byte[]> serializerBytes() {
            return serializer(Serializers.bytes());
        }

        public FlowableTransformer<String, String> serializerUtf8() {
            return serializer(Serializers.utf8());
        }

        public <T> FlowableTransformer<T, T> serializer(final Serializer<T> serializer) {
            final Options options = new Options(fileFactory, pageSizeBytes, scheduler);
            return new FlowableTransformer<T, T>() {

                @Override
                public Publisher<T> apply(Flowable<T> source) {
                    return new FlowableOnBackpressureBufferToFile<T>(source, null, options, serializer);
                }
            };
        }

        public <T> FlowableTransformer<T, T> serializer(DataSerializer<T> ds) {
            return serializer(Serializers.from(ds));
        }
    }

    public static final class BuilderObservable {

        private Callable<File> fileFactory = FileFactoryHolder.INSTANCE;
        private int pageSizeBytes = 1024 * 1024;
        private Optional<Scheduler> scheduler = Optional.absent();

        BuilderObservable() {
        }

        /**
         * Sets the page size in bytes. A page corresponds to a single memory
         * mapped file. If this method is not called the default value is 1MB
         * (1024*1024 bytes).
         * 
         * @param pageSizeBytes
         *            the page size in bytes.
         * @return this
         */
        public BuilderObservable pageSizeBytes(int pageSizeBytes) {
            this.pageSizeBytes = pageSizeBytes;
            return this;
        }

        /**
         * Sets the scheduler to use for reading items from files and emitting
         * them.
         * 
         * @param scheduler
         *            for emitting items
         * @return this
         */
        public BuilderObservable scheduler(Scheduler scheduler) {
            this.scheduler = Optional.of(scheduler);
            return this;
        }

        /**
         * Sets the file factory to be used by the queue storage mechanism.
         * Defaults to using {@code File.createTempFile("bufferToFileDb","")} if
         * this method is not called.
         * 
         * @param fileFactory
         *            the factory
         * @return the current builder
         */
        public BuilderObservable fileFactory(Callable<File> fileFactory) {
            this.fileFactory = fileFactory;
            return this;
        }

        /**
         * Returns a Flowable Transformer to buffer stream items to disk
         * (files). The serialization method uses
         * {@code java.io.ObjectOutputStream} for writing values to disk and
         * {@code java.io.ObjectInputStream} for reading values from disk and
         * emitting them. Note that this is a pretty verbose serialization
         * method for small items in that a 4 byte Integer is serialized to an
         * 81 byte array!
         * 
         * @param <T>
         *            stream item type
         * @return FlowableTransformer using the java io serializer
         */
        public <T extends Serializable> Function<Observable<T>, Flowable<T>> serializerJavaIO() {
            return serializer(Serializers.<T>javaIO());
        }

        /**
         * Returns a Flowable Transformer that buffers items to disk (files).
         * The serialization method passes through byte arrays directly for both
         * writing and reading.
         * 
         * @return FlowableTransformer using the bytes serializer
         */
        public Function<Observable<byte[]>, Flowable<byte[]>> serializerBytes() {
            return serializer(Serializers.bytes());
        }

        public Function<Observable<String>, Flowable<String>> serializerUtf8() {
            return serializer(Serializers.utf8());
        }

        public <T> Function<Observable<T>, Flowable<T>> serializer(final Serializer<T> serializer) {
            Scheduler s = scheduler.isPresent() ? scheduler.get() : Schedulers.io();
            final Options options = new Options(fileFactory, pageSizeBytes, s);
            return new Function<Observable<T>, Flowable<T>>() {

                @Override
                public Flowable<T> apply(Observable<T> source) {
                    return new FlowableOnBackpressureBufferToFile<T>(null, source, options, serializer);
                }
            };
        }

        public <T> Function<Observable<T>, Flowable<T>> serializer(DataSerializer<T> ds) {
            return serializer(Serializers.from(ds));
        }
    }

    @VisibleForTesting
    static final class FileFactoryHolder {
        private static final Callable<File> INSTANCE = new Callable<File>() {
            @Override
            public File call() {
                try {
                    return File.createTempFile(DEFAULT_FILE_PREFIX, ".obj");
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        };
    }

} public final class FlowableCollectWhile<T, R> extends Flowable<R> {

    private final Flowable<T> source;
    private final Callable<R> collectionFactory;
    private final BiFunction<? super R, ? super T, ? extends R> add;
    private final BiPredicate<? super R, ? super T> condition;
    private final boolean emitRemainder;

    public FlowableCollectWhile(Flowable<T> source, Callable<R> collectionFactory,
            BiFunction<? super R, ? super T, ? extends R> add,
            BiPredicate<? super R, ? super T> condition, boolean emitRemainder) {
        super();
        this.source = source;
        this.collectionFactory = collectionFactory;
        this.add = add;
        this.condition = condition;
        this.emitRemainder = emitRemainder;
    }

    @Override
    protected void subscribeActual(Subscriber<? super R> child) {
        CollectWhileSubscriber<T, R> subscriber = new CollectWhileSubscriber<T, R>(
                collectionFactory, add, condition, child, emitRemainder);
        source.subscribe(subscriber);
    }

    @SuppressWarnings("serial")
    private static final class CollectWhileSubscriber<T, R> extends AtomicInteger
            implements FlowableSubscriber<T>, Subscription, ConditionalSubscriber<T> {

        private final Callable<R> collectionFactory;
        private final BiFunction<? super R, ? super T, ? extends R> add;
        private final BiPredicate<? super R, ? super T> condition;
        private final Subscriber<? super R> child;
        private final boolean emitRemainder;
        private final AtomicLong requested = new AtomicLong();
        private final SimplePlainQueue<R> queue = new SpscLinkedArrayQueue<R>(16);

        private Subscription parent;
        private R collection;
        private volatile boolean done;
        private Throwable error; // does not need to be volatile because is set
                                 // before `done` and read after `done`

        private volatile boolean cancelled;

        CollectWhileSubscriber(Callable<R> collectionFactory,
                BiFunction<? super R, ? super T, ? extends R> add,
                BiPredicate<? super R, ? super T> condition, Subscriber<? super R> child,
                boolean emitRemainder) {
            this.collectionFactory = collectionFactory;
            this.add = add;
            this.condition = condition;
            this.child = child;
            this.emitRemainder = emitRemainder;
        }

        @Override
        public void onSubscribe(Subscription parent) {
            if (SubscriptionHelper.validate(this.parent, parent)) {
                this.parent = parent;
                child.onSubscribe(this);
            }
        }

        @Override
        public void onNext(T t) {
            // this path taken by upstream if not enabled to call `tryOnNext`
            if (!tryOnNext(t)) {
                parent.request(1);
            }
        }

        @Override
        public boolean tryOnNext(T t) {
            if (done) {
                return true;
            }
            if (collection == null && !collectionCreated()) {
                return true;
            }
            boolean collect;
            try {
                collect = condition.test(collection, t);
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                onError(e);
                return true;
            }
            if (!collect) {
                queue.offer(collection);
                if (!collectionCreated()) {
                    return true;
                }
            }
            try {
                collection = add.apply(collection, t);
                if (collection == null) {
                    throw new NullPointerException("add function should not return null");
                }
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                onError(e);
                return true;
            }
            drain();
            return !collect;
        }

        public boolean collectionCreated() {
            try {
                collection = collectionFactory.call();
                if (collection == null) {
                    throw new NullPointerException("collectionFactory should not return null");
                }
                return true;
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                onError(e);
                return false;
            }
        }

        @Override
        public void onError(Throwable e) {
            if (done) {
                RxJavaPlugins.onError(e);
                return;
            }
            //release for GC
            collection = null;
            // must set `error` before done because `error` is not volatile and
            // `done` is
            error = e;
            done = true;
            drain();
        }

        @Override
        public void onComplete() {
            if (done) {
                return;
            }
            R col = collection;
            if (col != null) {
                collection = null;
                // ensure that the remainder is emitted
                // if configured to
                if (emitRemainder) {
                    queue.offer(col);
                }
            }
            done = true;
            drain();
        }

        private void drain() {
            if (getAndIncrement() == 0) {
                int missed = 1;
                while (true) {
                    long r = requested.get();
                    long e = 0;
                    while (e != r) {
                        if (cancelled) {
                            queue.clear();
                            return;
                        }
                        //must read `done` before polling queue
                        boolean d = done;
                        R c = queue.poll();
                        if (c == null) {
                            if (d) {
                                // `error` must be read AFTER `done` for
                                // full visibility
                                Throwable err = error;
                                if (err != null) {
                                    error = null;
                                    child.onError(err);
                                } else {
                                    child.onComplete();
                                }
                                return;
                            } else {
                                // nothing to emit and not done
                                break;
                            }
                        } else {
                            child.onNext(c);
                            e++;
                        }
                    }
                    if (e != 0L && r != Long.MAX_VALUE) {
                        requested.addAndGet(-e);
                    }
                    missed = addAndGet(-missed);
                    if (missed == 0) {
                        return;
                    }
                }
            }

        }

        @Override
        public void request(long n) {
            if (SubscriptionHelper.validate(n)) {
                BackpressureHelper.add(requested, n);
                parent.request(n);
                drain();
            }
        }

        @Override
        public void cancel() {
            cancelled = true;
            parent.cancel();
        }

    }
} public final class FlowableDoOnEmpty<T> extends Flowable<T> {

    private final Publisher<T> source;
    private final Action onEmpty;

    public FlowableDoOnEmpty(Publisher<T> source, Action onEmpty) {
        Preconditions.checkNotNull(source, "source cannot be null");
        Preconditions.checkNotNull(onEmpty, "onEmpty cannot be null");
        this.source = source;
        this.onEmpty = onEmpty;
    }

    @Override
    protected void subscribeActual(Subscriber<? super T> child) {
        source.subscribe(new DoOnEmptySubscriber<T>(child, onEmpty));
    }

    private static final class DoOnEmptySubscriber<T> implements FlowableSubscriber<T>, Subscription {

        private final Subscriber<? super T> child;
        private final Action onEmpty;
        // mutable state
        private boolean done;
        private boolean empty = true;
        private Subscription parent;

        DoOnEmptySubscriber(Subscriber<? super T> child, Action onEmpty) {
            this.child = child;
            this.onEmpty = onEmpty;
        }

        @Override
        public void onSubscribe(Subscription parent) {
            if (SubscriptionHelper.validate(this.parent, parent)) {
                this.parent = parent;
                child.onSubscribe(this);
            }
        }

        @Override
        public void onComplete() {
            if (done) {
                return;
            }
            if (empty) {
                try {
                    onEmpty.run();
                } catch (Throwable e) {
                    Exceptions.throwIfFatal(e);
                    onError(e);
                    return;
                }
            }
            done = true;
            child.onComplete();
        }

        @Override
        public void onNext(T t) {
            empty = false;
            child.onNext(t);
        }

        @Override
        public void onError(Throwable e) {
            if (done) {
                return;
            }
            done = true;
            child.onError(e);
        }

        @Override
        public void cancel() {
            parent.cancel();
        }

        @Override
        public void request(long n) {
            parent.request(n);
        }
    }
} public final class FlowableInsertMaybe<T> extends Flowable<T> {

    private final Flowable<T> source;
    private final Function<? super T, ? extends Maybe<? extends T>> valueToInsert;

    public FlowableInsertMaybe(Flowable<T> source, Function<? super T, ? extends Maybe<? extends T>> valueToInsert) {
        Preconditions.checkNotNull(valueToInsert, "valueToInsert cannot be null");
        this.source = source;
        this.valueToInsert = valueToInsert;
    }

    @Override
    protected void subscribeActual(Subscriber<? super T> downstream) {
        source.subscribe(new InsertSubscriber<T>(downstream, valueToInsert));
    }

    static final class InsertSubscriber<T> extends AtomicInteger implements FlowableSubscriber<T>, Subscription {

        private static final long serialVersionUID = -15415234346097063L;

        private final Subscriber<? super T> downstream;
        private final Function<? super T, ? extends Maybe<? extends T>> valueToInsert;
        private final SimplePlainQueue<T> queue;
        private final AtomicLong requested;
        private final AtomicLong inserted;

        // define as type Object so can set with a non-null sentinel object that does
        // not gather a stacktrace. A further refinement would be to use just one
        // AtomicReference for `error` and `done` (leaving it for now)
        private final AtomicReference<Object> error;
        private final AtomicReference<Disposable> valueToInsertObserver;

        private Subscription upstream;
        private volatile boolean done;
        private volatile boolean cancelled;

        // used to prevent emission of events after a terminal event
        // does not need to be volatile
        private boolean finished;

        InsertSubscriber(Subscriber<? super T> downstream,
                Function<? super T, ? extends Maybe<? extends T>> valueToInsert) {
            this.downstream = downstream;
            this.valueToInsert = valueToInsert;
            this.queue = new MpscLinkedQueue<T>();
            this.requested = new AtomicLong();
            this.inserted = new AtomicLong();
            this.error = new AtomicReference<Object>();
            this.valueToInsertObserver = new AtomicReference<Disposable>();
        }

        @Override
        public void onSubscribe(Subscription upstream) {
            if (SubscriptionHelper.validate(this.upstream, upstream)) {
                this.upstream = upstream;
                downstream.onSubscribe(this);
            }
        }

        @Override
        public void onNext(T t) {
            if (finished) {
                return;
            }
            queue.offer(t);
            Maybe<? extends T> maybe;
            try {
                maybe = valueToInsert.apply(t);
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                // we cancel upstream ourselves because the
                // error did not originate from source
                upstream.cancel();
                onError(e);
                return;
            }
            ValueToInsertObserver<T> o = new ValueToInsertObserver<T>(this);
            if (DisposableHelper.set(valueToInsertObserver, o)) {
                // note that at this point we have to cover o being disposed
                // from another thread so the Observer class needs
                // to handle dispose being called before/during onSubscribe
                maybe.subscribe(o);
            }
            drain();
        }

        @Override
        public void onError(Throwable e) {
            if (finished) {
                RxJavaPlugins.onError(e);
                return;
            }
            finished = true;
            if (error.compareAndSet(null, e)) {
                DisposableHelper.dispose(valueToInsertObserver);
                done = true;
                drain();
            } else {
                RxJavaPlugins.onError(e);
            }
        }

        @Override
        public void onComplete() {
            if (finished) {
                return;
            }
            finished = true;
            DisposableHelper.dispose(valueToInsertObserver);
            done = true;
            drain();
        }

        private void drain() {
            if (getAndIncrement() != 0) {
                return;
            }
            // note that this drain loop does not shortcut errors
            int missed = 1;
            while (true) {
                long r = requested.get();
                long e = 0;
                while (e != r) {
                    if (cancelled) {
                        DisposableHelper.dispose(valueToInsertObserver);
                        queue.clear();
                        return;
                    }
                    // must read `done` before polling queue
                    boolean d = done;
                    T t = queue.poll();
                    if (t == null) {
                        if (d) {
                            Object err = error.get();
                            if (err != null) {
                                // clear the exception so can be gc'd
                                // `this` is not a real error, it just prevents
                                // it getting set again in a race because the other
                                // setters which use CAS assume initial value of null
                                error.set(this);
                                DisposableHelper.dispose(valueToInsertObserver);
                                downstream.onError((Throwable) err);
                            } else {
                                // don't need to dispose valueToInsertObserver because already done in
                                // onComplete
                                downstream.onComplete();
                            }
                            return;
                        } else {
                            // nothing to emit and not done
                            break;
                        }
                    } else {
                        downstream.onNext(t);
                        e++;
                    }
                }
                if (e != 0L && r != Long.MAX_VALUE) {
                    requested.addAndGet(-e);
                }
                missed = addAndGet(-missed);
                if (missed == 0) {
                    return;
                }
            }
        }

        @Override
        public void request(long n) {
            if (SubscriptionHelper.validate(n)) {
                BackpressureHelper.add(requested, n);
                // modify request to upstream to account for inserted values
                // use a CAS loop because request can be called from any thread
                while (true) {
                    long ins = inserted.get();
                    long d = Math.min(ins, n);
                    if (inserted.compareAndSet(ins, ins - d)) {
                        if (n - d > 0) {
                            upstream.request(n - d);
                        }
                        break;
                    }
                }
                drain();
            }
        }

        @Override
        public void cancel() {
            if (!cancelled) {
                cancelled = true;
                upstream.cancel();
                DisposableHelper.dispose(valueToInsertObserver);
                if (getAndIncrement() == 0) {
                    // use the same access control to queue as drain method
                    // because `clear` just calls `queue.poll()` repeatedly till nothing left on the
                    // queue (ignoring the dequeued items).
                    //
                    // this is best endeavours, there still exists a race with onNext and drain
                    // where items could be left on the queue after cancel
                    queue.clear();
                }
            }
        }

        void insert(T t) {
            inserted.incrementAndGet();
            queue.offer(t);
            drain();
        }

        void insertError(Throwable e) {
            if (error.compareAndSet(null, e)) {
                upstream.cancel();
                DisposableHelper.dispose(valueToInsertObserver);
                done = true;
                drain();
            } else {
                RxJavaPlugins.onError(e);
            }
        }

    }

    static final class ValueToInsertObserver<T> extends AtomicReference<Disposable>
            implements MaybeObserver<T>, Disposable {

        private static final long serialVersionUID = 41384726414575403L;

        private final InsertSubscriber<T> downstream;

        ValueToInsertObserver(InsertSubscriber<T> downstream) {
            this.downstream = downstream;
        }

        @Override
        public void onSubscribe(Disposable upstream) {
            // an AtomicReference is used to hold the upstream Disposable
            // because this Observer can be disposed before onSubscribe
            // is called (contrary to the normal contract).
            DisposableHelper.setOnce(this, upstream);
        }

        @Override
        public void onSuccess(T t) {
            lazySet(DisposableHelper.DISPOSED);
            downstream.insert(t);
        }

        @Override
        public void onError(Throwable e) {
            lazySet(DisposableHelper.DISPOSED);
            downstream.insertError(e);
        }

        @Override
        public void onComplete() {
            lazySet(DisposableHelper.DISPOSED);
            // don't do anything else because no value to insert was reported
        }

        @Override
        public void dispose() {
            DisposableHelper.dispose(this);
        }

        @Override
        public boolean isDisposed() {
            return get() == DisposableHelper.DISPOSED;
        }

    }
} public final class FlowableInsertTimeout<T> extends Flowable<T> {

    private final Flowable<T> source;
    private final Function<? super T, ? extends Long> timeout;
    private final TimeUnit unit;
    private final Function<? super T, ? extends T> value;
    private final Scheduler scheduler;

    public FlowableInsertTimeout(Flowable<T> source, Function<? super T, ? extends Long> timeout, TimeUnit unit,
            Function<? super T, ? extends T> value, Scheduler scheduler) {
        Preconditions.checkNotNull(timeout, "timeout cannot be null");
        Preconditions.checkNotNull(unit, "unit cannot be null");
        Preconditions.checkNotNull(value, "value cannot be null");
        Preconditions.checkNotNull(scheduler, "scheduler cannot be null");
        this.source = source;
        this.timeout = timeout;
        this.unit = unit;
        this.value = value;
        this.scheduler = scheduler;
    }

    @Override
    protected void subscribeActual(Subscriber<? super T> downstream) {
        source.subscribe(new InsertTimeoutSubscriber<T>(downstream, timeout, unit, value, scheduler));
    }

    static final class InsertTimeoutSubscriber<T> extends AtomicInteger implements FlowableSubscriber<T>, Subscription {

        private static final long serialVersionUID = 1812803226317104954L;

        private static final Object TERMINATED = new Object();

        private final Subscriber<? super T> downstream;
        private final Function<? super T, ? extends Long> timeout;
        private final TimeUnit unit;
        private final Function<? super T, ? extends T> value;

        private final SimplePlainQueue<T> queue;
        private final AtomicLong requested;
        private final AtomicLong inserted;

        // set either with null (not terminated), Throwable (error), or
        // TERMINATED (completion). TERMINATED is also used to replace an 
        // error after it is reported downstream so that the local copy of the 
        // error can be gc'd
        private final AtomicReference<Object> terminated;
        private final AtomicReference<Disposable> scheduled;

        private Subscription upstream;
        private volatile boolean cancelled;

        // used to prevent emission of events after a terminal event
        // does not need to be volatile
        private boolean finished;

        private final Worker worker;

        InsertTimeoutSubscriber(Subscriber<? super T> downstream, Function<? super T, ? extends Long> timeout,
                TimeUnit unit, Function<? super T, ? extends T> value, Scheduler scheduler) {
            this.downstream = downstream;
            this.timeout = timeout;
            this.unit = unit;
            this.value = value;
            this.queue = new MpscLinkedQueue<T>();
            this.requested = new AtomicLong();
            this.inserted = new AtomicLong();
            this.terminated = new AtomicReference<Object>();
            this.scheduled = new AtomicReference<Disposable>();
            this.worker = scheduler.createWorker();
        }

        @Override
        public void onSubscribe(Subscription upstream) {
            if (SubscriptionHelper.validate(this.upstream, upstream)) {
                this.upstream = upstream;
                downstream.onSubscribe(this);
            }
        }

        @Override
        public void onNext(final T t) {
            if (finished) {
                return;
            }
            queue.offer(t);
            final long waitTime;
            try {
                waitTime = timeout.apply(t);
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                // we cancel upstream ourselves because the
                // error did not originate from source
                upstream.cancel();
                onError(e);
                return;
            }
            TimeoutAction<T> action = new TimeoutAction<T>(this, t);
            Disposable d = worker.schedule(action, waitTime, unit);
            DisposableHelper.set(scheduled, d);
            drain();
        }

        @Override
        public void onError(Throwable e) {
            if (finished) {
                RxJavaPlugins.onError(e);
                return;
            }
            finished = true;
            if (terminated.compareAndSet(null, e)) {
                dispose();
                drain();
            } else {
                RxJavaPlugins.onError(e);
            }
        }

        @Override
        public void onComplete() {
            if (finished) {
                return;
            }
            finished = true;
            if (terminated.compareAndSet(null, TERMINATED)) {
                dispose();
                drain();
            }
        }

        private void drain() {
            if (getAndIncrement() != 0) {
                return;
            }
            // note that this drain loop does not shortcut errors
            int missed = 1;
            while (true) {
                long r = requested.get();
                long e = 0;
                while (e != r) {
                    if (cancelled) {
                        dispose();
                        queue.clear();
                        return;
                    }
                    Object d = terminated.get();
                    T t = queue.poll();
                    if (t == null) {
                        if (d != null) {
                            // d will be either TERMINATED or a Throwable
                            if (d == TERMINATED) {
                                // don't need to dispose scheduled because already done in
                                // onComplete
                                downstream.onComplete();
                            } else {
                                // clear the exception so can be gc'd
                                // setting the value to TERMINATED just prevents it getting set again in a race
                                // because the other setters which use CAS assume initial value of null
                                terminated.set(TERMINATED);
                                dispose();
                                downstream.onError((Throwable) d);
                            }
                            return;
                        } else {
                            // nothing to emit and not done
                            break;
                        }
                    } else {
                        downstream.onNext(t);
                        e++;
                    }
                }
                if (e != 0L && r != Long.MAX_VALUE) {
                    requested.addAndGet(-e);
                }
                missed = addAndGet(-missed);
                if (missed == 0) {
                    return;
                }
            }
        }

        @Override
        public void request(long n) {
            if (SubscriptionHelper.validate(n)) {
                BackpressureHelper.add(requested, n);
                // modify request to upstream to account for inserted values
                // use a CAS loop because request can be called from any thread
                while (true) {
                    long ins = inserted.get();
                    long d = Math.min(ins, n);
                    if (inserted.compareAndSet(ins, ins - d)) {
                        if (n - d > 0) {
                            upstream.request(n - d);
                        }
                        break;
                    }
                }
                drain();
            }
        }

        @Override
        public void cancel() {
            if (!cancelled) {
                cancelled = true;
                upstream.cancel();
                dispose();
                if (getAndIncrement() == 0) {
                    // use the same access control to queue as drain method
                    // because `clear` just calls `queue.poll()` repeatedly till nothing left on the
                    // queue (ignoring the dequeued items).
                    //
                    // this is best endeavours, there still exists a race with onNext and drain
                    // where items could be left on the queue after cancel
                    queue.clear();
                }
            }
        }

        private void dispose() {
            DisposableHelper.dispose(scheduled);
            worker.dispose();
        }
        
        void insert(T t) {
            inserted.incrementAndGet();
            queue.offer(t);
            drain();
        }

        void insertError(Throwable e) {
            if (terminated.compareAndSet(null, e)) {
                upstream.cancel();
                dispose();
                drain();
            } else {
                RxJavaPlugins.onError(e);
            }
        }

        T calculateValueToInsert(T t) throws Exception {
            return value.apply(t);
        }

    }

    private static final class TimeoutAction<T> implements Runnable {

        private final InsertTimeoutSubscriber<T> subscriber;
        private final T t;

        TimeoutAction(InsertTimeoutSubscriber<T> subscriber, T t) {
            this.subscriber = subscriber;
            this.t = t;
        }

        @Override
        public void run() {
            final T v;
            try {
                v = subscriber.calculateValueToInsert(t);
            } catch (Throwable e) {
                subscriber.insertError(e);
                return;
            }
            subscriber.insert(v);
        }

    }

} public final class FlowableMapLast<T> extends Flowable<T> {

    private final Flowable<T> source;
    private final Function<? super T, ? extends T> function;

    public FlowableMapLast(Flowable<T> source, Function<? super T, ? extends T> function) {
        this.source = source;
        this.function = function;
    }

    @Override
    protected void subscribeActual(Subscriber<? super T> s) {
        source.subscribe(new MapLastSubscriber<T>(s, function));
    }

    private final static class MapLastSubscriber<T> implements FlowableSubscriber<T>, Subscription {

        private static final Object EMPTY = new Object();

        private final Subscriber<? super T> actual;
        private final Function<? super T, ? extends T> function;
        private final AtomicBoolean firstRequest = new AtomicBoolean(true);

        // mutable state
        @SuppressWarnings("unchecked")
        private T value = (T) EMPTY;
        private Subscription parent;
        private boolean done;

        public MapLastSubscriber(Subscriber<? super T> actual, Function<? super T, ? extends T> function) {
            this.actual = actual;
            this.function = function;
        }

        @Override
        public void onSubscribe(Subscription subscription) {
            if (SubscriptionHelper.validate(this.parent, subscription)) {
                this.parent = subscription;
                actual.onSubscribe(this);
            }
        }

        @Override
        public void onNext(T t) {
            if (done) {
                return;
            }
            if (value == EMPTY) {
                value = t;
            } else {
                actual.onNext(value);
                value = t;
            }
        }

        @Override
        public void onComplete() {
            if (done) {
                return;
            }
            if (value != EMPTY) {
                T value2;
                try {
                    value2 = function.apply(value);
                } catch (Throwable e) {
                    Exceptions.throwIfFatal(e);
                    parent.cancel();
                    onError(e);
                    return;
                }
                actual.onNext(value2);
            }
            done = true;
            actual.onComplete();
        }

        @Override
        public void onError(Throwable e) {
            if (done) {
                return;
            }
            if (value != EMPTY) {
                actual.onNext(value);
            }
            done = true;
            actual.onError(e);
        }

        @Override
        public void cancel() {
            parent.cancel();
        }

        @Override
        public void request(long n) {
            if (SubscriptionHelper.validate(n)) {
                if (firstRequest.compareAndSet(true, false)) {
                    long m = n + 1;
                    if (m < 0) {
                        m = Long.MAX_VALUE;
                    }
                    parent.request(m);
                } else {
                    parent.request(n);
                }
            }
        }

    }

} public final class FlowableMatch<A, B, K, C> extends Flowable<C> {

    private final Flowable<A> a;
    private final Flowable<B> b;
    private final Function<? super A, ? extends K> aKey;
    private final Function<? super B, ? extends K> bKey;
    private final BiFunction<? super A, ? super B, C> combiner;
    private final long requestSize;

    public FlowableMatch(Flowable<A> a, Flowable<B> b, Function<? super A, ? extends K> aKey,
            Function<? super B, ? extends K> bKey, BiFunction<? super A, ? super B, C> combiner, long requestSize) {
        Preconditions.checkNotNull(a, "a should not be null");
        Preconditions.checkNotNull(b, "b should not be null");
        Preconditions.checkNotNull(aKey, "aKey cannot be null");
        Preconditions.checkNotNull(bKey, "bKey cannot be null");
        Preconditions.checkNotNull(combiner, "combiner cannot be null");
        Preconditions.checkArgument(requestSize >= 1, "requestSize must be >=1");
        this.a = a;
        this.b = b;
        this.aKey = aKey;
        this.bKey = bKey;
        this.combiner = combiner;
        this.requestSize = requestSize;
    }

    @Override
    protected void subscribeActual(Subscriber<? super C> child) {
        MatchCoordinator<A, B, K, C> coordinator = new MatchCoordinator<A, B, K, C>(aKey, bKey, combiner, requestSize,
                child);
        child.onSubscribe(coordinator);
        coordinator.subscribe(a, b);
    }

    interface Receiver {
        void offer(Object item);
    }

    @SuppressWarnings("serial")
    private static final class MatchCoordinator<A, B, K, C> extends AtomicInteger implements Receiver, Subscription {
        private final Map<K, Queue<A>> as = new HashMap<K, Queue<A>>();
        private final Map<K, Queue<B>> bs = new HashMap<K, Queue<B>>();
        private final Function<? super A, ? extends K> aKey;
        private final Function<? super B, ? extends K> bKey;
        private final BiFunction<? super A, ? super B, C> combiner;
        private final long requestSize;
        private final transient SimpleQueue<Object> queue;
        private final Subscriber<? super C> child;
        private final AtomicLong requested = new AtomicLong(0);

        // mutable fields, guarded by `this` atomics
        private int requestFromA = 0;
        private int requestFromB = 0;

        // completion state machine
        private int completed = COMPLETED_NONE;
        // completion states
        private static final int COMPLETED_NONE = 0;
        private static final int COMPLETED_A = 1;
        private static final int COMPLETED_B = 2;
        private static final int COMPLETED_BOTH = 3;

        private MySubscriber<A, K> aSub;
        private MySubscriber<B, K> bSub;

        private volatile boolean cancelled = false;

        MatchCoordinator(Function<? super A, ? extends K> aKey, Function<? super B, ? extends K> bKey,
                BiFunction<? super A, ? super B, C> combiner, long requestSize, Subscriber<? super C> child) {
            this.aKey = aKey;
            this.bKey = bKey;
            this.combiner = combiner;
            this.requestSize = requestSize;
            this.queue = new MpscLinkedQueue<Object>();
            this.child = child;
        }

        public void subscribe(Flowable<A> a, Flowable<B> b) {
            aSub = new MySubscriber<A, K>(Source.A, this, requestSize);
            bSub = new MySubscriber<B, K>(Source.B, this, requestSize);
            a.subscribe(aSub);
            b.subscribe(bSub);
        }

        @Override
        public void request(long n) {
            if (SubscriptionHelper.validate(n)) {
                BackpressureHelper.add(requested, n);
                drain();
            }
        }

        @Override
        public void cancel() {
            if (!cancelled) {
                cancelled = true;
                cancelAll();
            }
        }

        void cancelAll() {
            aSub.cancel();
            bSub.cancel();
        }

        void drain() {
            if (getAndIncrement() != 0) {
                // work already in progress
                // so exit
                return;
            }
            int missed = 1;
            while (true) {
                long r = requested.get();
                long emitted = 0;
                while (emitted != r) {
                    if (cancelled) {
                        return;
                    }
                    // note no null values on the queue
                    Object v;
                    try {
                        v = queue.poll();
                    } catch (Exception e) {
                        Exceptions.throwIfFatal(e);
                        clear();
                        child.onError(e);
                        return;
                    }
                    if (v == null) {
                        // queue is empty
                        break;
                    } else if (v instanceof ItemA) {
                        Emitted em = handleItem(((ItemA) v).value, Source.A);
                        if (em == Emitted.FINISHED) {
                            return;
                        } else if (em == Emitted.ONE) {
                            emitted += 1;
                        }
                    } else if (v instanceof Source) {
                        // source completed
                        Status status = handleCompleted((Source) v);
                        if (status == Status.FINISHED) {
                            return;
                        }
                    } else if (v instanceof MyError) {
                        // v must be an error
                        clear();
                        child.onError(((MyError) v).error);
                        return;
                    } else {
                        // is onNext from B
                        Emitted em = handleItem(v, Source.B);
                        if (em == Emitted.FINISHED) {
                            return;
                        } else if (em == Emitted.ONE) {
                            emitted += 1;
                        }
                    }
                    if (r == emitted) {
                        break;
                    }
                }
                // reduce requested by emitted which will always be positive
                BackpressureHelper.produced(requested, emitted);
                missed = this.addAndGet(-missed);
                if (missed == 0) {
                    return;
                }
            }
        }

        private Emitted handleItem(Object value, Source source) {
            final Emitted result;

            // logic duplication occurs below
            // would be nice to simplify without making code
            // unreadable. A bit of a toss-up.
            if (source == Source.A) {
                // look for match
                @SuppressWarnings("unchecked")
                A a = (A) value;
                K key;
                try {
                    key = aKey.apply(a);
                } catch (Throwable e) {
                    clear();
                    child.onError(e);
                    return Emitted.FINISHED;
                }
                Queue<B> q = bs.get(key);
                if (q == null) {
                    // cache value
                    add(as, key, a);
                    result = Emitted.NONE;
                } else {
                    // emit match
                    B b = poll(bs, q, key);
                    C c;
                    try {
                        c = combiner.apply(a, b);
                    } catch (Throwable e) {
                        clear();
                        child.onError(e);
                        return Emitted.FINISHED;
                    }
                    child.onNext(c);
                    result = Emitted.ONE;
                }
                // if the other source has completed and there
                // is nothing to match with then we should stop
                if (completed == COMPLETED_B && bs.isEmpty()) {
                    // can finish
                    clear();
                    child.onComplete();
                    return Emitted.FINISHED;
                } else {
                    requestFromA += 1;
                }
            } else {
                // look for match
                @SuppressWarnings("unchecked")
                B b = (B) value;
                K key;
                try {
                    key = bKey.apply(b);
                } catch (Throwable e) {
                    clear();
                    child.onError(e);
                    return Emitted.FINISHED;
                }
                Queue<A> q = as.get(key);
                if (q == null) {
                    // cache value
                    add(bs, key, b);
                    result = Emitted.NONE;
                } else {
                    // emit match
                    A a = poll(as, q, key);
                    C c;
                    try {
                        c = combiner.apply(a, b);
                    } catch (Throwable e) {
                        clear();
                        child.onError(e);
                        return Emitted.FINISHED;
                    }
                    child.onNext(c);
                    result = Emitted.ONE;
                }
                // if the other source has completed and there
                // is nothing to match with then we should stop
                if (completed == COMPLETED_A && as.isEmpty()) {
                    // can finish
                    clear();
                    child.onComplete();
                    return Emitted.FINISHED;
                } else {
                    requestFromB += 1;
                }
            }
            // requests are batched so that each source gets a turn
            checkToRequestMore();
            return result;
        }

        private enum Emitted {
            ONE, NONE, FINISHED;
        }

        private Status handleCompleted(Source source) {
            completed(source);
            final boolean done;
            if (source == Source.A) {
                aSub.cancel();
                done = (completed == COMPLETED_BOTH) || (completed == COMPLETED_A && as.isEmpty());
            } else {
                bSub.cancel();
                done = (completed == COMPLETED_BOTH) || (completed == COMPLETED_B && bs.isEmpty());
            }
            if (done) {
                clear();
                child.onComplete();
                return Status.FINISHED;
            } else {
                checkToRequestMore();
                return Status.KEEP_GOING;
            }
        }

        private enum Status {
            FINISHED, KEEP_GOING;
        }

        private void checkToRequestMore() {
            if (requestFromA == requestSize && completed == COMPLETED_B) {
                requestFromA = 0;
                aSub.request(requestSize);
            } else if (requestFromB == requestSize && completed == COMPLETED_A) {
                requestFromB = 0;
                bSub.request(requestSize);
            } else if (requestFromA == requestSize && requestFromB == requestSize) {
                requestFromA = 0;
                requestFromB = 0;
                aSub.request(requestSize);
                bSub.request(requestSize);
            }
        }

        private void completed(Source source) {
            if (source == Source.A) {
                if (completed == COMPLETED_NONE) {
                    completed = COMPLETED_A;
                } else if (completed == COMPLETED_B) {
                    completed = COMPLETED_BOTH;
                }
            } else {
                if (completed == COMPLETED_NONE) {
                    completed = COMPLETED_B;
                } else if (completed == COMPLETED_A) {
                    completed = COMPLETED_BOTH;
                }
            }
        }

        private void clear() {
            as.clear();
            bs.clear();
            queue.clear();
            aSub.cancel();
            bSub.cancel();
        }

        private static <K, T> void add(Map<K, Queue<T>> map, K key, T value) {
            Queue<T> q = map.get(key);
            if (q == null) {
                q = new LinkedList<T>();
                map.put(key, q);
            }
            q.offer(value);
        }

        private static <K, T> T poll(Map<K, Queue<T>> map, Queue<T> q, K key) {
            T t = q.poll();
            if (q.isEmpty()) {
                map.remove(key);
            }
            return t;
        }

        @Override
        public void offer(Object item) {
            queue.offer(item);
            drain();
        }

    }

    @SuppressWarnings("serial")
    private static final class MySubscriber<T, K> extends AtomicReference<Subscription> implements FlowableSubscriber<T>, Subscription {

        private final Receiver receiver;
        private final Source source;
        private final long requestSize;

        MySubscriber(Source source, Receiver receiver, long requestSize) {
            this.source = source;
            this.receiver = receiver;
            this.requestSize = requestSize;
        }

        @Override
        public void onSubscribe(Subscription subscription) {
            if (SubscriptionHelper.setOnce(this, subscription)) {
                subscription.request(requestSize);
            }
        }

        @Override
        public void request(long n) {
            get().request(n);
        }

        @Override
        public void cancel() {
            SubscriptionHelper.cancel(this);
        }

        @Override
        public void onNext(T t) {
            if (source == Source.A) {
                receiver.offer(new ItemA(t));
            } else {
                receiver.offer(t);
            }
        }

        @Override
        public void onComplete() {
            receiver.offer(source);
        }

        @Override
        public void onError(Throwable e) {
            receiver.offer(new MyError(e));
        }

    }

    private static final class MyError {
        final Throwable error;

        MyError(Throwable error) {
            this.error = error;
        }
    }

    private static final class ItemA {
        final Object value;

        ItemA(Object value) {
            this.value = value;
        }
    }

    private enum Source {
        A, B;
    }

} public final class FlowableMaxRequest<T> extends Flowable<T> {

    private final Flowable<T> source;
    private final long[] maxRequests;

    public FlowableMaxRequest(Flowable<T> source, long[] maxRequests) {
        Preconditions.checkArgument(maxRequests.length > 0, "maxRequests length must be greater than 0");
        for (int i = 0; i < maxRequests.length; i++) {
            Preconditions.checkArgument(maxRequests[i] > 0, "maxRequests items must be greater than zero");
        }
        this.source = source;
        this.maxRequests = maxRequests;
    }

    @Override
    protected void subscribeActual(Subscriber<? super T> child) {
        source.subscribe(new MaxRequestSubscriber<T>(maxRequests, child));
    }

    @SuppressWarnings("serial")
    private static final class MaxRequestSubscriber<T> extends AtomicInteger implements FlowableSubscriber<T>, Subscription {

        private final long[] maxRequests;
        private int requestNum;
        private final Subscriber<? super T> child;

        // the number of requests from downstream that have not been requested
        // from upstream yet
        private final AtomicLong requested = new AtomicLong();

        // the upstream subscription (which allows us to request from upstream
        // and cancel it)
        private Subscription parent;

        // the number of items still to be emitted from
        // upstream out of the last request to parent
        private long count;

        // when request made from `requestMore` this value is used to set the
        // next value of `count` in the `onNext` method
        private volatile long nextRequest;

        // indicates to the `requestMore` method that all items from the last
        // request to parent have arrived
        private volatile boolean allArrived = true;

        MaxRequestSubscriber(long[] maxRequests, Subscriber<? super T> child) {
            this.maxRequests = maxRequests;
            this.child = child;
        }

        @Override
        public void onSubscribe(Subscription parent) {
            if (SubscriptionHelper.validate(this.parent, parent)) {
                this.parent = parent;
                child.onSubscribe(this);
            }
        }

        @Override
        public void request(long n) {
            if (SubscriptionHelper.validate(n)) {
                BackpressureHelper.add(requested, n);
                requestMore();
            }
        }

        @Override
        public void cancel() {
            parent.cancel();
        }

        @Override
        public void onNext(T t) {
            if (count != Long.MAX_VALUE) {
                count--;
                if (count == -1) {
                    // request didn't happen from this onNext method so refresh
                    // count from the volatile set in `requestMore`
                    long nr = nextRequest;
                    if (nr == Long.MAX_VALUE) {
                        count = nr;
                    } else {
                        count = nr - 1;
                    }
                }
                if (count == 0) {
                    // All items from the last request made to parent have
                    // arrived

                    // CAS loop to update `requested`
                    long mr = peekNextMaxRequest();
                    while (true) {
                        long r = requested.get();
                        if (r == 0) {
                            // now must rely on dowstream requests to request
                            // more from upstream via `requestMore`
                            allArrived = true;
                            requestMore();
                            break;
                        } else if (r == Long.MAX_VALUE) {
                            nextMaxRequest();
                            count = mr;
                            parent.request(mr);
                            break;
                        } else {
                            long req = Math.min(r, mr);
                            if (requested.compareAndSet(r, r - req)) {
                                nextMaxRequest();
                                count = req;
                                parent.request(req);
                                break;
                            }
                        }
                    }
                }
            }
            child.onNext(t);
        }

        @Override
        public void onError(Throwable t) {
            child.onError(t);

        }

        @Override
        public void onComplete() {
            child.onComplete();
        }

        private void requestMore() {
            if (getAndIncrement() == 0) {
                int missed = 1;
                while (true) {
                    if (allArrived) {
                        // CAS loop to update requested
                        long mr = peekNextMaxRequest();
                        while (true) {
                            long r = requested.get();
                            long req = Math.min(r, mr);
                            if (r == 0) {
                                break;
                            } else if (r == Long.MAX_VALUE || requested.compareAndSet(r, r - req)) {
                                nextMaxRequest();
                                allArrived = false;
                                nextRequest = req;
                                parent.request(req);
                                break;
                            }
                        }
                    }
                    missed = addAndGet(-missed);
                    if (missed == 0) {
                        return;
                    }
                }
            }
        }

        private long peekNextMaxRequest() {
            return maxRequests[requestNum];
        }

        private void nextMaxRequest() {
            if (requestNum != maxRequests.length - 1) {
                requestNum++;
            }
        }

    }

} public final class FlowableMinRequest<T> extends Flowable<T> {

    private final Flowable<T> source;
    private final int[] minRequest;

    public FlowableMinRequest(Flowable<T> source, int[] minRequests) {
        Preconditions.checkArgument(minRequests.length > 0, "minRequests length must be > 0");
        for (int i = 0; i < minRequests.length; i++) {
            Preconditions.checkArgument(minRequests[i] > 0, "each item in minRequests must be > 0");
        }
        this.source = source;
        this.minRequest = minRequests;
    }

    @Override
    protected void subscribeActual(Subscriber<? super T> child) {
        source.subscribe(new MinRequestSubscriber<T>(minRequest, child));
    }

    @SuppressWarnings("serial")
    private static final class MinRequestSubscriber<T> extends AtomicInteger implements FlowableSubscriber<T>, Subscription {

        private final int[] minRequests;
        private int requestNum;
        private final Subscriber<? super T> child;
        private final AtomicLong requested = new AtomicLong();
        private final SimplePlainQueue<T> queue = new SpscLinkedArrayQueue<T>(16);

        private Subscription parent;
        private volatile boolean done;
        private Throwable error;
        private volatile boolean cancelled;
        private long count;

        MinRequestSubscriber(int[] minRequests, Subscriber<? super T> child) {
            this.minRequests = minRequests;
            this.child = child;
        }

        @Override
        public void onSubscribe(Subscription parent) {
            if (SubscriptionHelper.validate(this.parent, parent)) {
                this.parent = parent;
                child.onSubscribe(this);
            }
        }

        @Override
        public void request(long n) {
            if (SubscriptionHelper.validate(n)) {
                BackpressureHelper.add(requested, n);
                drain();
            }
        }

        @Override
        public void cancel() {
            cancelled = true;
            parent.cancel();
        }

        @Override
        public void onNext(T t) {
            queue.offer(t);
            drain();
        }

        @Override
        public void onError(Throwable e) {
            error = e;
            done = true;
            drain();
        }

        @Override
        public void onComplete() {
            done = true;
            drain();
        }

        private void drain() {
            if (getAndIncrement() == 0) {
                int missed = 1;
                while (true) {
                    long r = requested.get();
                    long e = 0;
                    boolean d = done;
                    while (e != r) {
                        if (cancelled) {
                            queue.clear();
                            return;
                        }
                        T t = queue.poll();
                        if (t == null) {
                            if (d) {
                                terminate();
                                return;
                            } else {
                                break;
                            }
                        } else {
                            child.onNext(t);
                            e++;
                            if (count != Long.MAX_VALUE) {
                                count--;
                            }
                        }
                        d = done;
                    }
                    if (d && queue.isEmpty()) {
                        terminate();
                        return;
                    }
                    if (e != 0 && r != Long.MAX_VALUE) {
                        r = requested.addAndGet(-e);
                    }
                    if (r != 0 && count == 0) {
                        // requests from parent have arrived so request some
                        // more
                        int min = minRequests[requestNum];
                        if (requestNum != minRequests.length - 1) {
                            requestNum++;
                        }
                        count = Math.max(r, min);
                        parent.request(count);
                    }
                    missed = addAndGet(-missed);
                    if (missed == 0) {
                        return;
                    }
                }
            }
        }

        private void terminate() {
            parent.cancel();
            Throwable err = error;
            if (err != null) {
                error = null;
                child.onError(err);
            } else {
                child.onComplete();
            }
        }
    }

} public final class FlowableRepeatingTransform<T> extends Flowable<T> {

    private final Flowable<T> source;
    private final Function<? super Flowable<T>, ? extends Flowable<T>> transform;
    private final int maxChained;
    private final long maxIterations;
    private final Function<Observable<T>, ? extends Observable<?>> tester;

    public FlowableRepeatingTransform(Flowable<T> source,
            Function<? super Flowable<T>, ? extends Flowable<T>> transform, int maxChained,
            long maxIterations, Function<Observable<T>, Observable<?>> tester) {
        Preconditions.checkArgument(maxChained > 0, "maxChained must be > 0");
        Preconditions.checkArgument(maxIterations > 0, "maxIterations must be > 0");
        Preconditions.checkNotNull(transform, "transform must not be null");
        Preconditions.checkNotNull(tester, "tester must not be null");
        this.source = source;
        this.transform = transform;
        this.maxChained = maxChained;
        this.maxIterations = maxIterations;
        this.tester = tester;
    }

    @Override
    protected void subscribeActual(Subscriber<? super T> child) {

        Flowable<T> f;
        try {
            f = transform.apply(source);
        } catch (Exception e) {
            Exceptions.throwIfFatal(e);
            child.onSubscribe(SubscriptionHelper.CANCELLED);
            child.onError(e);
            return;
        }
        AtomicReference<Chain<T>> chainRef = new AtomicReference<Chain<T>>();
        DestinationSerializedSubject<T> destination = new DestinationSerializedSubject<T>(child,
                chainRef);
        Chain<T> chain = new Chain<T>(transform, destination, maxIterations, maxChained, tester);
        chainRef.set(chain);
        // destination is not initially subscribed to the chain but will be when
        // tester function result completes
        destination.subscribe(child);
        ChainedReplaySubject<T> sub = ChainedReplaySubject.create(destination, chain, tester);
        chain.initialize(sub);
        f.onTerminateDetach() //
                .subscribe(sub);
    }

    private static enum EventType {
        TESTER_ADD, TESTER_DONE, TESTER_COMPLETE_OR_CANCEL, NEXT, ERROR, COMPLETE;
    }

    private static final class Event<T> {

        final EventType eventType;
        final ChainedReplaySubject<T> subject;
        final Subscriber<? super T> subscriber;
        final T t;
        final Throwable error;

        Event(EventType eventType, ChainedReplaySubject<T> subject,
                Subscriber<? super T> subscriber, T t, Throwable error) {
            this.eventType = eventType;
            this.subject = subject;
            this.subscriber = subscriber;
            this.t = t;
            this.error = error;
        }
    }

    @SuppressWarnings("serial")
    private static final class Chain<T> extends AtomicInteger implements Subscription {

        private final Function<? super Flowable<T>, ? extends Flowable<T>> transform;
        private final SimplePlainQueue<Event<T>> queue;
        private final DestinationSerializedSubject<T> destination;
        private final long maxIterations;
        private final int maxChained;
        private final Function<Observable<T>, ? extends Observable<?>> test;

        // state
        private int iteration = 1;
        private int length;
        private ChainedReplaySubject<T> finalSubscriber;
        private boolean destinationAttached;
        private volatile boolean cancelled;

        Chain(Function<? super Flowable<T>, ? extends Flowable<T>> transform,
                DestinationSerializedSubject<T> destination, long maxIterations, int maxChained,
                Function<Observable<T>, ? extends Observable<?>> test) {
            this.transform = transform;
            this.destination = destination;
            this.maxIterations = maxIterations;
            this.maxChained = maxChained;
            this.test = test;
            this.queue = new SpscLinkedArrayQueue<Event<T>>(16);
        }

        void initialize(ChainedReplaySubject<T> subject) {
            finalSubscriber = subject;
            if (maxIterations == 1) {
                finalSubscriber.subscribe(destination);
                destinationAttached = true;
            }
        }

        void tryAddSubscriber(ChainedReplaySubject<T> subject) {
            queue.offer(new Event<T>(EventType.TESTER_ADD, subject, null, null, null));
            drain();
        }

        void done(ChainedReplaySubject<T> subject) {
            queue.offer(new Event<T>(EventType.TESTER_DONE, subject, null, null, null));
            drain();
        }

        void completeOrCancel(ChainedReplaySubject<T> subject) {
            queue.offer(
                    new Event<T>(EventType.TESTER_COMPLETE_OR_CANCEL, subject, null, null, null));
            drain();
        }

        public void onError(Subscriber<? super T> child, Throwable err) {
            queue.offer(new Event<T>(EventType.ERROR, null, child, null, err));
            drain();

        }

        public void onCompleted(Subscriber<? super T> child) {
            queue.offer(new Event<T>(EventType.COMPLETE, null, child, null, null));
            drain();

        }

        public void onNext(Subscriber<? super T> child, T t) {
            queue.offer(new Event<T>(EventType.NEXT, null, child, t, null));
            drain();
        }

        void drain() {
            if (getAndIncrement() == 0) {
                if (cancelled) {
                    finalSubscriber.cancel();
                    queue.clear();
                    return;
                }
                int missed = 1;
                while (true) {
                    while (true) {
                        Event<T> v = queue.poll();
                        if (v == null) {
                            break;
                        } else if (v.eventType == EventType.TESTER_ADD) {
                            handleAdd(v);
                        } else if (v.eventType == EventType.TESTER_DONE) {
                            handleDone();
                        } else if (v.eventType == EventType.NEXT) {
                            v.subscriber.onNext(v.t);
                        } else if (v.eventType == EventType.COMPLETE) {
                            v.subscriber.onComplete();
                        } else if (v.eventType == EventType.ERROR) {
                            v.subscriber.onError(v.error);
                        } else {
                            handleCompleteOrCancel(v);
                        }
                    }
                    missed = addAndGet(-missed);
                    if (missed == 0) {
                        break;
                    }
                }
            }
        }

        private void handleAdd(Event<T> v) {
            debug("ADD " + v.subject);
            if (!destinationAttached && v.subject == finalSubscriber && length < maxChained
                    && !destinationAttached) {
                if (iteration <= maxIterations - 1) {
                    // ok to add another subject to the chain
                    ChainedReplaySubject<T> sub = ChainedReplaySubject.create(destination, this,
                            test);
                    if (iteration == maxIterations - 1) {
                        sub.subscribe(destination);
                        debug(sub + "subscribed to by destination");
                        destinationAttached = true;
                    }
                    addToChain(sub);
                    finalSubscriber = sub;
                    iteration++;
                    length += 1;
                }
            }
        }

        private void handleDone() {
            debug("DONE");
            if (!destinationAttached) {
                destinationAttached = true;
                finalSubscriber.subscribe(destination);
            }
        }

        private void handleCompleteOrCancel(Event<T> v) {
            debug("COMPLETE/CANCEL " + v.subject);
            if (destinationAttached) {
                return;
            }
            if (v.subject == finalSubscriber) {
                // TODO what to do here?
                // cancelWholeChain();
            } else if (iteration < maxIterations - 1) {
                ChainedReplaySubject<T> sub = ChainedReplaySubject.create(destination, this, test);
                addToChain(sub);
                finalSubscriber = sub;
                iteration++;
            } else if (iteration == maxIterations - 1) {
                ChainedReplaySubject<T> sub = ChainedReplaySubject.create(destination, this, test);
                destinationAttached = true;
                sub.subscribe(destination);
                addToChain(sub);
                debug(sub + "subscribed to by destination");
                finalSubscriber = sub;
                iteration++;
            } else {
                length--;
            }
        }

        private void addToChain(final Subscriber<T> sub) {
            Flowable<T> f;
            try {
                f = transform.apply(finalSubscriber);
            } catch (Exception e) {
                Exceptions.throwIfFatal(e);
                cancelWholeChain();
                destination.onError(e);
                return;
            }
            log("adding subscriber to " + finalSubscriber);
            f.onTerminateDetach().subscribe(sub);
            debug(finalSubscriber + " subscribed to by " + sub);
        }

        private void cancelWholeChain() {
            cancelled = true;
            drain();
        }

        @Override
        public void request(long n) {
            // ignore, just want to be able to cancel
        }

        @Override
        public void cancel() {
            cancelled = true;
            cancelWholeChain();
        }

    }

    private static class DestinationSerializedSubject<T> extends Flowable<T>
            implements FlowableSubscriber<T>, Subscription {

        private final Subscriber<? super T> child;
        private final AtomicReference<Chain<T>> chain;

        private final AtomicInteger wip = new AtomicInteger();
        private final AtomicReference<Subscription> parent = new AtomicReference<Subscription>();
        private final AtomicLong requested = new AtomicLong();
        private final SimplePlainQueue<T> queue = new SpscLinkedArrayQueue<T>(16);
        private final AtomicLong deferredRequests = new AtomicLong();

        private Throwable error;
        private volatile boolean done;
        private volatile boolean cancelled;

        DestinationSerializedSubject(Subscriber<? super T> child, AtomicReference<Chain<T>> chain) {
            this.child = child;
            this.chain = chain;
        }

        @Override
        protected void subscribeActual(Subscriber<? super T> child) {
            debug(this + " subscribed to by " + child);
            child.onSubscribe(new MultiSubscription(this, chain.get()));
            // don't need to drain because destination is always subscribed to
            // this before this is subscribed to parent
        }

        @Override
        public void onSubscribe(Subscription pr) {
            parent.set(pr);
            long r = deferredRequests.getAndSet(-1);
            if (r > 0L) {
                debug(this + " requesting of parent " + r);
                pr.request(r);
            }
            drain();
        }

        @Override
        public void request(long n) {
            debug(this + " request " + n);
            if (SubscriptionHelper.validate(n)) {
                BackpressureHelper.add(requested, n);
                while (true) {
                    Subscription p = parent.get();
                    long d = deferredRequests.get();
                    if (d == -1) {
                        // parent exists so can request of it
                        debug(this + " requesting from parent " + n);
                        p.request(n);
                        break;
                    } else {
                        long d2 = d + n;
                        if (d2 < 0) {
                            d2 = Long.MAX_VALUE;
                        }
                        if (deferredRequests.compareAndSet(d, d2)) {
                            break;
                        }
                    }
                }
                drain();
            }
        }

        @Override
        public void cancel() {
            cancelled = true;
            SubscriptionHelper.cancel(this.parent);
            chain.get().cancel();
        }

        @Override
        public void onNext(T t) {
            queue.offer(t);
            drain();
        }

        @Override
        public void onError(Throwable e) {
            error = e;
            done = true;
            drain();
        }

        @Override
        public void onComplete() {
            debug("final complete");
            done = true;
            drain();
        }

        private void drain() {
            // this is a pretty standard drain loop
            // default is to shortcut errors (don't delay them)
            if (wip.getAndIncrement() == 0) {
                int missed = 1;
                while (true) {
                    long r = requested.get();
                    long e = 0;
                    boolean d = done;
                    while (e != r) {
                        if (cancelled) {
                            queue.clear();
                            return;
                        }
                        if (d && terminate()) {
                            return;
                        }
                        T t = queue.poll();
                        if (t == null) {
                            if (d) {
                                cancel();
                                child.onComplete();
                                return;
                            } else {
                                break;
                            }
                        } else {
                            child.onNext(t);
                            e++;
                        }
                        d = done;
                    }
                    if (d && terminate()) {
                        return;
                    }
                    if (e != 0 && r != Long.MAX_VALUE) {
                        r = requested.addAndGet(-e);
                    }
                    missed = wip.addAndGet(-missed);
                    if (missed == 0) {
                        return;
                    }
                }
            }
        }

        private boolean terminate() {
            // done is true at this point
            Throwable err = error;
            if (err != null) {
                queue.clear();
                error = null;
                cancel();
                child.onError(err);
                return true;
            } else if (queue.isEmpty()) {
                cancel();
                child.onComplete();
                return true;
            } else {
                return false;
            }
        }

    }

    private static final class Tester<T> extends Observable<T> implements Observer<T> {

        private Observer<? super T> observer;

        @Override
        protected void subscribeActual(Observer<? super T> observer) {
            observer.onSubscribe(Disposables.empty());
            this.observer = observer;
        }

        @Override
        public void onSubscribe(Disposable d) {
            throw new RuntimeException("unexpected");
        }

        @Override
        public void onNext(T t) {
            observer.onNext(t);
        }

        @Override
        public void onError(Throwable e) {
            observer.onError(e);
        }

        @Override
        public void onComplete() {
            observer.onComplete();
        }
    }

    private static final class TesterObserver<T> implements Observer<Object> {

        private final Chain<T> chain;
        private final ChainedReplaySubject<T> subject;

        TesterObserver(Chain<T> chain, ChainedReplaySubject<T> subject) {
            this.chain = chain;
            this.subject = subject;
        }

        @Override
        public void onSubscribe(Disposable d) {
            // ignore
        }

        @Override
        public void onNext(Object t) {
            debug(subject + " TestObserver emits add " + t);
            chain.tryAddSubscriber(subject);
        }

        @Override
        public void onError(Throwable e) {
            chain.cancel();
            subject.destination().onError(e);
        }

        @Override
        public void onComplete() {
            debug(subject + " TestObserver emits done");
            chain.done(subject);
        }
    }

    /**
     * Requests minimally of upstream and buffers until this subscriber itself
     * is subscribed to. A maximum of {@code maxDepthConcurrent} subscribers can
     * be chained together at any one time.
     * 
     * @param <T>
     *            generic type
     */
    private static final class ChainedReplaySubject<T> extends Flowable<T>
            implements FlowableSubscriber<T>, Subscription {

        // assigned in constructor
        private final DestinationSerializedSubject<T> destination;
        private final Chain<T> chain;

        // assigned here
        private final SimplePlainQueue<T> queue = new SpscLinkedArrayQueue<T>(16);
        private final AtomicLong requested = new AtomicLong();
        private final AtomicReference<Requests<T>> requests = new AtomicReference<Requests<T>>(
                new Requests<T>(null, 0, 0, null));
        private final AtomicInteger wip = new AtomicInteger();
        private final Tester<T> tester;

        // mutable
        private volatile boolean done;
        // visibility controlled by `done`
        private Throwable error;
        private volatile boolean cancelled;
        private final Function<Observable<T>, ? extends Observable<?>> test;

        static <T> ChainedReplaySubject<T> create(DestinationSerializedSubject<T> destination,
                Chain<T> chain, Function<Observable<T>, ? extends Observable<?>> test) {
            ChainedReplaySubject<T> c = new ChainedReplaySubject<T>(destination, chain, test);
            c.init();
            return c;
        }

        private ChainedReplaySubject(DestinationSerializedSubject<T> destination, Chain<T> chain,
                Function<Observable<T>, ? extends Observable<?>> test) {
            this.destination = destination;
            this.chain = chain;
            this.test = test;
            this.tester = new Tester<T>();
        }

        private static final class Requests<T> {
            final Subscription parent;
            final long unreconciled;
            final long deferred;
            final Subscriber<? super T> child;

            Requests(Subscription parent, long unreconciled, long deferred,
                    Subscriber<? super T> child) {
                this.parent = parent;
                this.unreconciled = unreconciled;
                this.deferred = deferred;
                this.child = child;
            }
        }

        private void init() {
            Observable<?> o;
            try {
                o = test.apply(tester);
            } catch (Exception e) {
                // TODO
                throw new RuntimeException(e);
            }
            o.subscribe(new TesterObserver<T>(chain, this));
        }

        DestinationSerializedSubject<T> destination() {
            return destination;
        }

        @Override
        public void onSubscribe(Subscription parent) {
            while (true) {
                Requests<T> r = requests.get();
                Requests<T> r2;
                if (r.deferred == 0) {
                    r2 = new Requests<T>(parent, r.unreconciled + 1, 0, r.child);
                    if (requests.compareAndSet(r, r2)) {
                        parent.request(1);
                        break;
                    }
                } else {
                    r2 = new Requests<T>(parent, r.unreconciled, 0, r.child);
                    if (requests.compareAndSet(r, r2)) {
                        parent.request(r.deferred);
                        break;
                    }
                }
            }
            drain();
        }

        @Override
        protected void subscribeActual(Subscriber<? super T> child) {
            debug(this + " subscribed with " + child);
            while (true) {
                Requests<T> r = requests.get();
                Requests<T> r2 = new Requests<T>(r.parent, r.unreconciled, r.deferred, child);
                if (requests.compareAndSet(r, r2)) {
                    break;
                }
            }
            child.onSubscribe(this);
            drain();
        }

        @Override
        public void request(long n) {
            debug(this + " request " + n);
            if (SubscriptionHelper.validate(n)) {
                BackpressureHelper.add(requested, n);
                while (true) {
                    Requests<T> r = requests.get();
                    Requests<T> r2;
                    if (r.parent == null) {
                        long d = r.deferred + n;
                        if (d < 0) {
                            d = Long.MAX_VALUE;
                        }
                        r2 = new Requests<T>(r.parent, r.unreconciled, d, r.child);
                        if (requests.compareAndSet(r, r2)) {
                            break;
                        }
                    } else {
                        long x = n + r.deferred - r.unreconciled;
                        long u = Math.max(0, -x);
                        r2 = new Requests<T>(r.parent, u, 0, r.child);
                        if (requests.compareAndSet(r, r2)) {
                            if (x > 0) {
                                r.parent.request(x);
                            }
                            break;
                        }
                    }
                }
                drain();
            }
        }

        @Override
        public void onNext(T t) {
            debug(this + " arrived " + t);
            if (done) {
                return;
            }
            queue.offer(t);
            tester.onNext(t);
            while (true) {
                Requests<T> r = requests.get();
                Requests<T> r2;
                if (r.child == null) {
                    r2 = new Requests<T>(r.parent, r.unreconciled + 1, r.deferred, r.child);
                    if (requests.compareAndSet(r, r2)) {
                        // make minimal request to keep upstream producing
                        r.parent.request(1);
                        break;
                    }
                } else {
                    r2 = new Requests<T>(r.parent, r.unreconciled, 0, r.child);
                    if (requests.compareAndSet(r, r2)) {
                        break;
                    }
                }
            }
            drain();
        }

        @Override
        public void onComplete() {
            debug(this + " complete");
            if (done) {
                return;
            }
            done = true;
            cancelParent();
            debug(this + " emits complete to tester");
            tester.onComplete();
            drain();
        }

        @Override
        public void onError(Throwable t) {
            debug(this + " error " + t);
            if (done) {
                RxJavaPlugins.onError(t);
                return;
            }
            error = t;
            done = true;
            tester.onError(t);
            drain();
        }

        private void drain() {
            // this is a pretty standard drain loop
            // default is to shortcut errors (don't delay them)
            if (wip.getAndIncrement() == 0) {
                int missed = 1;
                while (true) {
                    long r = requested.get();
                    long e = 0;
                    boolean d = done;
                    while (e != r) {
                        if (cancelled) {
                            queue.clear();
                            return;
                        }
                        Subscriber<? super T> child = requests.get().child;
                        if (child == null) {
                            break;
                        }
                        Throwable err = error;
                        if (err != null) {
                            queue.clear();
                            error = null;
                            cancel();
                            chain.onError(child, err);
                            return;
                        }

                        T t = queue.poll();
                        if (t == null) {
                            if (d) {
                                cancel();
                                chain.onCompleted(child);
                                return;
                            } else {
                                break;
                            }
                        } else {
                            debug(this + " emitting " + t + " to " + requests.get().child + ":"
                                    + requests.get().child.getClass().getSimpleName());
                            chain.onNext(child, t);
                            e++;
                        }
                        d = done;
                    }
                    if (d && queue.isEmpty() && terminate()) {
                        return;
                    }
                    if (e != 0 && r != Long.MAX_VALUE) {
                        r = requested.addAndGet(-e);
                    }
                    missed = wip.addAndGet(-missed);
                    if (missed == 0) {
                        return;
                    }
                }
            }
        }

        private boolean terminate() {
            Subscriber<? super T> child = requests.get().child;
            if (child != null) {
                Throwable err = error;
                if (err != null) {
                    queue.clear();
                    error = null;
                    cancel();
                    chain.onError(child, err);
                    return true;
                } else {
                    cancel();
                    chain.onCompleted(child);
                    return true;
                }
            }
            return false;
        }

        @Override
        public void cancel() {
            if (!cancelled) {
                cancelled = true;
                cancelParentTryToAddSubscriberToChain();
            }
        }

        private void cancelParentTryToAddSubscriberToChain() {
            cancelParent();
            chain.completeOrCancel(this);
        }

        private void cancelParent() {
            Subscription par = requests.get().parent;
            if (par != null) {
                par.cancel();
            }
        }

    }

    private static final class MultiSubscription implements Subscription {

        private final Subscription primary;
        private final Subscription secondary;

        MultiSubscription(Subscription primary, Subscription secondary) {
            this.primary = primary;
            this.secondary = secondary;
        }

        @Override
        public void request(long n) {
            primary.request(n);
        }

        @Override
        public void cancel() {
            primary.cancel();
            secondary.cancel();
        }

    }

    static void debug(String message) {
        // System.out.println(message);
    }

    static void log(String message) {
        // System.out.println(message);
    }

} public final class FlowableReverse {

    private FlowableReverse() {
        // prevent instantiation
    }

    @SuppressWarnings("unchecked")
    public static <T> Flowable<T> reverse(Flowable<T> source) {
        return source.toList().toFlowable()
                .flatMapIterable((Function<List<T>, Iterable<T>>) (Function<?, ?>) REVERSE_LIST);
    }

    private static final Function<List<Object>, Iterable<Object>> REVERSE_LIST = new Function<List<Object>, Iterable<Object>>() {
        @Override
        public Iterable<Object> apply(List<Object> list) {
            return reverse(list);
        }
    };

    private static <T> Iterable<T> reverse(final List<T> list) {
        return new Iterable<T>() {

            @Override
            public Iterator<T> iterator() {
                return new Iterator<T>() {

                    int i = list.size();

                    @Override
                    public boolean hasNext() {
                        return i > 0;
                    }

                    @Override
                    public T next() {
                        i--;
                        return list.get(i);
                    }

                    @Override
                    public void remove() {
                        throw new UnsupportedOperationException();
                    }

                };
            }
        };
    }

} public final class FlowableWindowMinMax<T> extends Flowable<T> {

    private final Flowable<T> source;
    private final int windowSize;
    private final Comparator<? super T> comparator;
    private final Metric metric;

    public FlowableWindowMinMax(Flowable<T> source, int windowSize, Comparator<? super T> comparator, Metric metric) {
        Preconditions.checkArgument(windowSize > 0, "windowSize must be greater than zero");
        Preconditions.checkNotNull(comparator, "comparator cannot be null");
        Preconditions.checkNotNull(metric, "metric cannot be null");
        this.source = source;
        this.windowSize = windowSize;
        this.comparator = comparator;
        this.metric = metric;
    }

    @Override
    protected void subscribeActual(org.reactivestreams.Subscriber<? super T> child) {
        source.subscribe(new WindowMinMaxSubscriber<T>(windowSize, comparator, metric, child));
    }

    private static final class WindowMinMaxSubscriber<T> implements FlowableSubscriber<T>, Subscription {

        private final int windowSize;
        private final Comparator<? super T> comparator;
        private final Metric metric;
        private final Subscriber<? super T> child;

        // map index to value
        private final Map<Long, T> values;

        // queue of indices
        private final Deque<Long> indices;

        private long count = 0;
        private Subscription parent;

        WindowMinMaxSubscriber(int windowSize, Comparator<? super T> comparator, Metric metric,
                Subscriber<? super T> child) {
            this.windowSize = windowSize;
            this.comparator = comparator;
            this.metric = metric;
            this.child = child;
            this.values = new HashMap<Long, T>(windowSize);
            this.indices = new ArrayDeque<Long>(windowSize);
        }

        @Override
        public void onSubscribe(Subscription parent) {
            if (SubscriptionHelper.validate(this.parent, parent)) {
                this.parent = parent;
                child.onSubscribe(this);
                parent.request(windowSize - 1);
            }
        }

        @Override
        public void request(long n) {
            if (SubscriptionHelper.validate(n)) {
                parent.request(n);
            }
        }

        @Override
        public void cancel() {
            parent.cancel();
            // would be nice to clear the window here but would have performance
            // impact because would need to worry about allowing concurrent
            // changes to `indices` and `map`
        }

        @Override
        public void onComplete() {
            child.onComplete();
        }

        @Override
        public void onError(Throwable e) {
            child.onError(e);
        }

        @Override
        public void onNext(T t) {
            count++;
            // add to queue
            addToQueue(t);
            if (count >= windowSize) {
                // emit max

                // head of queue is max
                Long head = indices.peekFirst();
                final T value;
                if (head == count - windowSize) {
                    // if window past that index then remove from map
                    values.remove(indices.pollFirst());
                    value = values.get(indices.peekFirst());
                } else {
                    value = values.get(head);
                }
                child.onNext(value);
            }
        }

        private void addToQueue(T t) {
            Long v;
            while ((v = indices.peekLast()) != null && compare(t, values.get(v)) <= 0) {
                values.remove(indices.pollLast());
            }
            values.put(count, t);
            indices.offerLast(count);
        }

        private int compare(T a, T b) {
            if (metric == Metric.MIN) {
                return comparator.compare(a, b);
            } else {
                return comparator.compare(b, a);
            }
        }
    }

    public enum Metric {
        MIN, MAX;
    }

} public enum Metric {
    MIN, MAX;
} public final class TransformerStateMachine<State, In, Out> implements FlowableTransformer<In, Out> {

    private final Callable<? extends State> initialState;
    private final Function3<? super State, ? super In, ? super FlowableEmitter<Out>, ? extends State> transition;
    private final BiPredicate<? super State, ? super FlowableEmitter<Out>> completion;
    private final BackpressureStrategy backpressureStrategy;
    private final int requestBatchSize;

    private TransformerStateMachine(Callable<? extends State> initialState,
            Function3<? super State, ? super In, ? super FlowableEmitter<Out>, ? extends State> transition,
            BiPredicate<? super State, ? super FlowableEmitter<Out>> completion,
            BackpressureStrategy backpressureStrategy, int requestBatchSize) {
        Preconditions.checkNotNull(initialState);
        Preconditions.checkNotNull(transition);
        Preconditions.checkNotNull(completion);
        Preconditions.checkNotNull(backpressureStrategy);
        Preconditions.checkArgument(requestBatchSize > 0, "initialRequest must be greater than zero");
        this.initialState = initialState;
        this.transition = transition;
        this.completion = completion;
        this.backpressureStrategy = backpressureStrategy;
        this.requestBatchSize = requestBatchSize;
    }

    public static <State, In, Out> FlowableTransformer<In, Out> create(Callable<? extends State> initialState,
            Function3<? super State, ? super In, ? super FlowableEmitter<Out>, ? extends State> transition,
            BiPredicate<? super State, ? super FlowableEmitter<Out>> completion,
            BackpressureStrategy backpressureStrategy, int requestBatchSize) {
        return new TransformerStateMachine<State, In, Out>(initialState, transition, completion, backpressureStrategy,
                requestBatchSize);
    }

    @Override
    public Flowable<Out> apply(final Flowable<In> source) {
        // use defer so we can have a single state reference for each
        // subscription
        return Flowable.defer(new Callable<Flowable<Out>>() {
            @Override
            public Flowable<Out> call() throws Exception {
                Mutable<State> state = new Mutable<State>(initialState.call());
                return source.materialize()
                        // do state transitions and emit notifications
                        // use flatMap to emit notification values
                        .flatMap(execute(transition, completion, state, backpressureStrategy), requestBatchSize)
                        // complete if we encounter an unsubscribed sentinel
                        .takeWhile(NOT_UNSUBSCRIBED)
                        // flatten notifications to a stream which will enable
                        // early termination from the state machine if desired
                        .dematerialize();
            }
        });
    }

    private static <State, Out, In> Function<Notification<In>, Flowable<Notification<Out>>> execute(
            final Function3<? super State, ? super In, ? super FlowableEmitter<Out>, ? extends State> transition,
            final BiPredicate<? super State, ? super FlowableEmitter<Out>> completion, final Mutable<State> state,
            final BackpressureStrategy backpressureStrategy) {

        return new Function<Notification<In>, Flowable<Notification<Out>>>() {

            @Override
            public Flowable<Notification<Out>> apply(final Notification<In> in) {

                return Flowable.create(new FlowableOnSubscribe<Notification<Out>>() {

                    @Override
                    public void subscribe(FlowableEmitter<Notification<Out>> emitter) throws Exception {
                        FlowableEmitter<Out> w = wrap(emitter);
                        if (in.isOnNext()) {
                            state.value = transition.apply(state.value, in.getValue(), w);
                            if (!emitter.isCancelled())
                                emitter.onComplete();
                            else {
                                // this is a special emission to indicate that
                                // the transition called unsubscribe. It will be
                                // filtered later.
                                emitter.onNext(UnsubscribedNotificationHolder.<Out>unsubscribedNotification());
                            }
                        } else if (in.isOnComplete()) {
                            if (completion.test(state.value, w) && !emitter.isCancelled()) {
                                w.onComplete();
                            }
                        } else if (!emitter.isCancelled()) {
                            w.onError(in.getError());
                        }
                    }

                }, backpressureStrategy);
            }
        };
    }

    private static final class UnsubscribedNotificationHolder {
        private static final Notification<Object> INSTANCE = Notification.createOnNext(new Object());

        @SuppressWarnings("unchecked")
        static <T> Notification<T> unsubscribedNotification() {
            return (Notification<T>) INSTANCE;
        }
    }

    private static final Predicate<Notification<?>> NOT_UNSUBSCRIBED = new Predicate<Notification<?>>() {

        @Override
        public boolean test(Notification<?> t) {
            return t != UnsubscribedNotificationHolder.unsubscribedNotification();
        }

    };

    private static final class Mutable<T> {
        T value;

        Mutable(T value) {
            this.value = value;
        }
    }

    private static <Out> NotificationEmitter<Out> wrap(FlowableEmitter<? super Notification<Out>> emitter) {
        return new NotificationEmitter<Out>(emitter);
    }

    private static final class NotificationEmitter<Out> implements FlowableEmitter<Out> {

        private final FlowableEmitter<? super Notification<Out>> emitter;

        NotificationEmitter(FlowableEmitter<? super Notification<Out>> emitter) {
            this.emitter = emitter;
        }

        @Override
        public void onComplete() {
            emitter.onNext(Notification.<Out>createOnComplete());
        }

        @Override
        public void onError(Throwable e) {
            if (!tryOnError(e)) {
                RxJavaPlugins.onError(e);
            }
        }

        @Override
        public void onNext(Out t) {
            emitter.onNext(Notification.createOnNext(t));
        }

        @Override
        public void setDisposable(Disposable s) {
            throw new UnsupportedOperationException();
        }

        @Override
        public void setCancellable(Cancellable c) {
            throw new UnsupportedOperationException();
        }

        @Override
        public long requested() {
            return emitter.requested();
        }

        @Override
        public boolean isCancelled() {
            return emitter.isCancelled();

        }

        @Override
        public FlowableEmitter<Out> serialize() {
            throw new UnsupportedOperationException();
        }

        @Override
        public boolean tryOnError(Throwable e) {
            if (emitter.isCancelled()) {
                return false;
            } else {
                emitter.onNext(Notification.<Out>createOnError(e));
                return true;
            }
        }

    }

} public class Pair<T, S> {

    private final T a;
    private final S b;

    public Pair(T a, S b) {
        this.a = a;
        this.b = b;
    }

    public static <T, S> Pair<T, S> create(T t, S s) {
        return new Pair<T, S>(t, s);
    }

    public T a() {
        return a;
    }

    public S b() {
        return b;
    }

    public T left() {
        return a;
    }

    public S right() {
        return b;
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + ((a == null) ? 0 : a.hashCode());
        result = prime * result + ((b == null) ? 0 : b.hashCode());
        return result;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj)
            return true;
        if (obj == null)
            return false;
        if (getClass() != obj.getClass())
            return false;
        Pair<?, ?> other = (Pair<?, ?>) obj;
        if (a == null) {
            if (other.a != null)
                return false;
        } else if (!a.equals(other.a))
            return false;
        if (b == null) {
            if (other.b != null)
                return false;
        } else if (!b.equals(other.b))
            return false;
        return true;
    }

    @Override
    public String toString() {
        StringBuilder builder = new StringBuilder();
        builder.append("Pair [left=");
        builder.append(a);
        builder.append(", right=");
        builder.append(b);
        builder.append("]");
        return builder.toString();
    }

    public T _1() {
        return a;
    }

    public S _2() {
        return b;
    }
}
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import org.reactivestreams.Publisher;
import com.github.davidmoten.guavamini.Preconditions;
import com.github.davidmoten.rx2.BiFunctions;
import com.github.davidmoten.rx2.Flowables;
import com.github.davidmoten.rx2.Functions;
import com.github.davidmoten.rx2.StateMachine;
import com.github.davidmoten.rx2.StateMachine2;
import com.github.davidmoten.rx2.Statistics;
import com.github.davidmoten.rx2.buffertofile.Options;
import com.github.davidmoten.rx2.internal.flowable.FlowableCollectWhile;
import com.github.davidmoten.rx2.internal.flowable.FlowableDoOnEmpty;
import com.github.davidmoten.rx2.internal.flowable.FlowableInsertMaybe;
import com.github.davidmoten.rx2.internal.flowable.FlowableInsertTimeout;
import com.github.davidmoten.rx2.internal.flowable.FlowableMapLast;
import com.github.davidmoten.rx2.internal.flowable.FlowableMatch;
import com.github.davidmoten.rx2.internal.flowable.FlowableMaxRequest;
import com.github.davidmoten.rx2.internal.flowable.FlowableMinRequest;
import com.github.davidmoten.rx2.internal.flowable.FlowableRepeatingTransform;
import com.github.davidmoten.rx2.internal.flowable.FlowableReverse;
import com.github.davidmoten.rx2.internal.flowable.FlowableWindowMinMax;
import com.github.davidmoten.rx2.internal.flowable.FlowableWindowMinMax.Metric;
import com.github.davidmoten.rx2.internal.flowable.TransformerStateMachine;
import com.github.davidmoten.rx2.util.Pair;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableTransformer;
import io.reactivex.Maybe;
import io.reactivex.Notification;
import io.reactivex.Observable;
import io.reactivex.Scheduler;
import io.reactivex.functions.Action;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.BiPredicate;
import io.reactivex.functions.Function;
import io.reactivex.functions.Function3;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;
package com.github.davidmoten.rx2.flowable;





public final class Transformers {

    private Transformers() {
        // prevent instantiation
    }

    public static <State, In, Out> FlowableTransformer<In, Out> stateMachine(Callable<? extends State> initialState,
            Function3<? super State, ? super In, ? super FlowableEmitter<Out>, ? extends State> transition,
            BiPredicate<? super State, ? super FlowableEmitter<Out>> completion,
            BackpressureStrategy backpressureStrategy, int requestBatchSize) {
        return TransformerStateMachine.create(initialState, transition, completion, backpressureStrategy,
                requestBatchSize);
    }

    public static StateMachine.Builder stateMachine() {
        return StateMachine.builder();
    }

    public static StateMachine2.Builder stateMachine2() {