/*
 * Copyright 2016-present David Karnok
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package hu.akarnokd.rxjava3.basetypes;

import java.util.Objects;
import java.util.concurrent.*;

import org.reactivestreams.*;

import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.exceptions.Exceptions;
import io.reactivex.rxjava3.functions.*;
import io.reactivex.rxjava3.internal.functions.Functions;
import io.reactivex.rxjava3.internal.subscribers.LambdaSubscriber;
import io.reactivex.rxjava3.internal.util.ExceptionHelper;
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
import io.reactivex.rxjava3.schedulers.Schedulers;
import io.reactivex.rxjava3.subscribers.TestSubscriber;

/**
 * A 0-1-error base reactive type, similar to Maybe, implementing the Reactive-Streams Publisher
 * interface.
 *
 * @param <T> the value type
 * 
 * @since 0.14.0
 */
public abstract class Perhaps<T> implements Publisher<T> {

    /**
     * Hook called when assembling Perhaps sequences.
     */
    @SuppressWarnings("rawtypes")
    private static volatile Function<Perhaps, Perhaps> onAssembly;

    /**
     * Returns the current onAssembly handler.
     * @param <T> the target value type
     * @return the current handler, maybe null
     */
    @SuppressWarnings({ "unchecked", "rawtypes" })
    public static <T> Function<Perhaps<T>, Perhaps<T>> getOnAssemblyHandler() {
        return (Function)onAssembly;
    }

    /**
     * Set the onAssembly handler.
     * @param <T> the target value type
     * @param handler the handler, null clears the handler
     */
    @SuppressWarnings({ "unchecked", "rawtypes" })
    public static <T> void setOnAssemblyHandler(Function<Perhaps<T>, Perhaps<T>> handler) {
        onAssembly = (Function)handler;
    }

    @SuppressWarnings({ "rawtypes", "unchecked" })
    protected static <T> Perhaps<T> onAssembly(Perhaps<T> source) {
        Function<Perhaps, Perhaps> f = onAssembly;
        if (f == null) {
            return source;
        }
        try {
            return f.apply(source);
        } catch (Throwable ex) {
            throw ExceptionHelper.wrapOrThrow(ex);
        }
    }

    // ----------------------------------------------------
    // Factory methods (enter)
    // ----------------------------------------------------

    /**
     * Create a Perhaps that for each incoming Subscriber calls a callback to
     * emit a sync or async events in a thread-safe, backpressure-aware and
     * cancellation-safe manner.
     * @param <T> the value type emitted
     * @param onCreate the callback called for each individual subscriber with an
     * abstraction of the incoming Subscriber.
     * @return th new Perhaps instance
     */
    public static <T> Perhaps<T> create(MaybeOnSubscribe<T> onCreate) {
        Objects.requireNonNull(onCreate, "onCreate is null");
        return onAssembly(new PerhapsCreate<>(onCreate));
    }

    /**
     * Returns a Perhaps that signals the given item.
     * @param <T> the value type
     * @param item the item to signal, not null
     * @return the new Perhaps instance
     */
    public static <T> Perhaps<T> just(T item) {
        Objects.requireNonNull(item, "item is null");
        return onAssembly(new PerhapsJust<>(item));
    }

    /**
     * Returns an empty Perhaps.
     * @param <T> the value type
     * @return the shared Perhaps instance
     */
    public static <T> Perhaps<T> empty() {
        return onAssembly(PerhapsEmpty.<T>instance());
    }

    /**
     * Returns a Perhaps that signals the given error to Subscribers.
     * @param <T> the value type
     * @param error the error to signal, not null
     * @return the new Perhaps instance
     */
    public static <T> Perhaps<T> error(Throwable error) {
        Objects.requireNonNull(error, "error is null");
        return onAssembly(new PerhapsError<T>(error));
    }

    /**
     * Returns a Perhaps that signals the error returned from
     * the errorSupplier to each individual Subscriber.
     * @param <T> the value type
     * @param errorSupplier the supplier called for each Subscriber to
     * return a Throwable to be signalled
     * @return the new Perhaps instance
     */
    public static <T> Perhaps<T> error(Supplier<? extends Throwable> errorSupplier) {
        Objects.requireNonNull(errorSupplier, "errorSupplier is null");
        return onAssembly(new PerhapsErrorSupplier<T>(errorSupplier));
    }

    /**
     * Returns a Perhaps that never signals any item or terminal event.
     * @param <T> the value type
     * @return the shared Perhaps instance
     */
    public static <T> Perhaps<T> never() {
        return onAssembly(PerhapsNever.<T>instance());
    }

    /**
     * Runs a Callable and emits its resulting value or its
     * exception; null is considered to be an indication for an empty Perhaps.
     * @param <T> the value type
     * @param callable the callable to call for each individual Subscriber
     * @return the new Perhaps instance
     */
    public static <T> Perhaps<T> fromCallable(Callable<T> callable) {
        Objects.requireNonNull(callable, "callable is null");
        return onAssembly(new PerhapsFromCallable<>(callable));
    }

    /**
     * Run an action for each individual Subscriber and terminate.
     * @param <T> the value type
     * @param action the action to call for each individual Subscriber
     * @return the new Perhaps instance
     */
    public static <T> Perhaps<T> fromAction(Action action) {
        Objects.requireNonNull(action, "action is null");
        return onAssembly(new PerhapsFromAction<T>(action));
    }

    /**
     * When subscribed, the future is awaited blockingly and
     * indefinitely for its result value; null result
     * will yield a NoSuchElementException.
     * @param <T> the value type
     * @param future the future to await
     * @return the new Perhaps instance
     */
    public static <T> Perhaps<T> fromFuture(Future<? extends T> future) {
        Objects.requireNonNull(future, "future is null");
        return onAssembly(new PerhapsFromFuture<T>(future, 0, null));
    }

    /**
     * When subscribed, the future is awaited blockingly for
     * a given amount of time for its result value; null result
     * will yield a NoSuchElementException and a timeout
     * will yield a TimeoutException.
     * @param <T> the value type
     * @param future the future to await
     * @param timeout the timeout value
     * @param unit the time unit
     * @return the new Perhaps instance
     */
    public static <T> Perhaps<T> fromFuture(Future<? extends T> future, long timeout, TimeUnit unit) {
        Objects.requireNonNull(future, "future is null");
        Objects.requireNonNull(unit, "unit is null");
        return onAssembly(new PerhapsFromFuture<T>(future, timeout, unit));
    }

    /**
     * Wraps a Publisher and signals its single value or completion signal or
     * signals IndexOutOfBoundsException if the Publisher has more than one element.
     * @param <T> the value type
     * @param source the source Publisher
     * @return the new Perhaps instance
     */
    public static <T> Perhaps<T> fromPublisher(Publisher<T> source) {
        Objects.requireNonNull(source, "source is null");
        return onAssembly(new PerhapsFromPublisher<>(source));
    }

