content: /*
   * 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.annotations.SchedulerSupport;
  import io.reactivex.rxjava3.core.*;
  import io.reactivex.rxjava3.disposables.Disposable;
  import io.reactivex.rxjava3.exceptions.*;
  import io.reactivex.rxjava3.functions.*;
  import io.reactivex.rxjava3.internal.functions.*;
  import io.reactivex.rxjava3.internal.util.*;
  import io.reactivex.rxjava3.plugins.RxJavaPlugins;
  import io.reactivex.rxjava3.schedulers.Schedulers;
  import io.reactivex.rxjava3.subscribers.TestSubscriber;
  
  /**
   * Represents the base reactive class with fluent API for Publisher-based,
   * no-item just onError or onComplete source.
   * <p>
   * Since this type never emits any value, the implementations ignore
   * the downstream request and emit the terminal events even if there was
   * no request (which is allowed by the Reactive-Streams specification).
   * <p>
   * Since there is no bottom type in Java (that is T is a subtype of all other types),
   * Nono implements the Publisher interface via the Void type parameter.
   * @since 0.11.0
   */
  public abstract class Nono implements Publisher<Void> {
  
      /**
       * Hook called when assembling Nono sequences.
       */
      private static volatile Function<Nono, Nono> onAssemblyHandler;
  
      /**
       * Returns the default buffer or prefetch size.
       * @return the buffer or prefetch size
       */
      public static int bufferSize() {
          return Flowable.bufferSize();
      }
  
      /**
       * Optionally apply a function to the raw source and return a
       * potentially modified Nono instance.
       * @param source the source to apply to
       * @return the possibly wrapped Nono instance
       */
      protected static Nono onAssembly(Nono source) {
          Function<Nono, Nono> f = onAssemblyHandler;
          if (f != null) {
              try {
                  return Objects.requireNonNull(f.apply(source), "The onAssemblyHandler returned a null Nono");
              } catch (Throwable ex) {
                  throw ExceptionHelper.wrapOrThrow(ex);
              }
          }
          return source;
      }
  
      /**
       * Returns the current onAssembly handler function or null if not set.
       * @return the current onAssembly handler function, maybe null
       */
      public static Function<Nono, Nono> getOnAssemblyHandler() {
          return onAssemblyHandler;
      }
  
      /**
       * Sets the onAssembly handler.
       * @param handler the new onAssembly handler, null clears the handler
       */
      public static void setOnAssemblyHandler(Function<Nono, Nono> handler) {
          onAssemblyHandler = handler;
      }
  
      // -----------------------------------------------------------
      // Static factories (enter)
      // -----------------------------------------------------------
  
      /**
       * Creates a Nono instance that when subscribed to, the given onCreate is
       * called for each individual subscriber to generate a terminal event
       * synchronously and synchronously in a cancellation-safe manner.
       * @param onCreate called for each individual subscriber with the abstraction
       * of the incoming Subscriber
       * @return the new Nono instance
       */
      public static Nono create(CompletableOnSubscribe onCreate) {
          Objects.requireNonNull(onCreate, "onCreate is null");
          return onAssembly(new NonoCreate(onCreate));
      }
  
      /**
       * Returns a Nono that completes normally.
       * @return the new Nono instance
       */
      public static Nono complete() {
          return onAssembly(NonoComplete.INSTANCE);
      }
  
      /**
       * Returns a Nono that never terminates.
       * @return the new Nono instance
       */
      public static Nono never() {
          return onAssembly(NonoNever.INSTANCE);
      }
  
      /**
       * Returns a Nono that signals the given Throwable to all
       * subscribers.
       * @param ex the Throwable to signal, not null
       * @return the new Nono instance
       */
      public static Nono error(Throwable ex) {
          Objects.requireNonNull(ex, "ex is null");
          return onAssembly(new NonoError(ex));
      }
  
      /**
       * Returns a Nono that signals a Throwable generated by the
       * supplier for each individual subscriber.
       * @param errorSupplier the Throwable error supplier, not null
       * @return the new Nono instance
       */
      public static Nono error(Supplier<? extends Throwable> errorSupplier) {
          Objects.requireNonNull(errorSupplier, "errorSupplier is null");
          return onAssembly(new NonoErrorSupplier(errorSupplier));
      }
  
      /**
       * Defers the creation of the actual Nono instance until a subscriber
       * subscribes.
       * @param supplier the supplier of Nono instances for each individual
       * subscriber.
       * @return the new Nono instance
       */
      public static Nono defer(Supplier<? extends Nono> supplier) {
          Objects.requireNonNull(supplier, "supplier is null");
          return onAssembly(new NonoDefer(supplier));
      }
  
      /**
       * Executes an action when a subscriber subscribes to the returned
       * Nono.
       * @param action the action to execute, not null
       * @return the new nono instance
       */
      public static Nono fromAction(Action action) {
          Objects.requireNonNull(action, "action is null");
          return onAssembly(new NonoFromAction(action));
      }
  
      /**
       * Blockingly waits indefinitely for the given Future to terminate,
       * relaying any error the Future signals.
       * @param future the future to await
       * @return the new Nono instance
       */
      public static Nono fromFuture(Future<?> future) {
          Objects.requireNonNull(future, "future is null");
          return onAssembly(new NonoFromFuture(future, 0L, TimeUnit.NANOSECONDS));
      }
  
      /**
       * Blockingly waits the given Future for the given timeout to terminate,
       * relaying any error the Future signals.
       * @param future the future to await
       * @param timeout the timeout value to wait for termination
       * @param unit the unit for the timeout parameter
       * @return the new Nono instance
       */
      public static Nono fromFuture(Future<?> future, long timeout, TimeUnit unit) {
          Objects.requireNonNull(future, "future is null");
          Objects.requireNonNull(unit, "unit is null");
          return onAssembly(new NonoFromFuture(future, timeout, unit));
      }
  
      /**
       * Returns a Nono that terminates when the first Nono from the
       * sources sequence terminates.
       * @param sources the Iterable sequence of sources
       * @return the new Nono instance
       */
      public static Nono amb(Iterable<? extends Nono> sources) {
          Objects.requireNonNull(sources, "sources is null");
          return onAssembly(new NonoAmbIterable(sources));
      }
  
      /**
       * Returns a Nono that terminates when the first Nono from the
       * array terminates.
       * @param sources the array of sources
       * @return the new Nono instance
       */
      public static Nono ambArray(Nono... sources) {
          Objects.requireNonNull(sources, "sources is null");
          return onAssembly(new NonoAmbArray(sources));
      }
  
      /**
       * Runs the Nono sources one after the other.
       * @param sources the Iterable sequence of sources
       * @return the new Nono instances
       */
      public static Nono concat(Iterable<? extends Nono> sources) {
          Objects.requireNonNull(sources, "sources is null");
          return onAssembly(new NonoConcatIterable(sources, false));
      }
  
      /**
       * Runs the Nono sources emitted by the Publisher one after the other.
       * @param sources the Publisher of Nono sources
       * @return the new Nono instance
       */
      public static Nono concat(Publisher<? extends Nono> sources) {
          return concat(sources, 2);
      }
  
      /**
       * Runs the Nono sources emitted by the Publisher one after the other,
       * prefetching the given number of Nono sources.
       * @param sources the Publisher of Nono sources
       * @param prefetch the number of Nono sources to prefetch from upstream
       * @return the new Nono instance
       */
      public static Nono concat(Publisher<? extends Nono> sources, int prefetch) {
          Objects.requireNonNull(sources, "sources is null");
          ObjectHelper.verifyPositive(prefetch, "prefetch");
          return onAssembly(new NonoConcat(sources, prefetch, ErrorMode.IMMEDIATE));
      }
  
      /**
       * Runs the Nono sources one after the other.
       * @param sources the array of sources
       * @return the new Nono instances
       */
      public static Nono concatArray(Nono... sources) {
          Objects.requireNonNull(sources, "sources is null");
          return onAssembly(new NonoConcatArray(sources, false));
      }
  
      /**
       * Runs the Nono sources one after the other, delaying errors from them
       * till all sources have terminated.
       * @param sources the Iterable sequence of sources
       * @return the new Nono instances
       */
      public static Nono concatDelayError(Iterable<? extends Nono> sources) {
          Objects.requireNonNull(sources, "sources is null");
          return onAssembly(new NonoConcatIterable(sources, true));
      }
  
      /**
       * Runs the Nono sources emitted by the Publisher, one after the other, delaying errors from them
       * till all sources have terminated.
       * @param sources the Publisher of source Nonos
       * @return the new Nono instances
       */
      public static Nono concatDelayError(Publisher<? extends Nono> sources) {
          return concatDelayError(sources, 2, true);
      }
  
      /**
       * Runs the Nono sources emitted by the Publisher, one after the other, delaying errors from them
       * till all sources have terminated and prefetching Nonos from the upstream.
       * @param sources the Publisher of source Nonos
       * @param prefetch the number of Nonos to prefetch from the upstream
       * @param tillTheEnd if true the errors from the source are also delayed till the end;
       *                   if false, error(s) are emitted when an inner Nono source terminates
       * @return the new Nono instances
       */
      public static Nono concatDelayError(Publisher<? extends Nono> sources, int prefetch, boolean tillTheEnd) {
          Objects.requireNonNull(sources, "sources is null");
          ObjectHelper.verifyPositive(prefetch, "prefetch");
          return onAssembly(new NonoConcat(sources, prefetch, tillTheEnd ? ErrorMode.END : ErrorMode.BOUNDARY));
      }
  
      /**
       * Runs the Nono sources one after the other, delaying errors from them
       * till all sources have terminated.
       * @param sources the array of sources
       * @return the new Nono instances
       */
      public static Nono concatArrayDelayError(Nono... sources) {
          Objects.requireNonNull(sources, "sources is null");
          return onAssembly(new NonoConcatArray(sources, true));
      }
  
      /**
       * Runs all Nono sources at once and completes once all of them complete.
       * @param sources the Iterable sequence of Nono sources
       * @return the new Nono instance
       */
      public static Nono merge(Iterable<? extends Nono> sources) {
          return merge(sources, Integer.MAX_VALUE);
      }
  
      /**
       * Runs the maximum number of Nono sources at once and completes when all source
       * Nono complete.
       * @param sources the Iterable sequence of Nono sources
       * @param maxConcurrency the maximum number of active Nono sources at a given time
       * @return the new Nono instance
       */
      public static Nono merge(Iterable<? extends Nono> sources, int maxConcurrency) {
          Objects.requireNonNull(sources, "sources is null");
          ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");
          return onAssembly(new NonoMergeIterable(sources, false, maxConcurrency));
      }
  
      /**
       * Runs all Nono sources emitted by the Publisher at once and completes once all of them complete.
       * @param sources the Publisher of Nono sources
       * @return the new Nono instance
       */
      public static Nono merge(Publisher<? extends Nono> sources) {
          return merge(sources, Integer.MAX_VALUE);
      }
  
      /**
       * Runs the maximum number of Nono sources emitted by the Publisher at once and completes when all source
       * Nono complete.
       * @param sources the Publisher of Nono sources
       * @param maxConcurrency the maximum number of active Nono sources at a given time
       * @return the new Nono instance
       */
      public static Nono merge(Publisher<? extends Nono> sources, int maxConcurrency) {
          Objects.requireNonNull(sources, "sources is null");
          ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");
          return onAssembly(new NonoMerge(sources, false, maxConcurrency));
      }
  
      /**
       * Runs all Nono sources at once and completes once all of them complete.
       * @param sources the array of Nono sources
       * @return the new Nono instance
       */
      public static Nono mergeArray(Nono... sources) {
          return mergeArray(Integer.MAX_VALUE, sources);
      }
  
      /**
       * Runs the maximum number of Nono sources at once and completes when all source
       * Nono complete.
       * @param sources the array of Nono sources
       * @param maxConcurrency the maximum number of active Nono sources at a given time
       * @return the new Nono instance
       */
      public static Nono mergeArray(int maxConcurrency, Nono... sources) {
          ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");
          Objects.requireNonNull(sources, "sources is null");
          return onAssembly(new NonoMergeArray(sources, false, maxConcurrency));
      }
  
      /**
       * Runs all Nono sources at once and terminates once all of them terminate,
       * delaying errors in the process.
       * @param sources the Iterable sequence of Nono sources
       * @return the new Nono instance
       */
      public static Nono mergeDelayError(Iterable<? extends Nono> sources) {
          return mergeDelayError(sources, Integer.MAX_VALUE);
      }
  
      /**
       * Runs the maximum number of Nono sources at once and terminates when all source
       * Nono terminate, delaying errors in the process.
       * @param sources the Iterable sequence of Nono sources
       * @param maxConcurrency the maximum number of active Nono sources at a given time
       * @return the new Nono instance
       */
      public static Nono mergeDelayError(Iterable<? extends Nono> sources, int maxConcurrency) {
          Objects.requireNonNull(sources, "sources is null");
          ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");
          return onAssembly(new NonoMergeIterable(sources, true, maxConcurrency));
      }
  
      /**
       * Runs all Nono sources emitted by the Publisher at once and terminates
       *  once all of them terminate, delaying errors in the process.
       * @param sources the Publisher of Nono sources
       * @return the new Nono instance
       */
      public static Nono mergeDelayError(Publisher<? extends Nono> sources) {
          return mergeDelayError(sources, Integer.MAX_VALUE);
      }
  
      /**
       * Runs the maximum number of Nono sources emitted by the Publisher
       * at once and terminates when all source Nono terminate,
       * delaying errors in the process.
       * @param sources the Publisher of Nono sources
       * @param maxConcurrency the maximum number of active Nono sources at a given time
       * @return the new Nono instance
       */
      public static Nono mergeDelayError(Publisher<? extends Nono> sources, int maxConcurrency) {
          Objects.requireNonNull(sources, "sources is null");
          ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");
          return onAssembly(new NonoMerge(sources, true, maxConcurrency));
      }
  
      /**
       * Runs all Nono sources at once and terminates once all
       * of them terminate, delaying errors in the process.
       * @param sources the array of Nono sources
       * @return the new Nono instance
       */
      public static Nono mergeArrayDelayError(Nono... sources) {
          return mergeArrayDelayError(bufferSize(), sources);
      }
  
      /**
       * Runs the maximum number of Nono sources at once and terminates when all source
       * Nono terminate, delaying errors in the process.
       * @param sources the array of Nono sources
       * @param maxConcurrency the maximum number of active Nono sources at a given time
       * @return the new Nono instance
       */
      public static Nono mergeArrayDelayError(int maxConcurrency, Nono... sources) {
          ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");
          Objects.requireNonNull(sources, "sources is null");
          return onAssembly(new NonoMergeArray(sources, true, maxConcurrency));
      }
  
      /**
       * Completes after the specified amount of time on the computation scheduler.
       * @param delay the delay value
       * @param unit the delay time unit
       * @return the new Nono instance
       */
      @SchedulerSupport(SchedulerSupport.COMPUTATION)
      public static Nono timer(long delay, TimeUnit unit) {
          return timer(delay, unit, Schedulers.computation());
      }
  
      /**
       * Completes after the specified amount of time on the specified scheduler.
       * @param delay the delay value
       * @param unit the delay time unit
       * @param scheduler the scheduler to delay the completion signal
       * @return the new Nono instance
       */
      @SchedulerSupport(SchedulerSupport.CUSTOM)
      public static Nono timer(long delay, TimeUnit unit, Scheduler scheduler) {
          Objects.requireNonNull(unit, "unit is null");
          Objects.requireNonNull(scheduler, "scheduler is null");
          return onAssembly(new NonoTimer(delay, unit, scheduler));
      }
  
      /**
       * Generate a resource and a Nono based on that resource and then
       * dispose that resource eagerly when the Nono terminates or the
       * downstream cancels the sequence.
       * @param <R> the resource type
       * @param resourceSupplier the callback to get a resource for each subscriber
       * @param sourceSupplier the function that returns a Nono for the generated resource
       * @param disposer the consumer of the resource once the upstream terminates or the
       * downstream cancels
       * @return the new Nono instance
       */
      public static <R> Nono using(Supplier<R> resourceSupplier, Function<? super R, ? extends Nono> sourceSupplier,
              Consumer<? super R> disposer) {
          return using(resourceSupplier, sourceSupplier, disposer, true);
      }
  
      /**
       * Generate a resource and a Nono based on that resource and then
       * dispose that resource optionally eagerly when the Nono terminates or the
       * downstream cancels the sequence.
       * @param <R> the resource type
       * @param resourceSupplier the callback to get a resource for each subscriber
       * @param sourceSupplier the function that returns a Nono 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 Nono instance
       */
      public static <R> Nono using(Supplier<R> resourceSupplier, Function<? super R, ? extends Nono> 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 NonoUsing<>(resourceSupplier, sourceSupplier, disposer, eager));
      }
  
      /**
       * Wrap a general Publisher, ignore all of its values and terminate if
       * the source Publisher terminates.
       * @param source the Publisher to wrap into a Nono
       * @return the Nono instance
       */
      public static Nono fromPublisher(Publisher<?> source) {
          if (source instanceof Nono) {
              return (Nono)source;
          }
          Objects.requireNonNull(source, "source is null");
          return onAssembly(new NonoFromPublisher(source));
      }
  
      /**
       * Wrap a Single, ignore its success value and terminate if
       * the source Single terminates.
       * @param source the SingleSource to wrap into a Nono
       * @return the new Nono instance
       */
      public static Nono fromSingle(SingleSource<?> source) {
          Objects.requireNonNull(source, "source is null");
          return onAssembly(new NonoFromSingle(source));
      }
  
      /**
       * Wrap a Maybe, ignore its success value and terminate if
       * the source Maybe terminates.
       * @param source the MaybeSource to wrap into a Nono
       * @return the new Nono instance
       */
      public static Nono fromMaybe(MaybeSource<?> source) {
          Objects.requireNonNull(source, "source is null");
          return onAssembly(new NonoFromMaybe(source));
      }
  
      /**
       * Wrap a Completable into a Nono and terminate when the
       * source Completable terminates.
       * @param source the MaybeSource to wrap into a Nono
       * @return the new Nono instance
       */
      public static Nono fromCompletable(CompletableSource source) {
          Objects.requireNonNull(source, "source is null");
          return onAssembly(new NonoFromCompletable(source));
      }
  
      /**
       * Wrap a general Observable, ignore all of its values and terminate if
       * the source Observable terminates.
       * @param source the ObservableSource to wrap into a Nono
       * @return the new Nono instance
       */
      public static Nono fromObservable(ObservableSource<?> source) {
          Objects.requireNonNull(source, "source is null");
          return onAssembly(new NonoFromObservable(source));
      }
  
      // -----------------------------------------------------------
      // Instance operators (stay)
      // -----------------------------------------------------------
  
      /**
       * When this Nono completes, it is continued by the events of
       * the other Publisher.
       * @param <T> the value type of the other Publisher
       * @param other the other Publisher to continue with
       * @return the new Flowable instance
       */
      public final <T> Flowable<T> andThen(Publisher<? extends T> other) {
          Objects.requireNonNull(other, "other is null");
          return RxJavaPlugins.onAssembly(new NonoAndThenPublisher<T>(this, other));
      }
  
      /**
       * Run the other Nono when this Nono completes.
       * @param other the other Nono to continue with.
       * @return the new Nono instance
       */
      public final Nono andThen(Nono other) {
          Objects.requireNonNull(other, "other is null");
          return onAssembly(new NonoAndThen(this, other));
      }
  
      /**
       * Delay the emission of the terminal events of this Nono
       * by the given time amount.
       * @param delay the delay amount
       * @param unit the time unit
       * @return the new Nono instance
       */
      @SchedulerSupport(SchedulerSupport.COMPUTATION)
      public final Nono delay(long delay, TimeUnit unit) {
          return delay(delay, unit, Schedulers.computation());
      }
  
      /**
       * Delay the emission of the terminal events of this Nono
       * by the given time amount.
       * @param delay the delay amount
       * @param unit the time unit
       * @param scheduler the scheduler to wait on
       * @return the new Nono instance
       */
      @SchedulerSupport(SchedulerSupport.CUSTOM)
      public final Nono delay(long delay, TimeUnit unit, Scheduler scheduler) {
          Objects.requireNonNull(unit, "unit is null");
          Objects.requireNonNull(scheduler, "scheduler is null");
          return onAssembly(new NonoDelay(this, delay, unit, scheduler));
      }
  
      /**
       * Delays the actual subscription to this Nono until the other
       * Publisher signals an item or completes.
       * @param other the other Publisher to await a signal from
       * @return the new Nono instance
       */
      public final Nono delaySubscription(Publisher<?> other) {
          Objects.requireNonNull(other, "other is null");
          return onAssembly(new NonoDelaySubscription(this, other));
      }
  
      /**
       * Delays the actual subscription to this Nono until the given
       * time passes.
       * @param delay the delay amount
       * @param unit the time unit
       * @return the new Nono instance
       */
      public final Nono delaySubscription(long delay, TimeUnit unit) {
          return delaySubscription(timer(delay, unit));
      }
  
      /**
       * Delays the actual subscription to this Nono until the given
       * time passes.
       * @param delay the delay amount
       * @param unit the time unit
       * @param scheduler the scheduler to wait on
       * @return the new Nono instance
       */
      public final Nono delaySubscription(long delay, TimeUnit unit, Scheduler scheduler) {
          return delaySubscription(timer(delay, unit, scheduler));
      }
  
      /**
       * Signals a TimeoutException if this Nono doesn't complete
       * within the specified timeout.
       * @param timeout the timeout amount
       * @param unit the time unit
       * @return the new Nono instance
       */
      public final Nono timeout(long timeout, TimeUnit unit) {
          return timeout(timeout, unit, Schedulers.computation());
      }
  
      /**
       * Switches to the fallback Nono if this Nono doesn't complete
       * within the specified timeout.
       * @param timeout the timeout amount
       * @param unit the time unit
       * @param fallback the Nono to switch to if this Nono times out
       * @return the new Nono instance
       */
      public final Nono timeout(long timeout, TimeUnit unit, Nono fallback) {
          return timeout(timeout, unit, Schedulers.computation(), fallback);
      }
  
      /**
       * Signals a TimeoutException if this Nono doesn't complete
       * within the specified timeout.
       * @param timeout the timeout amount
       * @param unit the time unit
       * @param scheduler the scheduler to wait on
       * @return the new Nono instance
       */
      public final Nono timeout(long timeout, TimeUnit unit, Scheduler scheduler) {
          return timeout(timer(timeout, unit, scheduler));
      }
  
      /**
       * Switches to the fallback Nono if this Nono doesn't complete
       * within the specified timeout.
       * @param timeout the timeout amount
       * @param unit the time unit
       * @param scheduler the scheduler to wait on
       * @param fallback the Nono to switch to if this Nono times out
       * @return the new Nono instance
       */
      public final Nono timeout(long timeout, TimeUnit unit, Scheduler scheduler, Nono fallback) {
          Objects.requireNonNull(unit, "unit is null");
          Objects.requireNonNull(scheduler, "scheduler is null");
          Objects.requireNonNull(fallback, "fallback is null");
          return timeout(timer(timeout, unit, scheduler), fallback);
      }
  
      /**
       * Signal a TimeoutException if the other Publisher signals an item
       * or completes before this Nono completes.
       * @param other the other Publisher instance
       * @return the new Nono instance
       */
      public final Nono timeout(Publisher<?> other) {
          Objects.requireNonNull(other, "other is null");
          return onAssembly(new NonoTimeout(this, other, null));
      }
  
      /**
       * Switch to the fallback Nono if the other Publisher signals an
       * item or completes before this Nono completes.
       * @param other the other Publisher instance
       * @param fallback the fallback Nono instance
       * @return the new Nono instance
       */
      public final Nono timeout(Publisher<?> other, Nono fallback) {
          Objects.requireNonNull(other, "other is null");
          Objects.requireNonNull(fallback, "fallback is null");
          return onAssembly(new NonoTimeout(this, other, fallback));
      }
  
      /**
       * If this Nono signals an error, signal an onComplete instead.
       * @return the new Nono instance
       */
      public final Nono onErrorComplete() {
          return onAssembly(new NonoOnErrorComplete(this));
      }
  
      /**
       * If this Nono signals an error, subscribe to the fallback Nono
       * returned by the error handler function.
       * @param errorHandler the function called with the error and should
       * return a Nono to resume with.
       * @return the new Nono instance
       */
      public final Nono onErrorResumeNext(Function<? super Throwable, ? extends Nono> errorHandler) {
          Objects.requireNonNull(errorHandler, "errorHandler is null");
          return onAssembly(new NonoOnErrorResume(this, errorHandler));
      }
  
      /**
       * Maps the upstream error into another Throwable via a function.
       * @param mapper the function that receives the upstream Throwable
       * and should return another Throwable to be emitted to downstream
       * @return the new Nono instance
       */
      public final Nono mapError(Function<? super Throwable, ? extends Throwable> mapper) {
          Objects.requireNonNull(mapper, "mapper is null");
          return onAssembly(new NonoMapError(this, mapper));
      }
  
      /**
       * Maps the upstream completion or error into a Publisher and emit
       * its events as a Flowable.
       * @param <T> the value type
       * @param onErrorMapper the function that receives the upstream error and
       * returns a Publisher to emit events of
       * @param onCompleteMapper the supplier that returns a Publisher to emit
       * events of
       * @return the new Flowable instance
       */
      public final <T> Flowable<T> flatMap(Function<? super Throwable, ? extends Publisher<? extends T>> onErrorMapper,
              Supplier<? extends Publisher<? extends T>> onCompleteMapper) {
          Objects.requireNonNull(onErrorMapper, "onErrorMapper is null");
          Objects.requireNonNull(onCompleteMapper, "onCompleteMapper is null");
          return RxJavaPlugins.onAssembly(new NonoFlatMapSignal<>(this, onErrorMapper, onCompleteMapper));
      }
  
      /**
       * Compose operators fluently via a function callback that returns a Nono for
       * this Nono.
       * @param composer the function receiving this and returns a Nono
       * @return the Nono returned by the function
       */
      public final Nono compose(Function<? super Nono, ? extends Nono> composer) {
          return to(composer);
      }
  
      /**
       * Fluently convert this Nono via a function callback into some type.
       * @param <R> the result value type
       * @param converter the function receiving this and returning a value
       * @return the value returned by the converter
       */
      public final <R> R to(Function<? super Nono, R> converter) {
          try {
              return converter.apply(this);
          } catch (Throwable ex) {
              throw ExceptionHelper.wrapOrThrow(ex);
          }
      }
  
      /**
       * Transform the downstream's Subscriber into a Subscriber for the upstream
       * via a function.
       * @param lifter the function receiving the downstream Subscriber and returns a Subscriber
       * for the upstream.
       * @return the new Nono instance
       */
      public final Nono lift(Function<Subscriber<? super Void>, Subscriber<? super Void>> lifter) {
          Objects.requireNonNull(lifter, "lifter is null");
          return onAssembly(new NonoLift(this, lifter));
      }
  
      /**
       * Convert this Nono instance into a Flowable that only terminates.
       * @param <T> the value type
       * @return the new Flowable instance
       */
      @SuppressWarnings({ "unchecked", "rawtypes" })
      public final <T> Flowable<T> toFlowable() {
          return (Flowable)Flowable.fromPublisher(this);
      }
  
      /**
       * Convert this Nono instance into an Observable that only terminates.
       * @param <T> the value type
       * @return the new Observable instance
       */
      @SuppressWarnings({ "unchecked", "rawtypes" })
      public final <T> Observable<T> toObservable() {
          return (Observable)Observable.fromPublisher(this);
      }
  
      /**
       * Convert this Nono instance into a Completable.
       * @return the new Completable instance
       */
      public final Completable toCompletable() {
          return Completable.fromPublisher(this);
      }
  
      /**
       * Convert this Nono instance into a Maybe that only terminates.
       * @param <T> the value type
       * @return the new Maybe instance
       */
      public final <T> Maybe<T> toMaybe() {
          return RxJavaPlugins.onAssembly(new NonoToMaybe<T>(this));
      }
  
      /**
       * Convert this Nono instance into a Perhaps that only terminates.
       * @param <T> the value type
       * @return the new Perhap instance
       */
      public final <T> Perhaps<T> toPerhaps() {
          return Perhaps.onAssembly(new NonoToPerhaps<T>(this));
      }
  
      /**
       * Subscribes to the upstream on the specified Scheduler.
       * @param scheduler the Scheduler to subscribe on
       * @return the new Nono instance
       */
      public final Nono subscribeOn(Scheduler scheduler) {
          Objects.requireNonNull(scheduler, "scheduler is null");
          return onAssembly(new NonoSubscribeOn(this, scheduler));
      }
  
      /**
       * Observes the onError and onComplete events on the specified
       * Scheduler.
       * @param scheduler the Scheduler to emit terminal events on
       * @return the new Nono instance
       */
      public final Nono observeOn(Scheduler scheduler) {
          Objects.requireNonNull(scheduler, "scheduler is null");
          return onAssembly(new NonoObserveOn(this, scheduler));
      }
  
      /**
       * If the downstream cancels the sequence, the cancellation towards
       * the upstream will happen on the specified Scheduler.
       * @param scheduler the Scheduler to cancel on
       * @return the new Nono instance
       */
      public final Nono unsubscribeOn(Scheduler scheduler) {
          Objects.requireNonNull(scheduler, "scheduler is null");
          return onAssembly(new NonoUnsubscribeOn(this, scheduler));
      }
  
      /**
       * 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 Nolo instance
       */
      public final Nono doOnComplete(Action onComplete) {
          Objects.requireNonNull(onComplete, "onComplete is null");
          return onAssembly(new NonoDoOnLifecycle(this,
                  Functions.emptyConsumer(),
                  onComplete,
                  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 Nolo instance
       */
      public final Nono doOnError(Consumer<? super Throwable> onError) {
          Objects.requireNonNull(onError, "onError is null");
          return onAssembly(new NonoDoOnLifecycle(this,
                  onError,
                  Functions.EMPTY_ACTION,
                  Functions.EMPTY_ACTION,
                  Functions.emptyConsumer(),
                  Functions.EMPTY_LONG_CONSUMER,
                  Functions.EMPTY_ACTION
              ));
      }
  
      /**
       * Executes a callback when the upstream calls onSubscribe.
       * @param onSubscribe the consumer called with the upstream Subscription
       * @return the new Nolo instance
       */
      public final Nono doOnSubscribe(Consumer<? super Subscription> onSubscribe) {
          Objects.requireNonNull(onSubscribe, "onSubscribe is null");
          return onAssembly(new NonoDoOnLifecycle(this,
                  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 Nolo.
       * @param onRequest the callback called with the request amount
       * @return the new Nolo instance
       */
      public final Nono doOnRequest(LongConsumer onRequest) {
          Objects.requireNonNull(onRequest, "onRequest is null");
          return onAssembly(new NonoDoOnLifecycle(this,
                  Functions.emptyConsumer(),
                  Functions.EMPTY_ACTION,
                  Functions.EMPTY_ACTION,
                  Functions.emptyConsumer(),
                  onRequest,
                  Functions.EMPTY_ACTION
              ));
      }
  
      /**
       * Executes the callback after this Nono terminates and the downstream
       * is notified.
       * @param onAfterTerminate the callback to call after the downstream is notified
       * @return the new Nono instance
       */
      public final Nono doAfterTerminate(Action onAfterTerminate) {
          Objects.requireNonNull(onAfterTerminate, "onAfterTerminate is null");
          return onAssembly(new NonoDoOnLifecycle(this,
                  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 Nono instance
       */
      public final Nono doFinally(Action onFinally) {
          Objects.requireNonNull(onFinally, "action is null");
          return onAssembly(new NonoDoFinally(this, onFinally));
      }
  
      /**
       * Executes the callback if the downstream cancels the sequence.
       * @param onCancel the action to call
       * @return the new Nono instance
       */
      public final Nono doOnCancel(Action onCancel) {
          Objects.requireNonNull(onCancel, "action is null");
          return onAssembly(new NonoDoOnLifecycle(this,
                      Functions.emptyConsumer(),
                      Functions.EMPTY_ACTION,
                      Functions.EMPTY_ACTION,
                      Functions.emptyConsumer(),
                      Functions.EMPTY_LONG_CONSUMER,
                      onCancel
                  ));
      }
  
      /**
       * Repeatedly run this Nono indefinitely.
       * @return the new Nono instance
       */
      public final Nono repeat() {
          return onAssembly(new NonoRepeat(this, Long.MAX_VALUE));
      }
  
      /**
       * Repeatedly run this Nono at most the given number of times.
       * @param times the repeat count
       * @return the new Nono instance
       */
      public final Nono repeat(long times) {
          return onAssembly(new NonoRepeat(this, times));
      }
  
      /**
       * Repeat until the given BooleanSupplier returns true.
       * @param stop the boolean supplier to return null to stop repeating
       * @return the new Nono instance
       */
      public final Nono repeat(BooleanSupplier stop) {
          Objects.requireNonNull(stop, "stop is null");
          return onAssembly(new NonoRepeatUntil(this, stop));
      }
  
      /**
       * Repeat when the Publisher returned by the handler function signals
       * a value or terminate accordingly.
       * @param handler the Function that receives a Flowable that emits an object
       * when this Nono completes normally and should return a Publisher that if
       * signals a normal item, it triggers a resubscription to this Nono.
       * @return the new Nono instance
       */
      public final Nono repeatWhen(Function<? super Flowable<Object>, ? extends Publisher<?>> handler) {
          Objects.requireNonNull(handler, "handler is null");
          return onAssembly(new NonoRepeatWhen(this, handler));
      }
  
      /**
       * Repeatedly run this Nono indefinitely if it fails.
       * @return the new Nono instance
       */
      public final Nono retry() {
          return onAssembly(new NonoRetry(this, Long.MAX_VALUE));
      }
  
      /**
       * Repeatedly run this Nono at most the given number of times if it fails.
       * @param times the repeat count
       * @return the new Nono instance
       */
      public final Nono retry(long times) {
          return onAssembly(new NonoRetry(this, times));
      }
  
      /**
       * Retry a failed Nono if the predicate return true.
       * @param predicate the predicate receiving the failure Throwable and
       * returns true to trigger a retry.
       * @return the new Nono instance
       */
      public final Nono retry(Predicate<? super Throwable> predicate) {
          Objects.requireNonNull(predicate, "predicate is null");
          return onAssembly(new NonoRetryWhile(this, predicate));
      }
  
      /**
       * Retry this Nono when the Publisher returned by the handler function
       * signals a normal item or terminate if the Publisher terminates.
       * @param handler the Function that receives a Flowable of the failure Throwable
       * and returns a Publisher that if signals a normal item, it triggers a
       * resubscription.
       * @return the new Nono instance
       */
      public final Nono retryWhen(Function<? super Flowable<Throwable>, ? extends Publisher<?>> handler) {
          Objects.requireNonNull(handler, "handler is null");
          return onAssembly(new NonoRetryWhen(this, handler));
      }
  
      /**
       * Hides the identity of this Nono.
       * <p>
       * This also breaks optimizations such as operator fusion - useful
       * when diagnosing issues.
       * @return the new Nono instance
       */
      public final Nono hide() {
          return onAssembly(new NonoHide(this));
      }
  
      /**
       * Run this Nono and cancel it when the other Publisher signals
       * an item or completes.
       * @param other the other Publisher
       * @return the new Nono instance
       */
      public final Nono takeUntil(Publisher<?> other) {
          Objects.requireNonNull(other, "other is null");
          return onAssembly(new NonoTakeUntil(this, other));
      }
  
      /**
       * Caches the terminal event of the upstream Nono
       * and relays/replays it to Subscribers.
       * @return the new Nono instance
       *
       * @since 0.14.1
       */
      public final Nono cache() {
          return onAssembly(new NonoCache(this));
      }
  
      // -----------------------------------------------------------
      // Consumers and subscribers (leave)
      // -----------------------------------------------------------
  
      @Override
      public final void subscribe(Subscriber<? super Void> 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 signal the terminal events to the given subscriber.
       * @param s the downstream subscriber, not null
       */
      protected abstract void subscribeActual(Subscriber<? super Void> s);
  
      /**
       * Subscribe with the given subscriber and return the same subscriber, allowing
       * chaining methods on it or fluently reusing the instance.
       * @param <T> the target value type of the subscriber
       * @param <E> the subscriber's (sub)type
       * @param subscriber the subscriber to subscribe with, not null
       * @return the subscriber
       */
      @SchedulerSupport(SchedulerSupport.NONE)
      @SuppressWarnings("unchecked")
      public final <T, E extends Subscriber<T>> E subscribeWith(E subscriber) {
          subscribe((Subscriber<Object>)subscriber);
          return subscriber;
      }
  
      /**
       * Create a TestSubscriber, subscribe it to this Nono and return
       * the TestSubscriber itself.
       * @return the TestSubscriber created
       */
      @SchedulerSupport(SchedulerSupport.NONE)
      public final TestSubscriber<Void> test() {
          TestSubscriber<Void> ts = new TestSubscriber<>();
          subscribe(ts);
          return ts;
      }
  
      /**
       * Create a TestSubscriber, optionally cancel it, subscribe it to this Nono and return
       * the TestSubscriber itself.
       * @param cancelled shoud the TestSubscriber be cancelled before the subscription
       * @return the TestSubscriber created
       */
      @SchedulerSupport(SchedulerSupport.NONE)
      public final TestSubscriber<Void> test(boolean cancelled) {
          TestSubscriber<Void> ts = new TestSubscriber<>();
          if (cancelled) {
              ts.cancel();
          }
          subscribe(ts);
          return ts;
      }
  
      /**
       * Blockingly await indefinitely the termination of this Nono and return
       * the Throwable if this Nono terminated with an error, null
       * otherwise.
       * @return the Throwable error of the Nono, null if completed normally
       */
      @SchedulerSupport(SchedulerSupport.NONE)
      public final Throwable blockingAwait() {
          if (this instanceof Supplier) {
              try {
                  ((Supplier<?>)this).get();
                  return null;
              } catch (Throwable ex) {
                  Exceptions.throwIfFatal(ex);
                  return ex;
              }
          }
          NonoBlockingAwaitSubscriber s = new NonoBlockingAwaitSubscriber();
          subscribe(s);
          return s.blockingAwait();
      }
  
      /**
       * Blockingly await for the given timeout the termination of this Nono and return
       * the Throwable if this Nono terminated with an error, null
       * otherwise.
       * @param timeout the timeout value
       * @param unit the time unit
       * @return the Throwable error of the Nono, null if completed normally
       */
      @SchedulerSupport(SchedulerSupport.NONE)
      public final Throwable blockingAwait(long timeout, TimeUnit unit) {
          if (this instanceof Supplier) {
              try {
                  ((Supplier<?>)this).get();
                  return null;
              } catch (Throwable ex) {
                  Exceptions.throwIfFatal(ex);
                  return ex;
              }
          }
          Objects.requireNonNull(unit, "unit is null");
          NonoBlockingAwaitSubscriber s = new NonoBlockingAwaitSubscriber();
          subscribe(s);
          return s.blockingAwait(timeout, unit);
      }
  
      /**
       * Subscribe to this Nono and ignore the events it produces.
       * @return the Disposable to cancel the subscription
       * @since 0.13.0
       */
      public final Disposable subscribe() {
          return subscribe(Functions.EMPTY_ACTION, Functions.ERROR_CONSUMER);
      }
  
      /**
       * Subscribe to this Nono and execute the given action if this Nono
       * completes.
       * @param onComplete the callback Action to be called when this Nono
       * completes
       * @return the Disposable to cancel the subscription
       */
      @SchedulerSupport(SchedulerSupport.NONE)
      public final Disposable subscribe(Action onComplete) {
          return subscribe(onComplete, Functions.ERROR_CONSUMER);
      }
  
      /**
       * Subscribe to this Nono and execute the given action if this Nono
       * completes or call the consumer if this Nono terminates with an error.
       * @param onComplete the callback Action to be called when this Nono
       * completes
       * @param onError the callback Consumer to be called with the terminal
       * error.
       * @return the Disposable to cancel the subscription
       */
      @SchedulerSupport(SchedulerSupport.NONE)
      public final Disposable subscribe(Action onComplete, Consumer<? super Throwable> onError) {
          Objects.requireNonNull(onComplete, "onComplete is null");
          Objects.requireNonNull(onError, "onError is null");
          NonoLambdaSubscriber s = new NonoLambdaSubscriber(onComplete, onError);
          subscribe(s);
          return s;
      }
  
      /**
       * Block until this Nono terminates and ignore the actual events.
       * @since 0.13.0
       */
      public final void blockingSubscribe() {
          blockingSubscribe(Functions.EMPTY_ACTION, Functions.ERROR_CONSUMER);
      }
  
      /**
       * Block until this Nono completes and call the Action on the thread
       * where the blocking happens.
       * @param onComplete the Action to call when this Nono terminates
       */
      public final void blockingSubscribe(Action onComplete) {
          blockingSubscribe(onComplete, Functions.ERROR_CONSUMER);
      }
  
      /**
       * Block until this Nono terminates and call the Action or Consumer
       * depending on the terminal event on the thread where the blocking
       * happens.
       * @param onComplete the Action to call when this Nono completes
       * @param onError the Consumer to call when this Nono terminates with an error
       */
      public final void blockingSubscribe(Action onComplete, Consumer<? super Throwable> onError) {
          Objects.requireNonNull(onComplete, "onComplete is null");
          Objects.requireNonNull(onError, "onError is null");
          Throwable ex = blockingAwait();
          if (ex != null) {
              try {
                  onError.accept(ex);
              } catch (Throwable exc) {
                  Exceptions.throwIfFatal(exc);
                  RxJavaPlugins.onError(new CompositeException(ex, exc));
              }
          } else {
              try {
                  onComplete.run();
              } catch (Throwable exc) {
                  Exceptions.throwIfFatal(exc);
                  RxJavaPlugins.onError(exc);
              }
          }
      }
  
      /**
       * Converts this Nono into a Future and signals its single
       * value.
       * @return the new Future instance
       */
      public final Future<Void> toFuture() {
          FuturePerhapsSubscriber<Void> fs = new FuturePerhapsSubscriber<>();
          subscribe(fs);
          return fs;
      }
  }
  