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.operators;
  
  import java.util.*;
  import java.util.concurrent.TimeUnit;
  
  import org.reactivestreams.Publisher;
  
  import io.reactivex.rxjava3.annotations.*;
  import io.reactivex.rxjava3.core.*;
  import io.reactivex.rxjava3.functions.*;
  import io.reactivex.rxjava3.internal.functions.*;
  import io.reactivex.rxjava3.internal.schedulers.ImmediateThinScheduler;
  import io.reactivex.rxjava3.plugins.RxJavaPlugins;
  import io.reactivex.rxjava3.schedulers.Schedulers;
  
  /**
   * Utility class to create Flowable sources.
   */
  public final class Flowables {
      /** Utility class. */
      private Flowables() {
          throw new IllegalStateException("No instances!");
      }
  
      /**
       * Merges the source Publishers in an ordered fashion picking the smallest of the available value from
       * them (determined by their natural order).
       * @param <T> the value type of all sources
       * @param sources the iterable sequence of sources
       * @return the new Flowable instance
       * 
       * @since 0.8.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      @SafeVarargs
      public static <T extends Comparable<? super T>> Flowable<T> orderedMerge(Publisher<T>... sources) {
          return orderedMerge(Comparator.naturalOrder(), false, Flowable.bufferSize(), sources);
      }
  
      /**
       * Merges the source Publishers in an ordered fashion picking the smallest of the available value from
       * them (determined by their natural order) and allows delaying any error they may signal.
       * @param <T> the value type of all sources
       * @param sources the iterable sequence of sources
       * @param delayErrors if true, source errors are delayed until all sources terminate in some way
       * @return the new Flowable instance
       * 
       * @since 0.8.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      @SafeVarargs
      public static <T extends Comparable<? super T>> Flowable<T> orderedMerge(boolean delayErrors, Publisher<T>... sources) {
          return orderedMerge(Comparator.naturalOrder(), delayErrors, Flowable.bufferSize(), sources);
      }
  
      /**
       * Merges the source Publishers in an ordered fashion picking the smallest of the available value from
       * them (determined by their natural order), allows delaying any error they may signal and sets the prefetch
       * amount when requesting from these Publishers.
       * @param <T> the value type of all sources
       * @param sources the iterable sequence of sources
       * @param delayErrors if true, source errors are delayed until all sources terminate in some way
       * @param prefetch the number of items to prefetch from the sources
       * @return the new Flowable instance
       * 
       * @since 0.8.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      @SafeVarargs
      public static <T extends Comparable<? super T>> Flowable<T> orderedMerge(boolean delayErrors, int prefetch, Publisher<T>... sources) {
          return orderedMerge(Comparator.naturalOrder(), delayErrors, prefetch, sources);
      }
  
      /**
       * Merges the source Publishers in an ordered fashion picking the smallest of the available value from
       * them (determined by the Comparator).
       * @param <T> the value type of all sources
       * @param sources the iterable sequence of sources
       * @param comparator the comparator to use for comparing items;
       *                   it is called with the last known smallest in its first argument
       * @return the new Flowable instance
       * 
       * @since 0.8.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      @SafeVarargs
      public static <T> Flowable<T> orderedMerge(Comparator<? super T> comparator, Publisher<T>... sources) {
          return orderedMerge(comparator, false, Flowable.bufferSize(), sources);
      }
  
      /**
       * Merges the source Publishers in an ordered fashion picking the smallest of the available value from
       * them (determined by the Comparator) and allows delaying any error they may signal.
       * @param <T> the value type of all sources
       * @param sources the iterable sequence of sources
       * @param comparator the comparator to use for comparing items;
       *                   it is called with the last known smallest in its first argument
       * @param delayErrors if true, source errors are delayed until all sources terminate in some way
       * @return the new Flowable instance
       * 
       * @since 0.8.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      @SafeVarargs
      public static <T> Flowable<T> orderedMerge(Comparator<? super T> comparator, boolean delayErrors, Publisher<T>... sources) {
          return orderedMerge(comparator, delayErrors, Flowable.bufferSize(), sources);
      }
  
      /**
       * Merges the source Publishers in an ordered fashion picking the smallest of the available value from
       * them (determined by the Comparator), allows delaying any error they may signal and sets the prefetch
       * amount when requesting from these Publishers.
       * @param <T> the value type of all sources
       * @param sources the iterable sequence of sources
       * @param comparator the comparator to use for comparing items;
       *                   it is called with the last known smallest in its first argument
       * @param delayErrors if true, source errors are delayed until all sources terminate in some way
       * @param prefetch the number of items to prefetch from the sources
       * @return the new Flowable instance
       * 
       * @since 0.8.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      @SafeVarargs
      public static <T> Flowable<T> orderedMerge(Comparator<? super T> comparator, boolean delayErrors, int prefetch, Publisher<T>... sources) {
          Objects.requireNonNull(comparator, "comparator is null");
          Objects.requireNonNull(sources, "sources is null");
          ObjectHelper.verifyPositive(prefetch, "prefetch");
          return RxJavaPlugins.onAssembly(new FlowableOrderedMerge<>(sources, null, comparator, delayErrors, prefetch));
      }
  
      /**
       * Merges the source Publishers in an ordered fashion picking the smallest of the available value from
       * them (determined by the Comparator).
       * @param <T> the value type of all sources
       * @param sources the iterable sequence of sources
       * @param comparator the comparator to use for comparing items;
       *                   it is called with the last known smallest in its first argument
       * @return the new Flowable instance
       * 
       * @since 0.8.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      public static <T> Flowable<T> orderedMerge(Iterable<? extends Publisher<T>> sources, Comparator<? super T> comparator) {
          return orderedMerge(sources, comparator, false, Flowable.bufferSize());
      }
  
      /**
       * Merges the source Publishers in an ordered fashion picking the smallest of the available value from
       * them (determined by the Comparator) and allows delaying any error they may signal.
       * @param <T> the value type of all sources
       * @param sources the iterable sequence of sources
       * @param comparator the comparator to use for comparing items;
       *                   it is called with the last known smallest in its first argument
       * @param delayErrors if true, source errors are delayed until all sources terminate in some way
       * @return the new Flowable instance
       * 
       * @since 0.8.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      public static <T> Flowable<T> orderedMerge(Iterable<? extends Publisher<T>> sources, Comparator<? super T> comparator, boolean delayErrors) {
          return orderedMerge(sources, comparator, delayErrors, Flowable.bufferSize());
      }
  
      /**
       * Merges the source Publishers in an ordered fashion picking the smallest of the available value from
       * them (determined by the Comparator), allows delaying any error they may signal and sets the prefetch
       * amount when requesting from these Publishers.
       * @param <T> the value type of all sources
       * @param sources the iterable sequence of sources
       * @param comparator the comparator to use for comparing items;
       *                   it is called with the last known smallest in its first argument
       * @param delayErrors if true, source errors are delayed until all sources terminate in some way
       * @param prefetch the number of items to prefetch from the sources
       * @return the new Flowable instance
       * 
       * @since 0.8.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      public static <T> Flowable<T> orderedMerge(Iterable<? extends Publisher<T>> sources, Comparator<? super T> comparator, boolean delayErrors, int prefetch) {
          Objects.requireNonNull(comparator, "comparator is null");
          Objects.requireNonNull(sources, "sources is null");
          ObjectHelper.verifyPositive(prefetch, "prefetch");
          return RxJavaPlugins.onAssembly(new FlowableOrderedMerge<>(null, sources, comparator, delayErrors, prefetch));
      }
  
      /**
       * Merges the source Publishers in an ordered fashion picking the smallest of the available value from
       * them (determined by their natural order).
       * @param <T> the value type of all sources
       * @param sources the iterable sequence of sources
       * @return the new Flowable instance
       * 
       * @since 0.8.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      public static <T extends Comparable<? super T>> Flowable<T> orderedMerge(Iterable<? extends Publisher<T>> sources) {
          return orderedMerge(sources, Comparator.naturalOrder(), false, Flowable.bufferSize());
      }
  
      /**
       * Merges the source Publishers in an ordered fashion picking the smallest of the available value from
       * them (determined by their natural order) and allows delaying any error they may signal.
       * @param <T> the value type of all sources
       * @param sources the iterable sequence of sources
       * @param delayErrors if true, source errors are delayed until all sources terminate in some way
       * @return the new Flowable instance
       * 
       * @since 0.8.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      public static <T extends Comparable<? super T>> Flowable<T> orderedMerge(Iterable<? extends Publisher<T>> sources, boolean delayErrors) {
          return orderedMerge(sources, Comparator.naturalOrder(), delayErrors, Flowable.bufferSize());
      }
  
      /**
       * Merges the source Publishers in an ordered fashion picking the smallest of the available value from
       * them (determined by their natural order), allows delaying any error they may signal and sets the prefetch
       * amount when requesting from these Publishers.
       * @param <T> the value type of all sources
       * @param sources the iterable sequence of sources
       * @param delayErrors if true, source errors are delayed until all sources terminate in some way
       * @param prefetch the number of items to prefetch from the sources
       * @return the new Flowable instance
       * 
       * @since 0.8.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      public static <T extends Comparable<? super T>> Flowable<T> orderedMerge(Iterable<? extends Publisher<T>> sources, boolean delayErrors, int prefetch) {
          return orderedMerge(sources, Comparator.naturalOrder(), delayErrors, prefetch);
      }
  
      /**
       * Repeats a scalar value indefinitely.
       * @param <T> the value type
       * @param item the value to repeat
       * @return the new Flowable instance
       * 
       * @since 0.14.2
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      public static <T> Flowable<T> repeat(T item) {
          Objects.requireNonNull(item, "item is null");
          return RxJavaPlugins.onAssembly(new FlowableRepeatScalar<>(item));
      }
  
      /**
       * Repeatedly calls the given Supplier to produce items indefinitely.
       * @param <T> the value type
       * @param supplier the Supplier to call
       * @return the new Flowable instance
       * 
       * @since 0.14.2
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      public static <T> Flowable<T> repeatSupplier(Supplier<T> supplier) {
          Objects.requireNonNull(supplier, "supplier is null");
          return RxJavaPlugins.onAssembly(new FlowableRepeatSupplier<>(supplier));
      }
  
      /**
       * Periodically tries to emit an ever increasing long value or
       * buffers (efficiently) such emissions until the downstream requests.
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors the backpressure of the downstream and
       *  no emission is lost, however, the timing of the reception of the
       *  values is now dependent on the downstream backpressure.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>The operator uses the {@code computation} {@link Scheduler} to time
       *  the emission and likely deliver the value (unless backpressured).</dd>
       * </dl>
       * 
       * @param period the emission period (including the delay for the first emission)
       * @param unit the emission time unit
       * @return the new Flowable instance
       * 
       * @since 0.15.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.COMPUTATION)
      public static Flowable<Long> intervalBackpressure(long period, TimeUnit unit) {
          return intervalBackpressure(period, period, unit, Schedulers.computation());
      }
  
      /**
       * Periodically tries to emit an ever increasing long value or
       * buffers (efficiently) such emissions until the downstream requests.
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors the backpressure of the downstream and
       *  no emission is lost, however, the timing of the reception of the
       *  values is now dependent on the downstream backpressure.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>The operator uses the {@code computation} {@link Scheduler} to time
       *  the emission and likely deliver the value (unless backpressured).</dd>
       * </dl>
       * 
       * @param period the emission period (including the delay for the first emission)
       * @param unit the emission time unit
       * @param scheduler the scheduler to use for timing and likely emitting items
       * @return the new Flowable instance
       * 
       * @since 0.15.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.CUSTOM)
      public static Flowable<Long> intervalBackpressure(long period, TimeUnit unit, Scheduler scheduler) {
          return intervalBackpressure(period, period, unit, scheduler);
      }
  
      /**
       * Periodically tries to emit an ever increasing long value or
       * buffers (efficiently) such emissions until the downstream requests.
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors the backpressure of the downstream and
       *  no emission is lost, however, the timing of the reception of the
       *  values is now dependent on the downstream backpressure.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>The operator uses the {@code computation} {@link Scheduler} to time
       *  the emission and likely deliver the value (unless backpressured).</dd>
       * </dl>
       * 
       * @param initialDelay the initial delay before emitting the first 0L
       * @param period the emission period after the first emission
       * @param unit the emission time unit
       * @return the new Flowable instance
       * 
       * @since 0.15.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.COMPUTATION)
      public static Flowable<Long> intervalBackpressure(long initialDelay, long period, TimeUnit unit) {
          return intervalBackpressure(initialDelay, period, unit, Schedulers.computation());
      }
  
      /**
       * Periodically tries to emit an ever increasing long value or
       * buffers (efficiently) such emissions until the downstream requests.
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors the backpressure of the downstream and
       *  no emission is lost, however, the timing of the reception of the
       *  values is now dependent on the downstream backpressure.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>The operator uses the {@link Scheduler} provided to time
       *  the emission and likely deliver the value (unless backpressured).</dd>
       * </dl>
       * 
       * @param initialDelay the initial delay before emitting the first 0L
       * @param period the emission period (including the delay for the first emission)
       * @param unit the emission time unit
       * @param scheduler the scheduler to use for timing and likely emitting items
       * @return the new Flowable instance
       * 
       * @since 0.15.0
       */
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.CUSTOM)
      public static Flowable<Long> intervalBackpressure(long initialDelay, long period, TimeUnit unit, Scheduler scheduler) {
          return RxJavaPlugins.onAssembly(new FlowableIntervalBackpressure(initialDelay, period, unit, scheduler));
      }
  
      /**
       * Zips the latest available values of the source Publishers via a combiner function where the
       * emission rate is determined by the slowest Publisher and the downstream consumption rate.
       * <p>
       * Non-consumed source values are overwritten by newer values. Unlike {@code combineLatest}, source
       * values are not reused to form new combinations.
       * <p>
       * If any of the sources runs out of items, the other sources are cancelled and the sequence completes.
       * <pre><code>
       * A: ---o-o-o------o-o----o---o-|-------
       * B: ---------x-x--x-------x-----x--x---
       * ======= zipLatest (o, x -&gt; M) ========
       * R: ---------M----M-------M-----M|-----
       * </code></pre>
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors the backpressure of the downstream and consumes
       *  the source Publishers in an unbounded manner, keeping only their latest values temporarily.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>The operator doesn't run on any particular {@link Scheduler}
       *  and the combined item emission happens on the thread that won the internal emission-right race.</dd>
       * </dl>
       * 
       * @param <T> the common source value type
       * @param <R> the result type
       * @param combiner the function receiving the latest values of the sources and returns a value
       *                 to be emitted to the downstream.
       * @param sources the array of source Publishers to zip/combine
       * @return the new Flowable instance.
       */
      @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
      @SchedulerSupport(SchedulerSupport.NONE)
      @SafeVarargs
      public static <T, R> Flowable<R> zipLatest(Function<? super Object[], ? extends R> combiner, Publisher<? extends T>... sources) {
          return zipLatest(combiner, ImmediateThinScheduler.INSTANCE, sources);
      }
  
      /**
       * Zips the latest available values of the source Publishers via a combiner function where the
       * emission rate is determined by the slowest Publisher and the downstream consumption rate.
       * <p>
       * Non-consumed source values are overwritten by newer values. Unlike {@code combineLatest}, source
       * values are not reused to form new combinations.
       * <p>
       * If any of the sources runs out of items, the other sources are cancelled and the sequence completes.
       * <pre><code>
       * A: ---o-o-o------o-o----o---o-|-------
       * B: ---------x-x--x-------x-----x--x---
       * ======= zipLatest (o, x -&gt; M) ========
       * R: ---------M----M-------M-----M|-----
       * </code></pre>
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors the backpressure of the downstream and consumes
       *  the source Publishers in an unbounded manner, keeping only their latest values temporarily.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>The operator emits the combined items on the {@link Scheduler} provided.</dd>
       * </dl>
       * 
       * @param <T> the common source value type
       * @param <R> the result type
       * @param combiner the function receiving the latest values of the sources and returns a value
       *                 to be emitted to the downstream.
       * @param sources the array of source Publishers to zip/combine
       * @param scheduler the Scheduler to use for emitting items and/or terminal signals
       * @return the new Flowable instance.
       */
      @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
      @SchedulerSupport(SchedulerSupport.CUSTOM)
      @SafeVarargs
      public static <T, R> Flowable<R> zipLatest(Function<? super Object[], ? extends R> combiner, Scheduler scheduler, Publisher<? extends T>... sources) {
          Objects.requireNonNull(combiner, "combiner is null");
          Objects.requireNonNull(scheduler, "scheduler is null");
          Objects.requireNonNull(sources, "sources is null");
          return RxJavaPlugins.onAssembly(new FlowableZipLatest<T, R>(sources, null, combiner, scheduler));
      }
  
      /**
       * Zips the latest available values of the source Publishers via a combiner function where the
       * emission rate is determined by the slowest Publisher and the downstream consumption rate.
       * <p>
       * Non-consumed source values are overwritten by newer values. Unlike {@code combineLatest}, source
       * values are not reused to form new combinations.
       * <p>
       * If any of the sources runs out of items, the other sources are cancelled and the sequence completes.
       * <pre><code>
       * A: ---o-o-o------o-o----o---o-|-------
       * B: ---------x-x--x-------x-----x--x---
       * ======= zipLatest (o, x -&gt; M) ========
       * R: ---------M----M-------M-----M|-----
       * </code></pre>
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors the backpressure of the downstream and consumes
       *  the source Publishers in an unbounded manner, keeping only their latest values temporarily.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>The operator doesn't run on any particular {@link Scheduler} scheduler
       *  and the combined item emission happens on the thread that won the internal emission-right race.</dd>
       * </dl>
       * 
       * @param <T> the common source value type
       * @param <R> the result type
       * @param combiner the function receiving the latest values of the sources and returns a value
       *                 to be emitted to the downstream.
       * @param sources the array of source Publishers to zip/combine
       * @return the new Flowable instance.
       */
      @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
      @SchedulerSupport(SchedulerSupport.NONE)
      public static <T, R> Flowable<R> zipLatest(Iterable<? extends Publisher<? extends T>> sources, Function<? super Object[], ? extends R> combiner) {
          return zipLatest(sources, combiner, ImmediateThinScheduler.INSTANCE);
      }
  
      /**
       * Zips the latest available values of the source Publishers via a combiner function where the
       * emission rate is determined by the slowest Publisher and the downstream consumption rate.
       * <p>
       * Non-consumed source values are overwritten by newer values. Unlike {@code combineLatest}, source
       * values are not reused to form new combinations.
       * <p>
       * If any of the sources runs out of items, the other sources are cancelled and the sequence completes.
       * <pre><code>
       * A: ---o-o-o------o-o----o---o-|-------
       * B: ---------x-x--x-------x-----x--x---
       * ======= zipLatest (o, x -&gt; M) ========
       * R: ---------M----M-------M-----M|-----
       * </code></pre>
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors the backpressure of the downstream and consumes
       *  the source Publishers in an unbounded manner, keeping only their latest values temporarily.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>The operator emits the combined items on the {@link Scheduler} provided.</dd>
       * </dl>
       * 
       * @param <T> the common source value type
       * @param <R> the result type
       * @param combiner the function receiving the latest values of the sources and returns a value
       *                 to be emitted to the downstream.
       * @param sources the array of source Publishers to zip/combine
       * @param scheduler the Scheduler to use for emitting items and/or terminal signals
       * @return the new Flowable instance.
       */
      @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
      @SchedulerSupport(SchedulerSupport.CUSTOM)
      public static <T, R> Flowable<R> zipLatest(Iterable<? extends Publisher<? extends T>> sources, Function<? super Object[], ? extends R> combiner, Scheduler scheduler) {
          Objects.requireNonNull(sources, "sources is null");
          Objects.requireNonNull(combiner, "combiner is null");
          Objects.requireNonNull(scheduler, "scheduler is null");
          return RxJavaPlugins.onAssembly(new FlowableZipLatest<T, R>(null, sources, combiner, scheduler));
      }
  
      /**
       * Zips the latest available values of the source Publishers via a combiner function where the
       * emission rate is determined by the slowest Publisher and the downstream consumption rate.
       * <p>
       * Non-consumed source values are overwritten by newer values. Unlike {@code combineLatest}, source
       * values are not reused to form new combinations.
       * <p>
       * If any of the sources runs out of items, the other sources are cancelled and the sequence completes.
       * <pre><code>
       * A: ---o-o-o------o-o----o---o-|-------
       * B: ---------x-x--x-------x-----x--x---
       * ======= zipLatest (o, x -&gt; M) ========
       * R: ---------M----M-------M-----M|-----
       * </code></pre>
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors the backpressure of the downstream and consumes
       *  the source Publishers in an unbounded manner, keeping only their latest values temporarily.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>The operator doesn't run on any particular {@link Scheduler} scheduler
       *  and the combined item emission happens on the thread that won the internal emission-right race.</dd>
       * </dl>
       * 
       * @param <T1> the value type of the first source Publisher
       * @param <T2> the value type of the second source Publisher
       * @param <R> the result type
       * @param combiner the function receiving the latest values of the sources and returns a value
       *                 to be emitted to the downstream.
       * @param source1 the first source Publisher instance
       * @param source2 the second source Publisher instance
       * @return the new Flowable instance.
       */
      @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
      @SchedulerSupport(SchedulerSupport.NONE)
      public static <T1, T2, R> Flowable<R> zipLatest(Publisher<T1> source1, Publisher<T2> source2, BiFunction<? super T1, ? super T2, ? extends R> combiner) {
          return zipLatest(source1, source2, combiner, ImmediateThinScheduler.INSTANCE);
      }
  
      /**
       * Zips the latest available values of the source Publishers via a combiner function where the
       * emission rate is determined by the slowest Publisher and the downstream consumption rate.
       * <p>
       * Non-consumed source values are overwritten by newer values. Unlike {@code combineLatest}, source
       * values are not reused to form new combinations.
       * <p>
       * If any of the sources runs out of items, the other sources are cancelled and the sequence completes.
       * <pre><code>
       * A: ---o-o-o------o-o----o---o-|-------
       * B: ---------x-x--x-------x-----x--x---
       * ======= zipLatest (o, x -&gt; M) ========
       * R: ---------M----M-------M-----M|-----
       * </code></pre>
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors the backpressure of the downstream and consumes
       *  the source Publishers in an unbounded manner, keeping only their latest values temporarily.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>The operator emits the combined items on the {@link Scheduler} provided.</dd>
       * </dl>
       * 
       * @param <T1> the value type of the first source Publisher
       * @param <T2> the value type of the second source Publisher
       * @param <R> the result type
       * @param combiner the function receiving the latest values of the sources and returns a value
       *                 to be emitted to the downstream.
       * @param source1 the first source Publisher instance
       * @param source2 the second source Publisher instance
       * @param scheduler the Scheduler to use for emitting items and/or terminal signals
       * @return the new Flowable instance.
       */
      @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
      @SchedulerSupport(SchedulerSupport.CUSTOM)
      @SuppressWarnings("unchecked")
      public static <T1, T2, R> Flowable<R> zipLatest(Publisher<T1> source1, Publisher<T2> source2, BiFunction<? super T1, ? super T2, ? extends R> combiner, Scheduler scheduler) {
          Objects.requireNonNull(source1, "source1 is null");
          Objects.requireNonNull(source2, "source2 is null");
          Objects.requireNonNull(combiner, "combiner is null");
          Objects.requireNonNull(scheduler, "scheduler is null");
          return RxJavaPlugins.onAssembly(new FlowableZipLatest<Object, R>(
                  new Publisher[] { source1, source2 }, null,
                  Functions.toFunction(combiner), scheduler));
      }
  
      /**
       * Zips the latest available values of the source Publishers via a combiner function where the
       * emission rate is determined by the slowest Publisher and the downstream consumption rate.
       * <p>
       * Non-consumed source values are overwritten by newer values. Unlike {@code combineLatest}, source
       * values are not reused to form new combinations.
       * <p>
       * If any of the sources runs out of items, the other sources are cancelled and the sequence completes.
       * <pre><code>
       * A: ---o-o-o------o-o----o---o-|-------
       * B: ---------x-x--x-------x-----x--x---
       * ======= zipLatest (o, x -&gt; M) ========
       * R: ---------M----M-------M-----M|-----
       * </code></pre>
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors the backpressure of the downstream and consumes
       *  the source Publishers in an unbounded manner, keeping only their latest values temporarily.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>The operator doesn't run on any particular {@link Scheduler} scheduler
       *  and the combined item emission happens on the thread that won the internal emission-right race.</dd>
       * </dl>
       * 
       * @param <T1> the value type of the first source Publisher
       * @param <T2> the value type of the second source Publisher
       * @param <T3> the value type of the third source Publisher
       * @param <R> the result type
       * @param combiner the function receiving the latest values of the sources and returns a value
       *                 to be emitted to the downstream.
       * @param source1 the first source Publisher instance
       * @param source2 the second source Publisher instance
       * @param source3 the third source Publisher instance
       * @return the new Flowable instance.
       */
      @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
      @SchedulerSupport(SchedulerSupport.NONE)
      public static <T1, T2, T3, R> Flowable<R> zipLatest(Publisher<T1> source1, Publisher<T2> source2,
              Publisher<T3> source3, Function3<? super T1, ? super T2, ? super T3, ? extends R> combiner) {
          return zipLatest(source1, source2, source3, combiner, ImmediateThinScheduler.INSTANCE);
      }
  
      /**
       * Zips the latest available values of the source Publishers via a combiner function where the
       * emission rate is determined by the slowest Publisher and the downstream consumption rate.
       * <p>
       * Non-consumed source values are overwritten by newer values. Unlike {@code combineLatest}, source
       * values are not reused to form new combinations.
       * <p>
       * If any of the sources runs out of items, the other sources are cancelled and the sequence completes.
       * <pre><code>
       * A: ---o-o-o------o-o----o---o-|-------
       * B: ---------x-x--x-------x-----x--x---
       * ======= zipLatest (o, x -&gt; M) ========
       * R: ---------M----M-------M-----M|-----
       * </code></pre>
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors the backpressure of the downstream and consumes
       *  the source Publishers in an unbounded manner, keeping only their latest values temporarily.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>The operator emits the combined items on the {@link Scheduler} provided.</dd>
       * </dl>
       * 
       * @param <T1> the value type of the first source Publisher
       * @param <T2> the value type of the second source Publisher
       * @param <T3> the value type of the third source Publisher
       * @param <R> the result type
       * @param combiner the function receiving the latest values of the sources and returns a value
       *                 to be emitted to the downstream.
       * @param source1 the first source Publisher instance
       * @param source2 the second source Publisher instance
       * @param source3 the third source Publisher instance
       * @param scheduler the Scheduler to use for emitting items and/or terminal signals
       * @return the new Flowable instance.
       */
      @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
      @SchedulerSupport(SchedulerSupport.CUSTOM)
      @SuppressWarnings("unchecked")
      public static <T1, T2, T3, R> Flowable<R> zipLatest(Publisher<T1> source1, Publisher<T2> source2,
              Publisher<T3> source3, Function3<? super T1, ? super T2, ? super T3, ? extends R> combiner,
              Scheduler scheduler) {
          Objects.requireNonNull(source1, "source1 is null");
          Objects.requireNonNull(source2, "source2 is null");
          Objects.requireNonNull(source3, "source3 is null");
          Objects.requireNonNull(combiner, "combiner is null");
          Objects.requireNonNull(scheduler, "scheduler is null");
          return RxJavaPlugins.onAssembly(new FlowableZipLatest<Object, R>(
                  new Publisher[] { source1, source2, source3 }, null,
                  Functions.toFunction(combiner), scheduler));
      }
  
      /**
       * Zips the latest available values of the source Publishers via a combiner function where the
       * emission rate is determined by the slowest Publisher and the downstream consumption rate.
       * <p>
       * Non-consumed source values are overwritten by newer values. Unlike {@code combineLatest}, source
       * values are not reused to form new combinations.
       * <p>
       * If any of the sources runs out of items, the other sources are cancelled and the sequence completes.
       * <pre><code>
       * A: ---o-o-o------o-o----o---o-|-------
       * B: ---------x-x--x-------x-----x--x---
       * ======= zipLatest (o, x -&gt; M) ========
       * R: ---------M----M-------M-----M|-----
       * </code></pre>
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors the backpressure of the downstream and consumes
       *  the source Publishers in an unbounded manner, keeping only their latest values temporarily.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>The operator doesn't run on any particular {@link Scheduler} scheduler
       *  and the combined item emission happens on the thread that won the internal emission-right race.</dd>
       * </dl>
       * 
       * @param <T1> the value type of the first source Publisher
       * @param <T2> the value type of the second source Publisher
       * @param <T3> the value type of the third source Publisher
       * @param <T4> the value type of the fourth source Publisher
       * @param <R> the result type
       * @param combiner the function receiving the latest values of the sources and returns a value
       *                 to be emitted to the downstream.
       * @param source1 the first source Publisher instance
       * @param source2 the second source Publisher instance
       * @param source3 the third source Publisher instance
       * @param source4 the fourth source Publisher instance
       * @return the new Flowable instance.
       */
      @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
      @SchedulerSupport(SchedulerSupport.NONE)
      public static <T1, T2, T3, T4, R> Flowable<R> zipLatest(Publisher<T1> source1, Publisher<T2> source2,
              Publisher<T3> source3, Publisher<T4> source4,
              Function4<? super T1, ? super T2, ? super T3, ? super T4, ? extends R> combiner) {
          return zipLatest(source1, source2, source3, source4, combiner, ImmediateThinScheduler.INSTANCE);
      }
  
      /**
       * Zips the latest available values of the source Publishers via a combiner function where the
       * emission rate is determined by the slowest Publisher and the downstream consumption rate.
       * <p>
       * Non-consumed source values are overwritten by newer values. Unlike {@code combineLatest}, source
       * values are not reused to form new combinations.
       * <p>
       * If any of the sources runs out of items, the other sources are cancelled and the sequence completes.
       * <pre><code>
       * A: ---o-o-o------o-o----o---o-|-------
       * B: ---------x-x--x-------x-----x--x---
       * ======= zipLatest (o, x -&gt; M) ========
       * R: ---------M----M-------M-----M|-----
       * </code></pre>
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors the backpressure of the downstream and consumes
       *  the source Publishers in an unbounded manner, keeping only their latest values temporarily.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>The operator emits the combined items on the {@link Scheduler} provided.</dd>
       * </dl>
       * 
       * @param <T1> the value type of the first source Publisher
       * @param <T2> the value type of the second source Publisher
       * @param <T3> the value type of the third source Publisher
       * @param <T4> the value type of the fourth source Publisher
       * @param <R> the result type
       * @param combiner the function receiving the latest values of the sources and returns a value
       *                 to be emitted to the downstream.
       * @param source1 the first source Publisher instance
       * @param source2 the second source Publisher instance
       * @param source3 the third source Publisher instance
       * @param source4 the fourth source Publisher instance
       * @param scheduler the Scheduler to use for emitting items and/or terminal signals
       * @return the new Flowable instance.
       */
      @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
      @SchedulerSupport(SchedulerSupport.CUSTOM)
      @SuppressWarnings("unchecked")
      public static <T1, T2, T3, T4, R> Flowable<R> zipLatest(Publisher<T1> source1, Publisher<T2> source2,
              Publisher<T3> source3, Publisher<T4> source4,
              Function4<? super T1, ? super T2, ? super T3, ? super T4, ? extends R> combiner,
              Scheduler scheduler) {
          Objects.requireNonNull(source1, "source1 is null");
          Objects.requireNonNull(source2, "source2 is null");
          Objects.requireNonNull(source3, "source3 is null");
          Objects.requireNonNull(source4, "source4 is null");
          Objects.requireNonNull(combiner, "combiner is null");
          Objects.requireNonNull(scheduler, "scheduler is null");
          return RxJavaPlugins.onAssembly(new FlowableZipLatest<Object, R>(
                  new Publisher[] { source1, source2, source3, source4 }, null,
                  Functions.toFunction(combiner), scheduler));
      }
      /**
       * Generates items by invoking a callback, for each downstream request one by one, that sets up an
       * asynchronous call to some API that eventually responds with an item, an error or termination, while
       * making sure there is only one such outstanding API call in progress and honoring the
       * backpressure of the downstream.
       * <p>
       * This operator allows the bridging of the asynchronous and backpressurable world with the reactive world,
       * where backpressure is the emergent effect of making sure there is only one outstanding API call
       * at a time which responds with at most one item per invocation.
       * <p>
       * Note that the implementation may have one outstanding API call even if the downstream hasn't requested more
       * and as such, the resulting item may get cached until the downstream requests for more.
       * <p>
       * During the async response, the invocation protocol of the {@link FlowableAsyncEmitter} should be as follows:
       * <pre><code>
       *     (onNext | onNothing)? (onError | onComplete)?
       * </code></pre>
       * In words, an {@code onNext} or {@code onNothing} (which indicates this particular API call resulted in no
       * items and the next API call can proceed) may be followed by a terminal event.
       * <p>
       * The methods {@link FlowableAsyncEmitter#onNext(Object)}, {@link FlowableAsyncEmitter#onError(Throwable)},
       * {@link FlowableAsyncEmitter#onComplete()} and {@link FlowableAsyncEmitter#onNothing()} should not be called
       * concurrently with each other or outside the context of the generator. The rest of the
       * {@link FlowableAsyncEmitter} methods are thread-safe.
       * <p>
       * <b>Example:</b><br>
       * Let's assume there is an async API with the following interface definition:
       * <pre><code>
       * interface AsyncAPI&lt;T&gt; extends AutoCloseable {
       *
       *     CompletableFuture&lt;Void&gt; nextValue(Consumer&lt;? super T&gt; onValue);
       *
       * }
       * </code></pre>
       * When the call succeeds, the {@code onValue} is invoked with it. If there are no more items, the
       * {@code CompletableFuture} returned by the last {@code nextValue} is completed (with null).
       * If there is an error, the same {@code CompletableFuture} is completed exceptionally. Each
       * {@code nextValue} invocation creates a fresh {@code CompletableFuture} which can be cancelled
       * if necesary. {@code nextValue} should not be invoked again until the {@code onValue} callback
       * has been notified.<br>
       * An instance of this API can be obtained on demand, thus the state of this operator consists of the
       * {@code AsyncAPI} instance supplied for each individual {@code Subscriber}. The API can be transformed into
       * a {@code Flowable} as follows:
       * <pre><code>
       * Flowable&lt;Integer&gt; source = Flowable.&lt;Integer, AsyncAPI&lt;Integer&gt;&gt;generateAsync(
       *
       *     // create a fresh API instance for each individual Subscriber
       *     () -&gt; new AsyncAPIImpl&lt;Integer&gt;(),
       *
       *     // this BiFunction will be called once the operator is ready to receive the next item
       *     // and will invoke it again only when that item is delivered via emitter.onNext()
       *     (state, emitter) -&gt; {
       *
       *         // issue the async API call
       *         CompletableFuture&lt;Void&gt; f = state.nextValue(
       *
       *             // handle the value received
       *             value -&gt; {
       *
       *                 // we have the option to signal that item
       *                 if (value % 2 == 0) {
       *                     emitter.onNext(value);
       *                 } else if (value == 101) {
       *                     // or stop altogether, which will also trigger a cleanup
       *                     emitter.onComplete();
       *                 } else {
       *                     // or drop it and have the operator start a new call
       *                     emitter.onNothing();
       *                 }
       *             }
       *         );
       *
       *         // This API call may not produce further items or fail
       *         f.whenComplete((done, error) -&gt; {
       *             // As per the CompletableFuture API, error != null is the error outcome,
       *             // done is always null due to the Void type
       *             if (error != null) {
       *                 emitter.onError(error);
       *             } else {
       *                 emitter.onComplete();
       *             }
       *         });
       *
       *         // In case the downstream cancels, the current API call
       *         // should be cancelled as well
       *         emitter.replaceCancellable(() -&gt; f.cancel(true));
       *
       *         // some sources may want to create a fresh state object
       *         // after each invocation of this generator
       *         return state;
       *     },
       *
       *     // cleanup the state object
       *     state -&gt; { state.close(); }
       * )
       * </code></pre>
       * <dl>
       *  <dt><b>Backpressure:</b></dt>
       *  <dd>The operator honors downstream backpressure.</dd>
       *  <dt><b>Scheduler:</b></dt>
       *  <dd>{@code generateAsync} does not operate by default on a particular {@link Scheduler}, however,
       *  the signals emitted through the {@link FlowableAsyncEmitter} may happen on any thread,
       *  depending on the asynchronous API.</dd>
       * </dl>
       * @param <T> the generated item type
       * @param <S> the state associated with an individual subscription.
       * @param initialState the {@link Supplier} that returns a state object for each individual
       *                     {@link org.reactivestreams.Subscriber Subscriber} to the returned {@code Flowable}.
       * @param asyncGenerator the {@link BiFunction} called with the current state value and the
       *                       {@link FlowableAsyncEmitter} object and should return a new state value
       *                       as well as prepare and issue the async API call in a way that
       *                       the call's outcome is (eventually) converted into {@code onNext}, {@code onError} or
       *                       {@code onComplete} calls. The operator ensures the {@code BiFunction} is
       *                       only invoked when the previous async call produced {@code onNext} item and
       *                       this item has been delivered to the downstream.
       * @param stateCleanup called at most once with the current state object to allow cleaning it up after
       *                     the flow is cancelled or terminates via {@link FlowableAsyncEmitter#onError(Throwable)}
       *                     or {@link FlowableAsyncEmitter#onComplete()}.
       * @return the new Flowable instance
       * @see Flowable#generate(Supplier, BiFunction, Consumer)
       * @see FlowableAsyncEmitter
       * @since 0.18.9
       */
      @CheckReturnValue
      @BackpressureSupport(BackpressureKind.FULL)
      @SchedulerSupport(SchedulerSupport.NONE)
      public static <T, S> Flowable<T> generateAsync(Supplier<S> initialState, BiFunction<S, FlowableAsyncEmitter<T>, S> asyncGenerator, Consumer<? super S> stateCleanup) {
          Objects.requireNonNull(initialState, "initialState is null");
          Objects.requireNonNull(asyncGenerator, "asyncGenerator is null");
          Objects.requireNonNull(stateCleanup, "stateCleanup is null");
          return RxJavaPlugins.onAssembly(new FlowableGenerateAsync<>(initialState, asyncGenerator, stateCleanup));
      }
  }
  