    /**
     * Wraps a Single and signals its events.
     * @param <T> the value type
     * @param source the source
     * @return the new Perhaps instance
     */
    public static <T> Perhaps<T> fromSingle(SingleSource<T> source) {
        Objects.requireNonNull(source, "source is null");
        return onAssembly(new PerhapsFromSingle<>(source));
    }

    /**
     * Wraps a Maybe and signals its events.
     * @param <T> the value type
     * @param source the source
     * @return the new Perhaps instance
     */
    public static <T> Perhaps<T> fromMaybe(MaybeSource<T> source) {
        Objects.requireNonNull(source, "source is null");
        return onAssembly(new PerhapsFromMaybe<>(source));
    }

    /**
     * Wraps a Completable and signals its terminal events.
     * @param <T> the value type
     * @param source the source
     * @return the new Perhaps instance
     */
    public static <T> Perhaps<T> fromCompletable(CompletableSource source) {
        Objects.requireNonNull(source, "source is null");
        return onAssembly(new PerhapsFromCompletable<T>(source));
    }

    /**
     * Defers the creation of the actual Perhaps instance until
     * subscription time and for each downstream Subscriber the given
     * Supplier is called.
     * @param <T> the value type
     * @param supplier the Supplier called for each individual Subscriber
     * to return a Perhaps to be subscribe to.
     * @return the new Perhaps instance
     */
    public static <T> Perhaps<T> defer(Supplier<? extends Perhaps<? extends T>> supplier) {
        Objects.requireNonNull(supplier, "supplier is null");
        return onAssembly(new PerhapsDefer<>(supplier));
    }

    /**
     * Emit the events of the Perhaps that reacts first.
     * @param <T> the common value type
     * @param sources the Iterable sequence of Perhaps sources
     * @return the new Perhaps instance
     */
    public static <T> Perhaps<T> amb(Iterable<? extends Perhaps<? extends T>> sources) {
        Objects.requireNonNull(sources, "sources is null");
        return onAssembly(new PerhapsAmbIterable<>(sources));
    }

    /**
     * Emit the events of the Perhaps that reacts first.
     * @param <T> the common value type
     * @param sources the array of Perhaps sources
     * @return the new Perhaps instance
     */
    @SafeVarargs
    public static <T> Perhaps<T> ambArray(Perhaps<? extends T>... sources) {
        Objects.requireNonNull(sources, "sources is null");
        return onAssembly(new PerhapsAmbArray<>(sources));
    }

    /**
     * Concatenate the values in order from a sequence of Perhaps sources.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> concat(Iterable<? extends Perhaps<? extends T>> sources) {
        return Flowable.concat(sources);
    }

    /**
     * Concatenate the values in order from a sequence of Perhaps sources.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> concat(Publisher<? extends Perhaps<? extends T>> sources) {
        return Flowable.concat(sources);
    }

    /**
     * Concatenate the values in order from a sequence of Perhaps sources.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @param prefetch the number of sources to prefetch from upstream
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> concat(Publisher<? extends Perhaps<? extends T>> sources, int prefetch) {
        return Flowable.concat(sources, prefetch);
    }

    /**
     * Concatenate the values in order from a sequence of Perhaps sources.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @return the new Flowable instance
     */
    @SafeVarargs
    public static <T> Flowable<T> concatArray(Perhaps<? extends T>... sources) {
        return Flowable.concatArray(sources);
    }

    /**
     * Concatenate the values in order from a sequence of Perhaps sources, delaying
     * errors till all sources terminate.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> concatDelayError(Iterable<? extends Perhaps<? extends T>> sources) {
        return Flowable.concatDelayError(sources);
    }

    /**
     * Concatenate the values in order from a sequence of Perhaps sources, delaying
     * errors till all sources terminate.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> concatDelayError(Publisher<? extends Perhaps<? extends T>> sources) {
        return Flowable.concatDelayError(sources);
    }

    /**
     * Concatenate the values in order from a sequence of Perhaps sources, delaying
     * errors till all sources terminate.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @param prefetch the number of sources to prefetch from upstream
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> concatDelayError(Publisher<? extends Perhaps<? extends T>> sources, int prefetch) {
        return Flowable.concatDelayError(sources, prefetch, false);
    }

    /**
     * Concatenate the values in order from a sequence of Perhaps sources, delaying
     * errors till a source terminates or the whole sequence terminates.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @param prefetch the number of sources to prefetch from upstream
     * @param tillTheEnd if true, errors are delayed to the very end;
     * if false, an error will be signalled at the end of one source
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> concatDelayError(Publisher<? extends Perhaps<? extends T>> sources, int prefetch, boolean tillTheEnd) {
        return Flowable.concatDelayError(sources, prefetch, tillTheEnd);
    }

    /**
     * Concatenate the values in order from a sequence of Perhaps sources, delaying
     * errors till all sources terminate.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @return the new Flowable instance
     */
    @SafeVarargs
    public static <T> Flowable<T> concatArrayDelayError(Perhaps<? extends T>... sources) {
        return Flowable.concatArrayDelayError(sources);
    }

    /**
     * Merge the values in arbitrary order from a sequence of Perhaps sources.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> merge(Iterable<? extends Perhaps<? extends T>> sources) {
        return Flowable.merge(sources);
    }

    /**
     * Merge the values in arbitrary order from a sequence of Perhaps sources.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @param maxConcurrency the maximum number of active subscriptions
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> merge(Iterable<? extends Perhaps<? extends T>> sources, int maxConcurrency) {
        return Flowable.merge(sources, maxConcurrency);
    }

    /**
     * Merge the values in arbitrary order from a sequence of Perhaps sources.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> merge(Publisher<? extends Perhaps<? extends T>> sources) {
        return Flowable.merge(sources);
    }

    /**
     * Merge the values in arbitrary order from a sequence of Perhaps sources.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @param maxConcurrency the maximum number of active subscriptions
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> merge(Publisher<? extends Perhaps<? extends T>> sources, int maxConcurrency) {
        return Flowable.merge(sources, maxConcurrency);
    }

    /**
     * Merge the values in arbitrary order from a sequence of Perhaps sources.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @return the new Flowable instance
     */
    @SafeVarargs
    public static <T> Flowable<T> mergeArray(Perhaps<? extends T>... sources) {
        return Flowable.mergeArray(sources);
    }

    /**
     * Merge the values in arbitrary order from a sequence of Perhaps sources.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @param maxConcurrency the maximum number of active subscriptions
     * @return the new Flowable instance
     */
    @SafeVarargs
    public static <T> Flowable<T> mergeArray(int maxConcurrency, Perhaps<? extends T>... sources) {
        return Flowable.mergeArray(maxConcurrency, 1, sources);
    }

    /**
     * Merge the values in arbitrary order from a sequence of Perhaps sources,
     * delaying errors till all sources terminate.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> mergeDelayError(Iterable<? extends Perhaps<? extends T>> sources) {
        return Flowable.mergeDelayError(sources);
    }

    /**
     * Merge the values in arbitrary order from a sequence of Perhaps sources,
     * delaying errors till all sources terminate.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @param maxConcurrency the maximum number of active subscriptions
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> mergeDelayError(Iterable<? extends Perhaps<? extends T>> sources, int maxConcurrency) {
        return Flowable.mergeDelayError(sources, maxConcurrency);
    }

    /**
     * Merge the values in arbitrary order from a sequence of Perhaps sources,
     * delaying errors till all sources terminate.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> mergeDelayError(Publisher<? extends Perhaps<? extends T>> sources) {
        return Flowable.mergeDelayError(sources);
    }

    /**
     * Merge the values in arbitrary order from a sequence of Perhaps sources,
     * delaying errors till all sources terminate.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @param maxConcurrency the maximum number of active subscriptions
     * @return the new Flowable instance
     */
    public static <T> Flowable<T> mergeDelayError(Publisher<? extends Perhaps<? extends T>> sources, int maxConcurrency) {
        return Flowable.mergeDelayError(sources, maxConcurrency);
    }

    /**
     * Merge the values in arbitrary order from a sequence of Perhaps sources,
     * delaying errors till all sources terminate.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @return the new Flowable instance
     */
    @SafeVarargs
    public static <T> Flowable<T> mergeArrayDelayError(Perhaps<? extends T>... sources) {
        return Flowable.mergeArrayDelayError(sources);
    }

    /**
     * Merge the values in arbitrary order from a sequence of Perhaps sources,
     * delaying errors till all sources terminate.
     * @param <T> the common base value type
     * @param sources the sequence of sources
     * @param maxConcurrency the maximum number of active subscriptions
     * @return the new Flowable instance
     */
    @SafeVarargs
    public static <T> Flowable<T> mergeArrayDelayError(int maxConcurrency, Perhaps<? extends T>... sources) {
        return Flowable.mergeArrayDelayError(maxConcurrency, 1, sources);
    }

    /**
     * Signals a 0L after the specified amount of time has passed since
     * subscription.
     * @param delay the delay time
     * @param unit the time unit
     * @return the new Perhaps instance
     */
    public static Perhaps<Long> timer(long delay, TimeUnit unit) {
        return timer(delay, unit, Schedulers.computation());
    }

    /**
     * Signals a 0L after the specified amount of time has passed since
     * subscription on the specified scheduler.
     * @param delay the delay time
     * @param unit the time unit
     * @param scheduler the scheduler to wait on
     * @return the new Perhaps instance
     */
    public static Perhaps<Long> timer(long delay, TimeUnit unit, Scheduler scheduler) {
        Objects.requireNonNull(unit, "unit is null");
        Objects.requireNonNull(scheduler, "scheduler is null");
        return onAssembly(new PerhapsTimer(delay, unit, scheduler));
    }

    /**
     * Generate a resource and a Perhaps based on that resource and then
     * dispose that resource eagerly when the Perhaps terminates or the
     * downstream cancels the sequence.
     * @param <T> the value type
     * @param <R> the resource type
     * @param resourceSupplier the callback to get a resource for each subscriber
     * @param sourceSupplier the function that returns a Perhaps for the generated resource
     * @param disposer the consumer of the resource once the upstream terminates or the
     * downstream cancels
     * @return the new Perhaps instance
     */
    public static <T, R> Perhaps<T> using(Supplier<R> resourceSupplier,
            Function<? super R, ? extends Perhaps<? extends T>> sourceSupplier,
            Consumer<? super R> disposer) {
        return using(resourceSupplier, sourceSupplier, disposer, true);
    }

    /**
     * Generate a resource and a Perhaps based on that resource and then
     * dispose that resource eagerly when the Perhaps terminates or the
     * downstream cancels the sequence.
     * @param <T> the value type
     * @param <R> the resource type
     * @param resourceSupplier the callback to get a resource for each subscriber
     * @param sourceSupplier the function that returns a Perhaps for the generated resource
     * @param disposer the consumer of the resource once the upstream terminates or the
     * downstream cancels
     * @param eager if true, the resource is disposed before the terminal event is emitted
     *              if false, the resource is disposed after the terminal event has been emitted
     * @return the new Perhaps instance
     */
    public static <T, R> Perhaps<T> using(Supplier<R> resourceSupplier,
            Function<? super R, ? extends Perhaps<? extends T>> sourceSupplier,
            Consumer<? super R> disposer, boolean eager) {
        Objects.requireNonNull(resourceSupplier, "resourceSupplier is null");
        Objects.requireNonNull(sourceSupplier, "sourceSupplier is null");
        Objects.requireNonNull(disposer, "disposer is null");
        return onAssembly(new PerhapsUsing<>(resourceSupplier, sourceSupplier, disposer, eager));
    }

    /**
     * Combines the Perhaps values of all the sources via a zipper function into a
     * single resulting value.
     * @param <T> the common input base type
     * @param <R> the result type
     * @param sources the sequence of Perhaps sources
     * @param zipper the function takin in an array of values and returns a Perhaps value
     * @return the new Perhaps instance
     */
    public static <T, R> Perhaps<R> zip(Iterable<? extends Perhaps<? extends T>> sources, Function<? super Object[], ? extends R> zipper) {
        Objects.requireNonNull(sources, "sources is null");
        Objects.requireNonNull(zipper, "zipper is null");
        return onAssembly(new PerhapsZipIterable<T, R>(sources, zipper));
    }

    /**
     * Combines the Perhaps values of all the sources via a zipper function into a
     * single resulting value.
     * @param <T> the common input base type
     * @param <R> the result type
     * @param sources the sequence of Perhaps sources
     * @param zipper the function takin in an array of values and returns a Perhaps value
     * @return the new Perhaps instance
     */
    @SafeVarargs
    public static <T, R> Perhaps<R> zipArray(Function<? super Object[], ? extends R> zipper, Perhaps<? extends T>... sources) {
        Objects.requireNonNull(sources, "sources is null");
        Objects.requireNonNull(zipper, "zipper is null");
        return onAssembly(new PerhapsZipArray<T, R>(sources, zipper));
    }

    // ----------------------------------------------------
    // Operators (stay)
    // ----------------------------------------------------

    /**
     * Signals the events of this or the other Perhaps whichever
     * signals first.
     * @param other the other Perhaps instance
     * @return the new Perhaps instance
     */
    public final Perhaps<T> ambWith(Perhaps<? extends T> other) {
        return ambArray(this, other);
    }

    /**
     * Runs this Perhaps and then runs the other Nono source, only
     * emitting this Perhaps' success value if the other Nono source
     * completes normally.
     * @param other the other Nono source to run after this
     * @return the new Perhaps instance
     */
    public final Perhaps<T> andThen(Nono other) {
        Objects.requireNonNull(other, "other is null");
        return onAssembly(new PerhapsAndThenNono<>(this, other));
    }

    /**
     * Runs this Perhaps and emits its value followed by running
     * the other Publisher and emitting its values.
     * @param other the other Publisher to run after this
     * @return the new Flowable instance
     */
    public final Flowable<T> andThen(Publisher<? extends T> other) {
        Objects.requireNonNull(other, "other is null");
        return Flowable.concat(this, other);
    }

    /**
     * Runs this Perhaps and emits its value followed by running
     * the other Publisher and emitting its values.
     * @param other the other Publisher to run after this
     * @return the new Flowable instance
     */
    public final Flowable<T> concatWith(Publisher<? extends T> other) {
        Objects.requireNonNull(other, "other is null");
        return Flowable.concat(this, other);
    }

    /**
     * Merges this Perhaps with another Publisher and emits all their
     * values.
     * @param other the other Publisher source instance
     * @return the new Flowable instance
     */
    public final Flowable<T> mergeWith(Publisher<? extends T> other) {
        Objects.requireNonNull(other, "other is null");
        return Flowable.merge(this, other);
    }

    /**
     * Zips the value of this Perhaps with the other Perhaps through
     * a BiFunction.
     * @param <U> the value type of the other source
     * @param <R> the result type
     * @param other the other Perhaps source
     * @param zipper the function receiving each source value and should
     * return a value to be emitted
     * @return the new Perhaps instance
     */
    public final <U, R> Perhaps<R> zipWith(Perhaps<? extends U> other, BiFunction<? super T, ? super U, ? extends R> zipper) {
        Objects.requireNonNull(other, "other is null");
        Objects.requireNonNull(zipper, "zipper is null");
        return zipArray(Functions.toFunction(zipper), this, other);
    }

    /**
     * Maps the value of this Perhaps into another value (of possibly different
     * type).
     * @param <R> the result value type
     * @param mapper the function that receives the onNext value from this Perhaps
     * and returns another value
     * @return the new Perhaps instance
     */
    public final <R> Perhaps<R> map(Function<? super T, ? extends R> mapper) {
        Objects.requireNonNull(mapper, "mapper is null");
        return onAssembly(new PerhapsMap<T, R>(this, mapper));
    }

    /**
     * Maps the error from upstream into another Throwable error.
     * @param errorMapper the function that receives the upstream error and
     * returns a Throwable
     * @return the new Perhaps instance
     */
    public final Perhaps<T> mapError(Function<? super Throwable, ? extends Throwable> errorMapper) {
        Objects.requireNonNull(errorMapper, "errorMapper is null");
        return onAssembly(new PerhapsMapError<>(this, errorMapper));
    }

    /**
     * Filters the value from upstream with a predicate and completes
     * if the filter didn't match it.
     * @param predicate the predicate receiving the upstream value and
     * returns true if it should be passed along.
     * @return the new Perhaps instance
     */
    public final Perhaps<T> filter(Predicate<? super T> predicate) {
        Objects.requireNonNull(predicate, "predicate is null");
        return onAssembly(new PerhapsFilter<>(this, predicate));
    }

    /**
     * Ignore the element of this Perhaps.
     * @return the new Nono instance.
     */
    public final Nono ignoreElement() {
        return Nono.fromPublisher(this);
    }

    /**
     * Hides the identity of this Perhaps instance, including
     * its subscriber.
     * <p>This allows preventing cerain optimizations as well
     * for diagnostic purposes.
     * @return the new Perhaps instance
     */
    public final Perhaps<T> hide() {
        return onAssembly(new PerhapsHide<>(this));
    }

    /**
     * Maps the upstream's value into another Perhaps and emits its
     * resulting events.
     * @param <R> the output value type
     * @param mapper the function that receives the upstream's value
     * and returns a Perhaps to be consumed.
     * @return the new Perhaps instance
     */
    public final <R> Perhaps<R> flatMap(Function<? super T, ? extends Perhaps<? extends R>> mapper) {
        Objects.requireNonNull(mapper, "mapper is null");
        return onAssembly(new PerhapsFlatMap<>(this, mapper));
    }

    /**
     * Maps the upstream events into other Perhaps instances and emits
     * its resulting events.
     * <p>Note that only one of the onXXXMapper is called based on what
     * the upstream signals, i.e., the usual onNext + onComplete will pick the
     * Perhaps of the onSuccessMapper only and never the onCompleteMapper.
     * @param <R> the result value type
     * @param onSuccessMapper the function called for the upstream value
     * @param onErrorMapper the function called for the upstream error
     * @param onCompleteMapper the function called when the upstream is empty
     * @return the new Perhaps instance
     */
    public final <R> Perhaps<R> flatMap(Function<? super T, ? extends Perhaps<? extends R>> onSuccessMapper,
            Function<? super Throwable, ? extends Perhaps<? extends R>> onErrorMapper,
                    Supplier<? extends Perhaps<? extends R>> onCompleteMapper) {
        Objects.requireNonNull(onSuccessMapper, "onSuccessMapper is null");
        Objects.requireNonNull(onErrorMapper, "onErrorMapper is null");
        Objects.requireNonNull(onCompleteMapper, "onCompleteMapper is null");
        return onAssembly(new PerhapsFlatMapSignal<>(this, onSuccessMapper, onErrorMapper, onCompleteMapper));
    }

    /**
     * Maps the upstream value into a Publisher and emits all of its events.
     * @param <R> the result value type
     * @param mapper the function that maps the success value into a Publisher that
     * gets subscribed to and streamed further
     * @return the new Flowable type
     */
    public final <R> Flowable<R> flatMapPublisher(Function<? super T, ? extends Publisher<? extends R>> mapper) {
        Objects.requireNonNull(mapper, "mapper is null");
        return RxJavaPlugins.onAssembly(new PerhapsFlatMapPublisher<>(this, mapper));
    }

    /**
     * Completes in case the upstream signals an error.
     * @return the new Perhaps instance
     */
    public final Perhaps<T> onErrorComplete() {
        return onAssembly(new PerhapsOnErrorReturnItem<>(this, null));
    }

    /**
     * If the upstream signals an error, it is replaced by a signal
     * of the given item and normal completion.
     * @param item the item to signal in case of an error
     * @return the new Perhaps instance
     */
    public final Perhaps<T> onErrorReturnItem(T item) {
        Objects.requireNonNull(item, "item is null");
        return onAssembly(new PerhapsOnErrorReturnItem<>(this, item));
    }

    /**
     * If the upstream signals an error, switch to the given fallback
     * Perhaps.
     * @param fallback the fallback to switch to in case of an error
     * @return the new Perhaps instance
     */
    public final Perhaps<T> onErrorResumeWith(Perhaps<? extends T> fallback) {
        Objects.requireNonNull(fallback, "fallback is null");
        return onAssembly(new PerhapsOnErrorResumeWith<>(this, fallback));
    }

    /**
     * If the upstream signals an error, apply the given function to that
     * Throwable error and resume with the returned Perhaps.
     * @param fallbackSupplier the function that receives the upstream Throwable
     * and should return the fallback Perhaps that will be subscribed to as
     * a resumptions
     * @return the new Perhaps instance
     */
    public final Perhaps<T> onErrorResumeNext(Function<? super Throwable, ? extends Perhaps<? extends T>> fallbackSupplier) {
        Objects.requireNonNull(fallbackSupplier, "fallbackSupplier is null");
        return onAssembly(new PerhapsOnErrorResumeNext<>(this, fallbackSupplier));
    }

    /**
     * Signals a TimeoutException if the Perhaps doesn't signal an item
     * within the specified time.
     * @param timeout the time to wait for an item
     * @param unit the unit of time
     * @return the new Perhaps instance
     */
    public final Perhaps<T> timeout(long timeout, TimeUnit unit) {
        return timeout(timeout, unit, Schedulers.computation());
    }

    /**
     * Signals a TimeoutException if the Perhaps doesn't signal an item
     * or (terminates) within the specified time.
     * @param timeout the time to wait for an item
     * @param unit the unit of time
     * @param scheduler the scheduler to wait on
     * @return the new Perhaps instance
     */
    public final Perhaps<T> timeout(long timeout, TimeUnit unit, Scheduler scheduler) {
        return timeout(timer(timeout, unit, scheduler));
    }

    /**
     * Switch to the fallback Perhaps if this Perhaps doesn't signal an
     * item (or terminates) within the specified time.
     * @param timeout the time to wait for an item
     * @param unit the unit of time
     * @param fallback the Perhaps to switch to if this Perhaps times out
     * @return the new Perhaps instance
     */
    public final Perhaps<T> timeout(long timeout, TimeUnit unit, Perhaps<? extends T> fallback) {
        return timeout(timeout, unit, Schedulers.computation(), fallback);
    }

    /**
     * Switch to the fallback Perhaps if this Perhaps doesn't signal an
     * item (or terminates) within the specified time.
     * @param timeout the time to wait for an item
     * @param unit the unit of time
     * @param scheduler the scheduler to wait on
     * @param fallback the Perhaps to switch to if this Perhaps times out
     * @return the new Perhaps instance
     */
    public final Perhaps<T> timeout(long timeout, TimeUnit unit, Scheduler scheduler, Perhaps<? extends T> fallback) {
        return timeout(timer(timeout, unit, scheduler), fallback);
    }

    /**
     * Signal a TimeoutException if the other Publisher signals an item or
     * completes before this Perhaps does.
     * @param other the other Publisher that signals the timeout
     * @return the new Perhaps instance
     */
    public final Perhaps<T> timeout(Publisher<?> other) {
        Objects.requireNonNull(other, "other is null");
        return onAssembly(new PerhapsTimeout<>(this, other, null));
    }

    /**
     * Switch to the fallback Perhaps if the other Publisher signals an item or
     * completes before this Perhaps does.
     * @param other the other Publisher that signals the timeout
     * @param fallback the Perhaps to switch to in case of a timeout
     * @return the new Perhaps instance
     */
    public final Perhaps<T> timeout(Publisher<?> other, Perhaps<? extends T> fallback) {
        Objects.requireNonNull(other, "other is null");
        Objects.requireNonNull(fallback, "fallback is null");
        return onAssembly(new PerhapsTimeout<>(this, other, fallback));
    }

    /**
     * Signal the given item if this Perhaps is empty.
     * @param item the item to signal
     * @return the new Perhaps instance
     */
    public final Perhaps<T> defaultIfEmpty(T item) {
        Objects.requireNonNull(item, "item is null");
        return onAssembly(new PerhapsDefaultIfEmpty<>(this, item));
    }

    /**
     * Switch to the other Perhaps if this Perhaps is empty.
     * @param other the other Perhaps to switch to
     * @return the new Perhaps instance
     */
    public final Perhaps<T> switchIfEmpty(Perhaps<? extends T> other) {
        Objects.requireNonNull(other, "other is null");
        return onAssembly(new PerhapsSwitchIfEmpty<>(this, other));
    }

    /**
     * Repeats this Perhaps indefinitely.
     * @return the new Flowable instance
     */
    public final Flowable<T> repeat() {
        return Flowable.fromPublisher(this).repeat();
    }

    /**
     * Repeats this Perhaps at most the given number of times.
     * @param times the number of times to repeat
     * @return the new Flowable instance
     */
    public final Flowable<T> repeat(long times) {
        return Flowable.fromPublisher(this).repeat(times);
    }

    /**
     * Repeats this Perhaps until the given boolean supplier returns true when an
     * repeat iteration of this Perhaps completes.
     * @param stop the supplier of a boolean value that should return true to
     * stop repeating.
     * @return the new Flowable instance
     */
    public final Flowable<T> repeat(BooleanSupplier stop) {
        return Flowable.fromPublisher(this).repeatUntil(stop);
    }

    /**
     * Repeats this Perhaps when the Publisher returned by the handler function emits
     * an item or terminates if this Publisher terminates.
     * @param handler the function that receives Flowable that emits an item
     * when this Perhaps completes and returns a Publisher that should emit an item
     * to trigger a repeat or terminate to trigger a termination.
     * @return the new Flowable instance
     */
    public final Flowable<T> repeatWhen(Function<? super Flowable<Object>, ? extends Publisher<?>> handler) {
        return Flowable.fromPublisher(this).repeatWhen(handler);
    }

    /**
     * Retry this Perhaps indefinitely if it fails.
     * @return the new Perhaps instance
     */
    public final Perhaps<T> retry() {
        return onAssembly(new PerhapsRetry<>(this, Long.MAX_VALUE));
    }

    /**
     * Retry this Perhaps at most the given number of times if it fails.
     * @param times the number of times, Long.MAX_VALUE means indefinitely
     * @return the new Perhaps instance
     */
    public final Perhaps<T> retry(long times) {
        if (times < 0L) {
            throw new IllegalArgumentException("times >= 0 required but it was " + times);
        }
        return onAssembly(new PerhapsRetry<>(this, times));
    }

    /**
     * Retry this Perhaps if the predicate returns true for the latest failure
     * Throwable.
     * @param predicate the predicate receiving the latest Throwable and
     * if returns true, the Perhaps is retried.
     * @return the new Perhaps instance
     */
    public final Perhaps<T> retry(Predicate<? super Throwable> predicate) {
        Objects.requireNonNull(predicate, "predicate is null");
        return onAssembly(new PerhapsRetryWhile<>(this, predicate));
    }

    /**
     * Retry this Perhaps if the Publisher returned by the handler signals an item
     * in response to the failure Throwable.
     * @param handler the function that receives a Flowable that signals the
     * failure Throwable of this Perhaps and returns a Publisher which triggers a retry
     * or termination.
     * @return the new Perhaps instance
     */
    public final Perhaps<T> retryWhen(Function<? super Flowable<Throwable>, ? extends Publisher<?>> handler) {
        Objects.requireNonNull(handler, "handler is null");
        return onAssembly(new PerhapsRetryWhen<>(this, handler));
    }

    /**
     * Subscribes to the upstream Perhaps and requests on the
     * specified Scheduler.
     * @param scheduler the scheduler to subscribe on
     * @return the new Perhaps instance
     */
    public final Perhaps<T> subscribeOn(Scheduler scheduler) {
        Objects.requireNonNull(scheduler, "scheduler is null");
        return onAssembly(new PerhapsSubscribeOn<>(this, scheduler));
    }

    /**
     * Observe the events of this Perhaps on the specified scheduler.
     * @param scheduler the scheduler to observe events on
     * @return the new Perhaps instance
     */
    public final Perhaps<T> observeOn(Scheduler scheduler) {
        Objects.requireNonNull(scheduler, "scheduler is null");
        return onAssembly(new PerhapsObserveOn<>(this, scheduler));
    }

    /**
     * If the downstream cancels, the upstream is cancelled on
     * the specified scheduler.
     * <p>Note that normal termination don't trigger cancellation.
     * @param scheduler the scheduler to unsubscribe on
     * @return the new Perhaps instance
     */
    public final Perhaps<T> unsubscribeOn(Scheduler scheduler) {
        Objects.requireNonNull(scheduler, "scheduler is null");
        return onAssembly(new PerhapsUnsubscribeOn<>(this, scheduler));
    }

    /**
     * Executes a callback when the upstream calls onSubscribe.
     * @param onSubscribe the consumer called with the upstream Subscription
     * @return the new Perhaps instance
     */
    public final Perhaps<T> doOnSubscribe(Consumer<? super Subscription> onSubscribe) {
        Objects.requireNonNull(onSubscribe, "onSubscribe is null");
        return onAssembly(new PerhapsDoOnLifecycle<>(this,
                Functions.emptyConsumer(),
                Functions.emptyConsumer(),
                Functions.emptyConsumer(),
                Functions.EMPTY_ACTION,
                Functions.EMPTY_ACTION,
                onSubscribe,
                Functions.EMPTY_LONG_CONSUMER,
                Functions.EMPTY_ACTION
        ));
    }

    /**
     * Executes the callback when the downstream requests from this Perhaps.
     * @param onRequest the callback called with the request amount
     * @return the new Perhaps instance
     */
    public final Perhaps<T> doOnRequest(LongConsumer onRequest) {
        Objects.requireNonNull(onRequest, "onRequest is null");
        return onAssembly(new PerhapsDoOnLifecycle<>(this,
                Functions.emptyConsumer(),
                Functions.emptyConsumer(),
                Functions.emptyConsumer(),
                Functions.EMPTY_ACTION,
                Functions.EMPTY_ACTION,
                Functions.emptyConsumer(),
                onRequest,
                Functions.EMPTY_ACTION
        ));
    }

    /**
     * Executes the callback if the downstream cancels the sequence.
     * @param onCancel the action to call
     * @return the new Perhaps instance
     */
    public final Perhaps<T> doOnCancel(Action onCancel) {
        Objects.requireNonNull(onCancel, "onCancel is null");
        return onAssembly(new PerhapsDoOnLifecycle<>(this,
                Functions.emptyConsumer(),
                Functions.emptyConsumer(),
                Functions.emptyConsumer(),
                Functions.EMPTY_ACTION,
                Functions.EMPTY_ACTION,
                Functions.emptyConsumer(),
                Functions.EMPTY_LONG_CONSUMER,
                onCancel
        ));
    }

    /**
     * Executes a callback before the value is emitted to downstream.
     * @param onNext the consumer called with the current value before it is
     * is emitted to downstream.
     * @return the new Perhaps instance
     */
    public final Perhaps<T> doOnNext(Consumer<? super T> onNext) {
        Objects.requireNonNull(onNext, "onNext is null");
        return onAssembly(new PerhapsDoOnLifecycle<>(this,
                onNext,
                Functions.emptyConsumer(),
                Functions.emptyConsumer(),
                Functions.EMPTY_ACTION,
                Functions.EMPTY_ACTION,
                Functions.emptyConsumer(),
                Functions.EMPTY_LONG_CONSUMER,
                Functions.EMPTY_ACTION
        ));
    }

    /**
     * Executes a callback after the value is emitted to downstream.
     * @param onAfterNext the consumer called with the current value after it is
     * is emitted to downstream.
     * @return the new Perhaps instance
     */
    public final Perhaps<T> doAfterNext(Consumer<? super T> onAfterNext) {
        Objects.requireNonNull(onAfterNext, "onAfterNext is null");
        return onAssembly(new PerhapsDoOnLifecycle<>(this,
                Functions.emptyConsumer(),
                onAfterNext,
                Functions.emptyConsumer(),
                Functions.EMPTY_ACTION,
                Functions.EMPTY_ACTION,
                Functions.emptyConsumer(),
                Functions.EMPTY_LONG_CONSUMER,
                Functions.EMPTY_ACTION
        ));
    }

    /**
     * Executes a callback when the upstream signals an error.
     * @param onError the consumer called before the error is emitted to
     * the downstream
     * @return the new Perhaps instance
     */
    public final Perhaps<T> doOnError(Consumer<? super Throwable> onError) {
        Objects.requireNonNull(onError, "onError is null");
        return onAssembly(new PerhapsDoOnLifecycle<>(this,
                Functions.emptyConsumer(),
                Functions.emptyConsumer(),
                onError,
                Functions.EMPTY_ACTION,
                Functions.EMPTY_ACTION,
                Functions.emptyConsumer(),
                Functions.EMPTY_LONG_CONSUMER,
                Functions.EMPTY_ACTION
        ));
    }

    /**
     * Executes a callback when the upstream completes normally.
     * @param onComplete the consumer called before the completion event
     * is emitted to the downstream.
     * @return the new Perhaps instance
     */
    public final Perhaps<T> doOnComplete(Action onComplete) {
        Objects.requireNonNull(onComplete, "onComplete is null");
        return onAssembly(new PerhapsDoOnLifecycle<>(this,
                Functions.emptyConsumer(),
                Functions.emptyConsumer(),
                Functions.emptyConsumer(),
                onComplete,
                Functions.EMPTY_ACTION,
                Functions.emptyConsumer(),
                Functions.EMPTY_LONG_CONSUMER,
                Functions.EMPTY_ACTION
        ));
    }

    /**
     * Executes the callback after this Perhaps terminates and the downstream
     * is notified.
     * @param onAfterTerminate the callback to call after the downstream is notified
     * @return the new Perhaps instance
     */
    public final Perhaps<T> doAfterTerminate(Action onAfterTerminate) {
        Objects.requireNonNull(onAfterTerminate, "onAfterTerminate is null");
        return onAssembly(new PerhapsDoOnLifecycle<>(this,
                Functions.emptyConsumer(),
                Functions.emptyConsumer(),
                Functions.emptyConsumer(),
                Functions.EMPTY_ACTION,
                onAfterTerminate,
                Functions.emptyConsumer(),
                Functions.EMPTY_LONG_CONSUMER,
                Functions.EMPTY_ACTION
        ));
    }

    /**
     * Executes the callback exactly if the upstream terminates or
     * the downstream cancels the sequence.
     * @param onFinally the action to call
     * @return the new Perhaps instance
     */
    public final Perhaps<T> doFinally(Action onFinally) {
        Objects.requireNonNull(onFinally, "onFinally is null");
        return onAssembly(new PerhapsDoFinally<>(this, onFinally));
    }

    /**
     * Delay the emission of the signals of this Perhaps by the
     * given amount of time.
     * @param delay the delay amount
     * @param unit the time unit of the delay
     * @return the new Perhaps instance
     */
    public final Perhaps<T> delay(long delay, TimeUnit unit) {
        return delay(delay, unit, Schedulers.computation());
    }

    /**
     * Delay the emission of the signals of this Perhaps by the
     * given amount of time.
     * @param delay the delay amount
     * @param unit the time unit of the delay
     * @param scheduler the scheduler to delay on
     * @return the new Perhaps instance
     */
    public final Perhaps<T> delay(long delay, TimeUnit unit, Scheduler scheduler) {
        return delay(timer(delay, unit, scheduler));
    }

    /**
     * Delay the emission of the signals of this Perhaps till
     * the other Publisher signals an item or completes.
     * @param other the other Publisher to delay with
     * @return the new Perhaps instance
     */
    public final Perhaps<T> delay(Publisher<?> other) {
        Objects.requireNonNull(other, "other is null");
        return onAssembly(new PerhapsDelayPublisher<>(this, other));
    }

    /**
     * Delay the subscription to this Perhaps by the given time amount.
     * @param delay the amount to delay the subscription
     * @param unit the delay time unit
     * @return the new Perhaps instance
     */
    public final Perhaps<T> delaySubscription(long delay, TimeUnit unit) {
        return delaySubscription(timer(delay, unit));
    }

    /**
     * Delay the subscription to this Perhaps by the given time amount,
     * running on the specified Scheduler.
     * @param delay the amount to delay the subscription
     * @param unit the delay time unit
     * @param scheduler the scheduler to wait on
     * @return the new Perhaps instance
     */
    public final Perhaps<T> delaySubscription(long delay, TimeUnit unit, Scheduler scheduler) {
        return delaySubscription(timer(delay, unit, scheduler));
    }

    /**
     * Delay the subscription to this Perhaps until the other Publisher
     * signals an item or completes.
     * @param other the other Publisher that will trigger the actual
     * subscription to this Perhaps.
     * @return the new Perhaps instance
     */
    public final Perhaps<T> delaySubscription(Publisher<?> other) {
        Objects.requireNonNull(other, "other is null");
        return onAssembly(new PerhapsDelaySubscription<>(this, other));
    }

    /**
     * Map the downstream Subscriber into an upstream Subscriber.
     * @param <R> the downstream value type
     * @param onLift the function called with the downstream's Subscriber and
     * should return a Subscriber to be subscribed to this Perhaps.
     * @return the new Perhaps type
     */
    public final <R> Perhaps<R> lift(Function<Subscriber<? super R>, Subscriber<? super T>> onLift) {
        Objects.requireNonNull(onLift, "onLift is null");
        return onAssembly(new PerhapsLift<>(this, onLift));
    }

    /**
     * Applies the function, fluently to this Perhaps and returns the value it returns.
     * @param <R> the result type
     * @param converter the function receiving this Perhaps and returns a value to be returned
     * @return the value returned by the function
     */
    public final <R> R to(Function<? super Perhaps<T>, R> converter) {
        try {
            return converter.apply(this);
        } catch (Throwable ex) {
            throw ExceptionHelper.wrapOrThrow(ex);
        }
    }

    /**
     * Applies a function to this Perhaps and returns the Perhaps it returned.
     * @param <R> the result type
     * @param composer the function that receives this Perhaps and should return a Perhaps
     * @return the new Perhaps instance
     */
    public final <R> Perhaps<R> compose(Function<? super Perhaps<T>, ? extends Perhaps<R>> composer) {
        return to(composer);
    }

    /**
     * Try consuming this Perhaps until the other Publisher signals an item
     * or completes which then completes the Perhaps.
     * @param other the other Publisher instance
     * @return the new Perhaps instance
     */
    public final Perhaps<T> takeUntil(Publisher<?> other) {
        Objects.requireNonNull(other, "other is null");
        return onAssembly(new PerhapsTakeUntil<>(this, other));
    }

    /**
     * Convert this Perhaps into a Flowable.
     * @return the new Flowable instance
     */
    public final Flowable<T> toFlowable() {
        return RxJavaPlugins.onAssembly(new PerhapsToFlowable<>(this));
    }

    /**
     * Convert this Perhaps into an Observable.
     * @return the new Observable instance
     */
    public final Observable<T> toObservable() {
        return RxJavaPlugins.onAssembly(new PerhapsToObservable<>(this));
    }

    /**
     * Convert this Perhaps into a Maybe.
     * @return the new Maybe instance
     */
    public final Maybe<T> toMaybe() {
        return RxJavaPlugins.onAssembly(new PerhapsToMaybe<>(this));
    }

    /**
     * Caches the value or error event of the upstream Perhaps
     * and relays/replays it to Subscribers.
     * @return the new Perhaps instance
     *
     * @since 0.14.1
     */
    public final Perhaps<T> cache() {
        return onAssembly(new PerhapsCache<>(this));
    }

    // ----------------------------------------------------
    // Consumers (leave)
    // ----------------------------------------------------

    @Override
    public final void subscribe(Subscriber<? super T> s) {
        Objects.requireNonNull(s, "s is null");
        try {
            subscribeActual(s);
        } catch (NullPointerException ex) {
            throw ex;
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            NullPointerException npe = new NullPointerException();
            npe.initCause(ex);
            throw npe;
        }
    }

    /**
     * Implement this method to react to a Subscriber subscribing to this Perhaps.
     * @param s the downstream Subscriber, never null
     */
    protected abstract void subscribeActual(Subscriber<? super T> s);

    /**
     * Subscribe with a Subscriber (subclass) and return it as is.
     * @param <E> the Subscriber subclass type
     * @param s the soubscriber, not null
     * @return the {@code s} as is
     */
    public final <E extends Subscriber<? super T>> E subscribeWith(E s) {
        subscribe(s);
        return s;
    }

    /**
     * Subscribe to this Perhaps and ignore any signal it produces.
     * @return the Disposable that allows cancelling the subscription
     */
    public final Disposable subscribe() {
        return subscribe(Functions.emptyConsumer(), Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION);
    }

    /**
     * Subscribes to this Perhaps and calls the onNext if this Perhaps succeeds.
     * @param onNext called when this Perhaps succeeds
     * @return the Disposable that allows cancelling the subscription
     */
    public final Disposable subscribe(Consumer<? super T> onNext) {
        return subscribe(onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION);
    }

    /**
     * Subscribes to this Perhaps and calls the appropriate callback for the resulting signal.
     * @param onNext called when this Perhaps succeeds
     * @param onError called when this Perhaps fails
     * @return the Disposable that allows cancelling the subscription
     */
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {
        return subscribe(onNext, onError, Functions.EMPTY_ACTION);
    }

    /**
     * Subscribes to this Perhaps and calls the appropriate callback for the resulting signal.
     * @param onNext called when this Perhaps succeeds
     * @param onError called when this Perhaps fails
     * @param onComplete called when this Perhaps succeeds after the call to onNext
     * @return the Disposable that allows cancelling the subscription
     */
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {
        LambdaSubscriber<T> s = new LambdaSubscriber<>(onNext, onError, onComplete, Functions.REQUEST_MAX);
        subscribe(s);
        return s;
    }

    /**
     * Blocks until this Perhaps terminates and ignores the signals it produced.
     */
    public final void blockingSubscribe() {
        blockingSubscribe(Functions.emptyConsumer(), Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION);
    }

    /**
     * Blocks until this Perhaps terminates and calls the onNext with the success value.
     * @param onNext the callback to call when this Perhaps completes with a success value
     */
    public final void blockingSubscribe(Consumer<? super T> onNext) {
        blockingSubscribe(onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION);
    }

    /**
     * Blocks until this Perhaps terminates and calls the onNext with the success value
     * or calls the onError with the error Throwable.
     * @param onNext the callback to call when this Perhaps completes with a success value
     * @param onError the callback to call when this Perhaps fails with an error
     */
    public final void blockingSubscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {
        blockingSubscribe(onNext, onError, Functions.EMPTY_ACTION);
    }

    /**
     * Blocks until this Perhaps produces its terminal signal and calls the
     * appropriate callback(s) based on the signal type.
     * @param onNext called when the Perhaps succeeds
     * @param onError called when the Perhaps fails
     * @param onComplete called when the Perhaps succeeds after the call to onNext.
     */
    public final void blockingSubscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {
        Objects.requireNonNull(onNext, "onNext is null");
        Objects.requireNonNull(onError, "onError is null");
        Objects.requireNonNull(onComplete, "onComplete is null");
        BlockingGetSubscriber<T> s = new BlockingGetSubscriber<>();
        subscribe(s);
        s.blockingCall(onNext, onError, onComplete);
    }

    /**
     * Blockingly awaits indefinitely the success value of this Perhaps or rethrows
     * its error (wrapped into a RuntimeException if necessary).
     * @return the success value of this Perhaps
     */
    public final T blockingGet() {
        BlockingGetSubscriber<T> s = new BlockingGetSubscriber<>();
        subscribe(s);
        return s.blockingGet();
    }

    /**
     * Blockingly awaits at most the given timeout for the success
     * value of this Perhaps or rethrows
     * its error (wrapped into a RuntimeException if necessary).
     * @param timeout the time to wait for a success value
     * @param unit the time unit of the timeout
     * @return the success value of this Perhaps
     */
    public final T blockingGet(long timeout, TimeUnit unit) {
        BlockingGetSubscriber<T> s = new BlockingGetSubscriber<>();
        subscribe(s);
        return s.blockingGet(timeout, unit);
    }

    /**
     * Creates a TestSubscriber and subscribes it to this Perhaps.
     * @return the new TestSubscriber instance
     */
    public final TestSubscriber<T> test() {
        return test(Long.MAX_VALUE, false);
    }

    /**
     * Creates a TestSubscriber, optionally cancels it, and subscribes
     * it to this Perhaps.
     * @param cancel if true, the TestSubscriber will be cancelled before
     * subscribing to this Perhaps
     * @return the new TestSubscriber instance
     */
    public final TestSubscriber<T> test(boolean cancel) {
        return test(Long.MAX_VALUE, cancel);
    }

    /**
     * Creates a TestSubscriber with the given initial request and
     * subscribes it to this Perhaps.
     * @param initialRequest the initial request amount, non-negative
     * @return the new TestSubscriber
     */
    public final TestSubscriber<T> test(long initialRequest) {
        return test(initialRequest, false);
    }

    /**
     * Creates a TestSubscriber with the given initial request,
     * optionally cancels it, and subscribes it to this Perhaps.
     * @param initialRequest the initial request amount, non-negative
     * @param cancel if true, the TestSubscriber will be cancelled before
     * subscribing to this Perhaps
     * @return the new TestSubscriber
     */
    public final TestSubscriber<T> test(long initialRequest, boolean cancel) {
        TestSubscriber<T> ts = new TestSubscriber<>(initialRequest);
        if (cancel) {
            ts.cancel();
        }
        subscribe(ts);
        return ts;
    }

    /**
     * Converts this Perhaps into a Future and signals its single
     * value or null if this Perhaps is empty.
     * @return the new Future instance
     */
    public final Future<T> toFuture() {
        FuturePerhapsSubscriber<T> fs = new FuturePerhapsSubscriber<>();
        subscribe(fs);
        return fs;
    }
}
