/**
* Copyright 2016 Netflix, Inc.
*
* 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 io.reactivex;
import java.util.*;
import java.util.concurrent.*;
import org.reactivestreams.*;
import io.reactivex.annotations.*;
import io.reactivex.disposables.Disposable;
import io.reactivex.exceptions.Exceptions;
import io.reactivex.flowables.*;
import io.reactivex.functions.*;
import io.reactivex.internal.functions.Functions;
import io.reactivex.internal.functions.ObjectHelper;
import io.reactivex.internal.fuseable.*;
import io.reactivex.internal.operators.completable.CompletableFromPublisher;
import io.reactivex.internal.operators.flowable.*;
import io.reactivex.internal.operators.maybe.MaybeFromPublisher;
import io.reactivex.internal.operators.observable.ObservableFromPublisher;
import io.reactivex.internal.operators.single.SingleFromPublisher;
import io.reactivex.internal.schedulers.ImmediateThinScheduler;
import io.reactivex.internal.subscribers.flowable.*;
import io.reactivex.internal.util.*;
import io.reactivex.plugins.RxJavaPlugins;
import io.reactivex.schedulers.*;
import io.reactivex.subscribers.*;
/**
* The Flowable class that implements the Reactive-Streams Pattern and offers factory methods,
* intermediate operators and the ability to consume reactive dataflows.
*
* Reactive-Streams operates with {@code Publisher}s which {@code Flowable} extends. Many operators
* therefore accept general {@code Publisher}s directly and allow direct interoperation with other
* Reactive-Streams implementations.
*
* The Flowable hosts the default buffer size of 128 elements for operators, accessible via {@link #bufferSize()},
* that can be overridden globally via the system parameter {@code rx2.buffer-size}. Most operators, however, have
* overloads that allow setting their internal buffer size explicitly.
*
* The documentation for this class makes use of marble diagrams. The following legend explains these diagrams:
*
* 
*
* For more information see the ReactiveX
* documentation.
*
* @param
* the type of the items emitted by the Flowable
*/
public abstract class Flowable implements Publisher {
/** The default buffer size. */
static final int BUFFER_SIZE;
static {
BUFFER_SIZE = Math.max(16, Integer.getInteger("rx2.buffer-size", 128));
}
/**
* Mirrors the one Publisher in an Iterable of several Publishers that first either emits an item or sends
* a termination notification.
*
* 
*
* - Backpressure:
* - The operator itself doesn't interfere with backpressure which is determined by the winning
* {@code Publisher}'s backpressure behavior.
* - Scheduler:
* - {@code amb} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the common element type
* @param sources
* an Iterable of Publishers sources competing to react first
* @return a Flowable that emits the same sequence as whichever of the source Publishers first
* emitted an item or sent a termination notification
* @see ReactiveX operators documentation: Amb
*/
@BackpressureSupport(BackpressureKind.PASS_THROUGH)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable amb(Iterable> sources) {
ObjectHelper.requireNonNull(sources, "sources is null");
return RxJavaPlugins.onAssembly(new FlowableAmb(null, sources));
}
/**
* Mirrors the one Publisher in an array of several Publishers that first either emits an item or sends
* a termination notification.
*
*
*
* - Backpressure:
* - The operator itself doesn't interfere with backpressure which is determined by the winning
* {@code Publisher}'s backpressure behavior.
* - Scheduler:
* - {@code amb} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the common element type
* @param sources
* an array of Publisher sources competing to react first
* @return a Flowable that emits the same sequence as whichever of the source Publishers first
* emitted an item or sent a termination notification
* @see ReactiveX operators documentation: Amb
*/
@BackpressureSupport(BackpressureKind.PASS_THROUGH)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable ambArray(Publisher... sources) {
ObjectHelper.requireNonNull(sources, "sources is null");
int len = sources.length;
if (len == 0) {
return empty();
} else
if (len == 1) {
return fromPublisher(sources[0]);
}
return RxJavaPlugins.onAssembly(new FlowableAmb(sources, null));
}
/**
* Returns the default internal buffer size used by most async operators.
* The value can be overridden via system parameter {@code rx2.buffer-size}
* before the Flowable class is loaded.
* @return the default internal buffer size.
*/
public static int bufferSize() {
return BUFFER_SIZE;
}
/**
* Combines a collection of source Publishers by emitting an item that aggregates the latest values of each of
* the source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function.
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param
* the common base type of source values
* @param
* the result type
* @param sources
* the collection of source Publishers
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SchedulerSupport(SchedulerSupport.NONE)
@BackpressureSupport(BackpressureKind.FULL)
public static Flowable combineLatest(Publisher[] sources, Function combiner) {
return combineLatest(sources, combiner, bufferSize());
}
/**
* Combines a collection of source Publishers by emitting an item that aggregates the latest values of each of
* the source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function.
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param
* the common base type of source values
* @param
* the result type
* @param sources
* the collection of source Publishers
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SchedulerSupport(SchedulerSupport.NONE)
@BackpressureSupport(BackpressureKind.FULL)
public static Flowable combineLatest(Function combiner, Publisher... sources) {
return combineLatest(sources, combiner, bufferSize());
}
/**
* Combines a collection of source Publishers by emitting an item that aggregates the latest values of each of
* the source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function.
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param
* the common base type of source values
* @param
* the result type
* @param sources
* the collection of source Publishers
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @param bufferSize
* the internal buffer size and prefetch amount applied to every source Flowable
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SchedulerSupport(SchedulerSupport.NONE)
@BackpressureSupport(BackpressureKind.FULL)
public static Flowable combineLatest(Publisher[] sources, Function combiner, int bufferSize) {
ObjectHelper.requireNonNull(sources, "sources is null");
ObjectHelper.requireNonNull(combiner, "combiner is null");
verifyPositive(bufferSize, "bufferSize");
if (sources.length == 0) {
return empty();
}
return RxJavaPlugins.onAssembly(new FlowableCombineLatest(sources, combiner, bufferSize, false));
}
/**
* Combines a collection of source Publishers by emitting an item that aggregates the latest values of each of
* the source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function.
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param
* the common base type of source values
* @param
* the result type
* @param sources
* the collection of source Publishers
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SchedulerSupport(SchedulerSupport.NONE)
@BackpressureSupport(BackpressureKind.FULL)
public static Flowable combineLatest(Iterable> sources,
Function combiner) {
return combineLatest(sources, combiner, bufferSize());
}
/**
* Combines a collection of source Publishers by emitting an item that aggregates the latest values of each of
* the source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function.
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param
* the common base type of source values
* @param
* the result type
* @param sources
* the collection of source Publishers
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @param bufferSize
* the internal buffer size and prefetch amount applied to every source Flowable
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SchedulerSupport(SchedulerSupport.NONE)
@BackpressureSupport(BackpressureKind.FULL)
public static Flowable combineLatest(Iterable> sources,
Function combiner, int bufferSize) {
ObjectHelper.requireNonNull(sources, "sources is null");
ObjectHelper.requireNonNull(combiner, "combiner is null");
verifyPositive(bufferSize, "bufferSize");
return RxJavaPlugins.onAssembly(new FlowableCombineLatest(sources, combiner, bufferSize, false));
}
/**
* Combines a collection of source Publishers by emitting an item that aggregates the latest values of each of
* the source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function.
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param
* the common base type of source values
* @param
* the result type
* @param sources
* the collection of source Publishers
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SchedulerSupport(SchedulerSupport.NONE)
@BackpressureSupport(BackpressureKind.FULL)
public static Flowable combineLatestDelayError(Publisher[] sources,
Function combiner) {
return combineLatestDelayError(sources, combiner, bufferSize());
}
/**
* Combines a collection of source Publishers by emitting an item that aggregates the latest values of each of
* the source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function and delays any error from the sources until
* all source Publishers terminate.
*
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param
* the common base type of source values
* @param
* the result type
* @param sources
* the collection of source Publishers
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SchedulerSupport(SchedulerSupport.NONE)
@BackpressureSupport(BackpressureKind.FULL)
public static Flowable combineLatestDelayError(Function combiner,
Publisher... sources) {
return combineLatestDelayError(sources, combiner, bufferSize());
}
/**
* Combines a collection of source Publishers by emitting an item that aggregates the latest values of each of
* the source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function and delays any error from the sources until
* all source Publishers terminate.
*
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param
* the common base type of source values
* @param
* the result type
* @param sources
* the collection of source Publishers
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @param bufferSize
* the internal buffer size and prefetch amount applied to every source Flowable
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SchedulerSupport(SchedulerSupport.NONE)
@BackpressureSupport(BackpressureKind.FULL)
public static Flowable combineLatestDelayError(Publisher[] sources,
Function combiner, int bufferSize) {
ObjectHelper.requireNonNull(sources, "sources is null");
ObjectHelper.requireNonNull(combiner, "combiner is null");
verifyPositive(bufferSize, "bufferSize");
if (sources.length == 0) {
return empty();
}
return RxJavaPlugins.onAssembly(new FlowableCombineLatest(sources, combiner, bufferSize, true));
}
/**
* Combines a collection of source Publishers by emitting an item that aggregates the latest values of each of
* the source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function and delays any error from the sources until
* all source Publishers terminate.
*
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param
* the common base type of source values
* @param
* the result type
* @param sources
* the collection of source Publishers
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SchedulerSupport(SchedulerSupport.NONE)
@BackpressureSupport(BackpressureKind.FULL)
public static Flowable combineLatestDelayError(Iterable> sources,
Function combiner) {
return combineLatestDelayError(sources, combiner, bufferSize());
}
/**
* Combines a collection of source Publishers by emitting an item that aggregates the latest values of each of
* the source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function and delays any error from the sources until
* all source Publishers terminate.
*
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param
* the common base type of source values
* @param
* the result type
* @param sources
* the collection of source Publishers
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @param bufferSize
* the internal buffer size and prefetch amount applied to every source Flowable
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SchedulerSupport(SchedulerSupport.NONE)
@BackpressureSupport(BackpressureKind.FULL)
public static Flowable combineLatestDelayError(Iterable> sources,
Function combiner, int bufferSize) {
ObjectHelper.requireNonNull(sources, "sources is null");
ObjectHelper.requireNonNull(combiner, "combiner is null");
verifyPositive(bufferSize, "bufferSize");
return RxJavaPlugins.onAssembly(new FlowableCombineLatest(sources, combiner, bufferSize, true));
}
/**
* Combines two source Publishers by emitting an item that aggregates the latest values of each of the
* source Publishers each time an item is received from either of the source Publishers, where this
* aggregation is defined by a specified function.
*
* 
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the element type of the first source
* @param the element type of the second source
* @param the combined output type
* @param source1
* the first source Publisher
* @param source2
* the second source Publisher
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SuppressWarnings("unchecked")
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable combineLatest(
Publisher source1, Publisher source2,
BiFunction combiner) {
Function f = Functions.toFunction(combiner);
return combineLatest(f, source1, source2);
}
/**
* Combines three source Publishers by emitting an item that aggregates the latest values of each of the
* source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function.
*
*
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the element type of the first source
* @param the element type of the second source
* @param the element type of the third source
* @param the combined output type
* @param source1
* the first source Publisher
* @param source2
* the second source Publisher
* @param source3
* the third source Publisher
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SuppressWarnings("unchecked")
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable combineLatest(
Publisher source1, Publisher source2,
Publisher source3,
Function3 combiner) {
return combineLatest(Functions.toFunction(combiner), source1, source2, source3);
}
/**
* Combines four source Publishers by emitting an item that aggregates the latest values of each of the
* source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function.
*
*
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the element type of the first source
* @param the element type of the second source
* @param the element type of the third source
* @param the element type of the fourth source
* @param the combined output type
* @param source1
* the first source Publisher
* @param source2
* the second source Publisher
* @param source3
* the third source Publisher
* @param source4
* the fourth source Publisher
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SuppressWarnings("unchecked")
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable combineLatest(
Publisher source1, Publisher source2,
Publisher source3, Publisher source4,
Function4 combiner) {
return combineLatest(Functions.toFunction(combiner), source1, source2, source3, source4);
}
/**
* Combines five source Publishers by emitting an item that aggregates the latest values of each of the
* source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function.
*
*
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the element type of the first source
* @param the element type of the second source
* @param the element type of the third source
* @param the element type of the fourth source
* @param the element type of the fifth source
* @param the combined output type
* @param source1
* the first source Publisher
* @param source2
* the second source Publisher
* @param source3
* the third source Publisher
* @param source4
* the fourth source Publisher
* @param source5
* the fifth source Publisher
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SuppressWarnings("unchecked")
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable combineLatest(
Publisher source1, Publisher source2,
Publisher source3, Publisher source4,
Publisher source5,
Function5 combiner) {
return combineLatest(Functions.toFunction(combiner), source1, source2, source3, source4, source5);
}
/**
* Combines six source Publishers by emitting an item that aggregates the latest values of each of the
* source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function.
*
*
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the element type of the first source
* @param the element type of the second source
* @param the element type of the third source
* @param the element type of the fourth source
* @param the element type of the fifth source
* @param the element type of the sixth source
* @param the combined output type
* @param source1
* the first source Publisher
* @param source2
* the second source Publisher
* @param source3
* the third source Publisher
* @param source4
* the fourth source Publisher
* @param source5
* the fifth source Publisher
* @param source6
* the sixth source Publisher
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SuppressWarnings("unchecked")
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable combineLatest(
Publisher source1, Publisher source2,
Publisher source3, Publisher source4,
Publisher source5, Publisher source6,
Function6 combiner) {
return combineLatest(Functions.toFunction(combiner), source1, source2, source3, source4, source5, source6);
}
/**
* Combines seven source Publishers by emitting an item that aggregates the latest values of each of the
* source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function.
*
*
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the element type of the first source
* @param the element type of the second source
* @param the element type of the third source
* @param the element type of the fourth source
* @param the element type of the fifth source
* @param the element type of the sixth source
* @param the element type of the seventh source
* @param the combined output type
* @param source1
* the first source Publisher
* @param source2
* the second source Publisher
* @param source3
* the third source Publisher
* @param source4
* the fourth source Publisher
* @param source5
* the fifth source Publisher
* @param source6
* the sixth source Publisher
* @param source7
* the seventh source Publisher
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SuppressWarnings("unchecked")
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable combineLatest(
Publisher source1, Publisher source2,
Publisher source3, Publisher source4,
Publisher source5, Publisher source6,
Publisher source7,
Function7 combiner) {
return combineLatest(Functions.toFunction(combiner), source1, source2, source3, source4, source5, source6, source7);
}
/**
* Combines eight source Publishers by emitting an item that aggregates the latest values of each of the
* source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function.
*
*
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the element type of the first source
* @param the element type of the second source
* @param the element type of the third source
* @param the element type of the fourth source
* @param the element type of the fifth source
* @param the element type of the sixth source
* @param the element type of the seventh source
* @param the element type of the eighth source
* @param the combined output type
* @param source1
* the first source Publisher
* @param source2
* the second source Publisher
* @param source3
* the third source Publisher
* @param source4
* the fourth source Publisher
* @param source5
* the fifth source Publisher
* @param source6
* the sixth source Publisher
* @param source7
* the seventh source Publisher
* @param source8
* the eighth source Publisher
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SuppressWarnings("unchecked")
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable combineLatest(
Publisher source1, Publisher source2,
Publisher source3, Publisher source4,
Publisher source5, Publisher source6,
Publisher source7, Publisher source8,
Function8 combiner) {
return combineLatest(Functions.toFunction(combiner), source1, source2, source3, source4, source5, source6, source7, source8);
}
/**
* Combines nine source Publishers by emitting an item that aggregates the latest values of each of the
* source Publishers each time an item is received from any of the source Publishers, where this
* aggregation is defined by a specified function.
*
*
*
* - Backpressure:
* - The returned {@code Publisher} honors backpressure from downstream. The source {@code Publisher}s
* are requested in a bounded manner, however, their backpressure is not enforced (the operator won't signal
* {@code MissingBackpressureException}) and may lead to {@code OutOfMemoryError} due to internal buffer bloat.
* - Scheduler:
* - {@code combineLatest} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the element type of the first source
* @param the element type of the second source
* @param the element type of the third source
* @param the element type of the fourth source
* @param the element type of the fifth source
* @param the element type of the sixth source
* @param the element type of the seventh source
* @param the element type of the eighth source
* @param the element type of the ninth source
* @param the combined output type
* @param source1
* the first source Publisher
* @param source2
* the second source Publisher
* @param source3
* the third source Publisher
* @param source4
* the fourth source Publisher
* @param source5
* the fifth source Publisher
* @param source6
* the sixth source Publisher
* @param source7
* the seventh source Publisher
* @param source8
* the eighth source Publisher
* @param source9
* the ninth source Publisher
* @param combiner
* the aggregation function used to combine the items emitted by the source Publishers
* @return a Flowable that emits items that are the result of combining the items emitted by the source
* Publishers by means of the given aggregation function
* @see ReactiveX operators documentation: CombineLatest
*/
@SuppressWarnings("unchecked")
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable combineLatest(
Publisher source1, Publisher source2,
Publisher source3, Publisher source4,
Publisher source5, Publisher source6,
Publisher source7, Publisher source8,
Publisher source9,
Function9 combiner) {
return combineLatest(Functions.toFunction(combiner), source1, source2, source3, source4, source5, source6, source7, source8, source9);
}
/**
* Concatenates elements of each Publisher provided via an Iterable sequence into a single sequence
* of elements without interleaving them.
*
* 
*
* - Backpressure:
* - The operator honors backpressure from downstream. The {@code Publisher}
* sources are expected to honor backpressure as well.
* If any of the source {@code Publisher}s violate this, it may throw an
* {@code IllegalStateException} when the source {@code Publisher} completes.
* - Scheduler:
* - {@code concat} does not operate by default on a particular {@link Scheduler}.
*
* @param the common value type of the sources
* @param sources the Iterable sequence of Publishers
* @return the new Flowable instance
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable concat(Iterable> sources) {
ObjectHelper.requireNonNull(sources, "sources is null");
// unlike general sources, fromIterable can only throw on a boundary because it is consumed only there
return fromIterable(sources).concatMapDelayError((Function)Functions.identity(), 2, false);
}
/**
* Returns a Flowable that emits the items emitted by each of the Publishers emitted by the source
* Publisher, one after the other, without interleaving them.
*
*
*
* - Backpressure:
* - The operator honors backpressure from downstream. Both the outer and inner {@code Publisher}
* sources are expected to honor backpressure as well. If the outer violates this, a
* {@code MissingBackpressureException} is signalled. If any of the inner {@code Publisher}s violates
* this, it may throw an {@code IllegalStateException} when an inner {@code Publisher} completes.
* - Scheduler:
* - {@code concat} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the common element base type
* @param sources
* a Publisher that emits Publishers
* @return a Flowable that emits items all of the items emitted by the Publishers emitted by
* {@code Publishers}, one after the other, without interleaving them
* @see ReactiveX operators documentation: Concat
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable concat(Publisher> sources) {
return concat(sources, bufferSize());
}
/**
* Returns a Flowable that emits the items emitted by each of the Publishers emitted by the source
* Publisher, one after the other, without interleaving them.
*
*
*
* - Backpressure:
* - The operator honors backpressure from downstream. Both the outer and inner {@code Publisher}
* sources are expected to honor backpressure as well. If the outer violates this, a
* {@code MissingBackpressureException} is signalled. If any of the inner {@code Publisher}s violates
* this, it may throw an {@code IllegalStateException} when an inner {@code Publisher} completes.
* - Scheduler:
* - {@code concat} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the common element base type
* @param sources
* a Publisher that emits Publishers
* @param prefetch
* the number of Publishers to prefetch from the sources sequence.
* @return a Flowable that emits items all of the items emitted by the Publishers emitted by
* {@code Publishers}, one after the other, without interleaving them
* @see ReactiveX operators documentation: Concat
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable concat(Publisher> sources, int prefetch) {
return fromPublisher(sources).concatMap((Function)Functions.identity(), prefetch);
}
/**
* Returns a Flowable that emits the items emitted by two Publishers, one after the other, without
* interleaving them.
*
*
*
* - Backpressure:
* - The operator honors backpressure from downstream. The {@code Publisher}
* sources are expected to honor backpressure as well.
* If any of the source {@code Publisher}s violate this, it may throw an
* {@code IllegalStateException} when the source {@code Publisher} completes.
* - Scheduler:
* - {@code concat} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the common element base type
* @param source1
* a Publisher to be concatenated
* @param source2
* a Publisher to be concatenated
* @return a Flowable that emits items emitted by the two source Publishers, one after the other,
* without interleaving them
* @see ReactiveX operators documentation: Concat
*/
@SuppressWarnings("unchecked")
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable concat(Publisher source1, Publisher source2) {
return concatArray(source1, source2);
}
/**
* Returns a Flowable that emits the items emitted by three Publishers, one after the other, without
* interleaving them.
*
*
*
* - Backpressure:
* - The operator honors backpressure from downstream. The {@code Publisher}
* sources are expected to honor backpressure as well.
* If any of the source {@code Publisher}s violate this, it may throw an
* {@code IllegalStateException} when the source {@code Publisher} completes.
* - Scheduler:
* - {@code concat} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the common element base type
* @param source1
* a Publisher to be concatenated
* @param source2
* a Publisher to be concatenated
* @param source3
* a Publisher to be concatenated
* @return a Flowable that emits items emitted by the three source Publishers, one after the other,
* without interleaving them
* @see ReactiveX operators documentation: Concat
*/
@SuppressWarnings("unchecked")
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable concat(
Publisher source1, Publisher source2,
Publisher source3) {
return concatArray(source1, source2, source3);
}
/**
* Returns a Flowable that emits the items emitted by four Publishers, one after the other, without
* interleaving them.
*
*
*
* - Backpressure:
* - The operator honors backpressure from downstream. The {@code Publisher}
* sources are expected to honor backpressure as well.
* If any of the source {@code Publisher}s violate this, it may throw an
* {@code IllegalStateException} when the source {@code Publisher} completes.
* - Scheduler:
* - {@code concat} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the common element base type
* @param source1
* a Publisher to be concatenated
* @param source2
* a Publisher to be concatenated
* @param source3
* a Publisher to be concatenated
* @param source4
* a Publisher to be concatenated
* @return a Flowable that emits items emitted by the four source Publishers, one after the other,
* without interleaving them
* @see ReactiveX operators documentation: Concat
*/
@SuppressWarnings("unchecked")
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable concat(
Publisher source1, Publisher source2,
Publisher source3, Publisher source4) {
return concatArray(source1, source2, source3, source4);
}
/**
* Concatenates a variable number of Publisher sources.
*
* Note: named this way because of overload conflict with concat(Publisher<Publisher>).
*
*
*
* - Backpressure:
* - The operator honors backpressure from downstream. The {@code Publisher}
* sources are expected to honor backpressure as well.
* If any of the source {@code Publisher}s violate this, it may throw an
* {@code IllegalStateException} when the source {@code Publisher} completes.
* - Scheduler:
* - {@code concat} does not operate by default on a particular {@link Scheduler}.
*
* @param sources the array of sources
* @param the common base value type
* @return the new Publisher instance
* @throws NullPointerException if sources is null
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable concatArray(Publisher... sources) {
if (sources.length == 0) {
return empty();
} else
if (sources.length == 1) {
return fromPublisher(sources[0]);
}
return RxJavaPlugins.onAssembly(new FlowableConcatArray(sources, false));
}
/**
* Concatenates a variable number of Publisher sources and delays errors from any of them
* till all terminate.
*
*
*
* - Backpressure:
* - The operator honors backpressure from downstream. The {@code Publisher}
* sources are expected to honor backpressure as well.
* If any of the source {@code Publisher}s violate this, it may throw an
* {@code IllegalStateException} when the source {@code Publisher} completes.
* - Scheduler:
* - {@code concat} does not operate by default on a particular {@link Scheduler}.
*
* @param sources the array of sources
* @param the common base value type
* @return the new Flowable instance
* @throws NullPointerException if sources is null
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable concatArrayDelayError(Publisher... sources) {
if (sources.length == 0) {
return empty();
} else
if (sources.length == 1) {
return fromPublisher(sources[0]);
}
return RxJavaPlugins.onAssembly(new FlowableConcatArray(sources, true));
}
/**
* Concatenates a sequence of Publishers eagerly into a single stream of values.
*
* Eager concatenation means that once a subscriber subscribes, this operator subscribes to all of the
* source Publishers. The operator buffers the values emitted by these Publishers and then drains them
* in order, each one after the previous one completes.
*
* - Backpressure:
* - The operator honors backpressure from downstream. The {@code Publisher}
* sources are expected to honor backpressure as well.
* If any of the source {@code Publisher}s violate this, the operator will signal a
* {@code MissingBackpressureException}.
* - Scheduler:
* - This method does not operate by default on a particular {@link Scheduler}.
*
* @param the value type
* @param sources a sequence of Publishers that need to be eagerly concatenated
* @return the new Publisher instance with the specified concatenation behavior
* @since 2.0
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable concatArrayEager(Publisher... sources) {
return concatArrayEager(bufferSize(), bufferSize(), sources);
}
/**
* Concatenates a sequence of Publishers eagerly into a single stream of values.
*
* Eager concatenation means that once a subscriber subscribes, this operator subscribes to all of the
* source Publishers. The operator buffers the values emitted by these Publishers and then drains them
* in order, each one after the previous one completes.
*
* - Backpressure:
* - The operator honors backpressure from downstream. The {@code Publisher}
* sources are expected to honor backpressure as well.
* If any of the source {@code Publisher}s violate this, the operator will signal a
* {@code MissingBackpressureException}.
* - Scheduler:
* - This method does not operate by default on a particular {@link Scheduler}.
*
* @param the value type
* @param sources a sequence of Publishers that need to be eagerly concatenated
* @param maxConcurrency the maximum number of concurrent subscriptions at a time, Integer.MAX_VALUE
* is interpreted as indication to subscribe to all sources at once
* @param prefetch the number of elements to prefetch from each Publisher source
* @return the new Publisher instance with the specified concatenation behavior
* @since 2.0
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
@SuppressWarnings({ "rawtypes", "unchecked" })
public static Flowable concatArrayEager(int maxConcurrency, int prefetch, Publisher... sources) {
return RxJavaPlugins.onAssembly(new FlowableConcatMapEager(new FlowableFromArray(sources), Functions.identity(), maxConcurrency, prefetch, ErrorMode.IMMEDIATE));
}
/**
* Concatenates the Iterable sequence of Publishers into a single sequence by subscribing to each Publisher,
* one after the other, one at a time and delays any errors till the all inner Publishers terminate.
*
*
* - Backpressure:
* - The operator honors backpressure from downstream. Both the outer and inner {@code Publisher}
* sources are expected to honor backpressure as well. If the outer violates this, a
* {@code MissingBackpressureException} is signalled. If any of the inner {@code Publisher}s violates
* this, it may throw an {@code IllegalStateException} when an inner {@code Publisher} completes.
* - Scheduler:
* - {@code concatDelayError} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the common element base type
* @param sources the Iterable sequence of Publishers
* @return the new Publisher with the concatenating behavior
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable concatDelayError(Iterable> sources) {
ObjectHelper.requireNonNull(sources, "sources is null");
return fromIterable(sources).concatMapDelayError((Function)Functions.identity());
}
/**
* Concatenates the Publisher sequence of Publishers into a single sequence by subscribing to each inner Publisher,
* one after the other, one at a time and delays any errors till the all inner and the outer Publishers terminate.
*
*
* - Backpressure:
* - {@code concatDelayError} fully supports backpressure.
* - Scheduler:
* - {@code concatDelayError} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the common element base type
* @param sources the Publisher sequence of Publishers
* @return the new Publisher with the concatenating behavior
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable concatDelayError(Publisher> sources) {
return concatDelayError(sources, bufferSize(), true);
}
/**
* Concatenates the Publisher sequence of Publishers into a single sequence by subscribing to each inner Publisher,
* one after the other, one at a time and delays any errors till the all inner and the outer Publishers terminate.
*
*
* - Backpressure:
* - {@code concatDelayError} fully supports backpressure.
* - Scheduler:
* - {@code concatDelayError} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the common element base type
* @param sources the Publisher sequence of Publishers
* @param prefetch the number of elements to prefetch from the outer Publisher
* @param tillTheEnd if true exceptions from the outer and all inner Publishers are delayed to the end
* if false, exception from the outer Publisher is delayed till the current Publisher terminates
* @return the new Publisher with the concatenating behavior
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable concatDelayError(Publisher> sources, int prefetch, boolean tillTheEnd) {
return fromPublisher(sources).concatMapDelayError((Function)Functions.identity(), prefetch, tillTheEnd);
}
/**
* Concatenates a Publisher sequence of Publishers eagerly into a single stream of values.
*
* Eager concatenation means that once a subscriber subscribes, this operator subscribes to all of the
* emitted source Publishers as they are observed. The operator buffers the values emitted by these
* Publishers and then drains them in order, each one after the previous one completes.
*
* - Backpressure:
* - Backpressure is honored towards the downstream and both the outer and inner Publishers are
* expected to support backpressure. Violating this assumption, the operator will
* signal {@code MissingBackpressureException}.
* - Scheduler:
* - This method does not operate by default on a particular {@link Scheduler}.
*
* @param the value type
* @param sources a sequence of Publishers that need to be eagerly concatenated
* @return the new Publisher instance with the specified concatenation behavior
* @since 2.0
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable concatEager(Publisher> sources) {
return concatEager(sources, bufferSize(), bufferSize());
}
/**
* Concatenates a Publisher sequence of Publishers eagerly into a single stream of values.
*
* Eager concatenation means that once a subscriber subscribes, this operator subscribes to all of the
* emitted source Publishers as they are observed. The operator buffers the values emitted by these
* Publishers and then drains them in order, each one after the previous one completes.
*
* - Backpressure:
* - Backpressure is honored towards the downstream and both the outer and inner Publishers are
* expected to support backpressure. Violating this assumption, the operator will
* signal {@code MissingBackpressureException}.
* - Scheduler:
* - This method does not operate by default on a particular {@link Scheduler}.
*
* @param the value type
* @param sources a sequence of Publishers that need to be eagerly concatenated
* @param maxConcurrency the maximum number of concurrently running inner Publishers; Integer.MAX_VALUE
* is interpreted as all inner Publishers can be active at the same time
* @param prefetch the number of elements to prefetch from each inner Publisher source
* @return the new Publisher instance with the specified concatenation behavior
* @since 2.0
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
@SuppressWarnings({ "rawtypes", "unchecked" })
public static Flowable concatEager(Publisher> sources, int maxConcurrency, int prefetch) {
return RxJavaPlugins.onAssembly(new FlowableConcatMapEager(sources, Functions.identity(), maxConcurrency, prefetch, ErrorMode.IMMEDIATE));
}
/**
* Concatenates a sequence of Publishers eagerly into a single stream of values.
*
* Eager concatenation means that once a subscriber subscribes, this operator subscribes to all of the
* source Publishers. The operator buffers the values emitted by these Publishers and then drains them
* in order, each one after the previous one completes.
*
* - Backpressure:
* - Backpressure is honored towards the downstream and both the outer and inner Publishers are
* expected to support backpressure. Violating this assumption, the operator will
* signal {@code MissingBackpressureException}.
* - Scheduler:
* - This method does not operate by default on a particular {@link Scheduler}.
*
* @param the value type
* @param sources a sequence of Publishers that need to be eagerly concatenated
* @return the new Publisher instance with the specified concatenation behavior
* @since 2.0
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable concatEager(Iterable> sources) {
return concatEager(sources, bufferSize(), bufferSize());
}
/**
* Concatenates a sequence of Publishers eagerly into a single stream of values.
*
* Eager concatenation means that once a subscriber subscribes, this operator subscribes to all of the
* source Publishers. The operator buffers the values emitted by these Publishers and then drains them
* in order, each one after the previous one completes.
*
* - Backpressure:
* - Backpressure is honored towards the downstream and both the outer and inner Publishers are
* expected to support backpressure. Violating this assumption, the operator will
* signal {@code MissingBackpressureException}.
* - Scheduler:
* - This method does not operate by default on a particular {@link Scheduler}.
*
* @param the value type
* @param sources a sequence of Publishers that need to be eagerly concatenated
* @param maxConcurrency the maximum number of concurrently running inner Publishers; Integer.MAX_VALUE
* is interpreted as all inner Publishers can be active at the same time
* @param prefetch the number of elements to prefetch from each inner Publisher source
* @return the new Publisher instance with the specified concatenation behavior
* @since 2.0
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
@SuppressWarnings({ "rawtypes", "unchecked" })
public static Flowable concatEager(Iterable> sources, int maxConcurrency, int prefetch) {
return RxJavaPlugins.onAssembly(new FlowableConcatMapEager(new FlowableFromIterable(sources), Functions.identity(), maxConcurrency, prefetch, ErrorMode.IMMEDIATE));
}
/**
* Provides an API (via a cold Flowable) that bridges the reactive world with the callback-style,
* generally non-backpressured world.
*
* Example:
*
* Flowable.<Event>create(emitter -> {
* Callback listener = new Callback() {
* @Override
* public void onEvent(Event e) {
* emitter.onNext(e);
* if (e.isLast()) {
* emitter.onComplete();
* }
* }
*
* @Override
* public void onFailure(Exception e) {
* emitter.onError(e);
* }
* };
*
* AutoCloseable c = api.someMethod(listener);
*
* emitter.setCancellable(c::close);
*
* }, BackpressureMode.BUFFER);
*
*
* You should call the FlowableEmitter onNext, onError and onComplete methods in a serialized fashion. The
* rest of its methods are thread-safe.
*
* @param the element type
* @param source the emitter that is called when a Subscriber subscribes to the returned {@code Flowable}
* @param mode the backpressure mode to apply if the downstream Subscriber doesn't request (fast) enough
* @return the new Flowable instance
* @see FlowableOnSubscribe
* @see FlowableEmitter.BackpressureMode
* @see Cancellable
*/
@BackpressureSupport(BackpressureKind.SPECIAL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable create(FlowableOnSubscribe source, FlowableEmitter.BackpressureMode mode) {
return RxJavaPlugins.onAssembly(new FlowableCreate(source, mode));
}
/**
* Returns a Flowable that calls a Publisher factory to create a Publisher for each new Subscriber
* that subscribes. That is, for each subscriber, the actual Publisher that subscriber observes is
* determined by the factory function.
*
* 
*
* The defer Subscriber allows you to defer or delay emitting items from a Publisher until such time as an
* Subscriber subscribes to the Publisher. This allows an {@link Subscriber} to easily obtain updates or a
* refreshed version of the sequence.
*
* - Backpressure:
* - The operator itself doesn't interfere with backpressure which is determined by the {@code Publisher}
* returned by the {@code PublisherFactory}.
* - Scheduler:
* - {@code defer} does not operate by default on a particular {@link Scheduler}.
*
*
* @param supplier
* the Publisher factory function to invoke for each {@link Subscriber} that subscribes to the
* resulting Publisher
* @param
* the type of the items emitted by the Publisher
* @return a Flowable whose {@link Subscriber}s' subscriptions trigger an invocation of the given
* Publisher factory function
* @see ReactiveX operators documentation: Defer
*/
@BackpressureSupport(BackpressureKind.PASS_THROUGH)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable defer(Callable> supplier) {
ObjectHelper.requireNonNull(supplier, "supplier is null");
return RxJavaPlugins.onAssembly(new FlowableDefer(supplier));
}
/**
* Returns a Flowable that emits no items to the {@link Subscriber} and immediately invokes its
* {@link Subscriber#onComplete onComplete} method.
*
* 
*
* - Backpressure:
* - This source doesn't produce any elements and effectively ignores downstream backpressure.
* - Scheduler:
* - {@code empty} does not operate by default on a particular {@link Scheduler}.
*
*
* @param
* the type of the items (ostensibly) emitted by the Publisher
* @return a Flowable that emits no items to the {@link Subscriber} but immediately invokes the
* {@link Subscriber}'s {@link Subscriber#onComplete() onComplete} method
* @see ReactiveX operators documentation: Empty
*/
@BackpressureSupport(BackpressureKind.PASS_THROUGH)
@SchedulerSupport(SchedulerSupport.NONE)
@SuppressWarnings("unchecked")
public static Flowable empty() {
return RxJavaPlugins.onAssembly((Flowable) FlowableEmpty.INSTANCE);
}
/**
* Returns a Flowable that invokes an {@link Subscriber}'s {@link Subscriber#onError onError} method when the
* Subscriber subscribes to it.
*
* 
*
* - Backpressure:
* - This source doesn't produce any elements and effectively ignores downstream backpressure.
* - Scheduler:
* - {@code error} does not operate by default on a particular {@link Scheduler}.
*
*
* @param supplier
* a Callable factory to return a Throwable for each individual Subscriber
* @param
* the type of the items (ostensibly) emitted by the Publisher
* @return a Flowable that invokes the {@link Subscriber}'s {@link Subscriber#onError onError} method when
* the Subscriber subscribes to it
* @see ReactiveX operators documentation: Throw
*/
@BackpressureSupport(BackpressureKind.PASS_THROUGH)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable error(Callable supplier) {
ObjectHelper.requireNonNull(supplier, "errorSupplier is null");
return RxJavaPlugins.onAssembly(new FlowableError(supplier));
}
/**
* Returns a Flowable that invokes an {@link Subscriber}'s {@link Subscriber#onError onError} method when the
* Subscriber subscribes to it.
*
*
*
* - Backpressure:
* - This source doesn't produce any elements and effectively ignores downstream backpressure.
* - Scheduler:
* - {@code error} does not operate by default on a particular {@link Scheduler}.
*
*
* @param throwable
* the particular Throwable to pass to {@link Subscriber#onError onError}
* @param
* the type of the items (ostensibly) emitted by the Publisher
* @return a Flowable that invokes the {@link Subscriber}'s {@link Subscriber#onError onError} method when
* the Subscriber subscribes to it
* @see ReactiveX operators documentation: Throw
*/
@BackpressureSupport(BackpressureKind.PASS_THROUGH)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable error(final Throwable throwable) {
ObjectHelper.requireNonNull(throwable, "throwable is null");
return error(Functions.justCallable(throwable));
}
/**
* Converts an Array into a Publisher that emits the items in the Array.
*
* 
*
* - Backpressure:
* - The operator honors backpressure from downstream and iterates the given {@code array}
* on demand (i.e., when requested).
* - Scheduler:
* - {@code from} does not operate by default on a particular {@link Scheduler}.
*
*
* @param items
* the array of elements
* @param
* the type of items in the Array and the type of items to be emitted by the resulting Publisher
* @return a Flowable that emits each item in the source Array
* @see ReactiveX operators documentation: From
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable fromArray(T... items) {
ObjectHelper.requireNonNull(items, "items is null");
if (items.length == 0) {
return empty();
}
if (items.length == 1) {
return just(items[0]);
}
return RxJavaPlugins.onAssembly(new FlowableFromArray(items));
}
/**
* Returns a Flowable that, when an Subscriber subscribes to it, invokes a function you specify and then
* emits the value returned from that function.
*
* 
*
* This allows you to defer the execution of the function you specify until an Subscriber subscribes to the
* Publisher. That is to say, it makes the function "lazy."
*
* - Backpressure:
* - The operator honors backpressure from downstream.
* - Scheduler:
* - {@code fromCallable} does not operate by default on a particular {@link Scheduler}.
*
*
* @param supplier
* a function, the execution of which should be deferred; {@code fromCallable} will invoke this
* function only when an Subscriber subscribes to the Publisher that {@code fromCallable} returns
* @param
* the type of the item emitted by the Publisher
* @return a Flowable whose {@link Subscriber}s' subscriptions trigger an invocation of the given function
* @see #defer(Callable)
* @since 2.0
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable fromCallable(Callable supplier) {
ObjectHelper.requireNonNull(supplier, "supplier is null");
return RxJavaPlugins.onAssembly(new FlowableFromCallable(supplier));
}
/**
* Converts a {@link Future} into a Publisher.
*
* 
*
* You can convert any object that supports the {@link Future} interface into a Publisher that emits the
* return value of the {@link Future#get} method of that object, by passing the object into the {@code from}
* method.
*
* Important note: This Publisher is blocking; you cannot unsubscribe from it.
*
* Unlike 1.x, cancelling the Flowable won't cancel the future. If necessary, one can use composition to achieve the
* cancellation effect: {@code futurePublisher.doOnCancel(() -> future.cancel(true));}.
*
* - Backpressure:
* - The operator honors backpressure from downstream.
* - Scheduler:
* - {@code from} does not operate by default on a particular {@link Scheduler}.
*
*
* @param future
* the source {@link Future}
* @param
* the type of object that the {@link Future} returns, and also the type of item to be emitted by
* the resulting Publisher
* @return a Flowable that emits the item from the source {@link Future}
* @see ReactiveX operators documentation: From
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable fromFuture(Future future) {
ObjectHelper.requireNonNull(future, "future is null");
return RxJavaPlugins.onAssembly(new FlowableFromFuture(future, 0L, null));
}
/**
* Converts a {@link Future} into a Publisher, with a timeout on the Future.
*
*
*
* You can convert any object that supports the {@link Future} interface into a Publisher that emits the
* return value of the {@link Future#get} method of that object, by passing the object into the {@code from}
* method.
*
* Unlike 1.x, cancelling the Flowable won't cancel the future. If necessary, one can use composition to achieve the
* cancellation effect: {@code futurePublisher.doOnCancel(() -> future.cancel(true));}.
*
* Important note: This Publisher is blocking; you cannot unsubscribe from it.
*
* - Backpressure:
* - The operator honors backpressure from downstream.
* - Scheduler:
* - {@code from} does not operate by default on a particular {@link Scheduler}.
*
*
* @param future
* the source {@link Future}
* @param timeout
* the maximum time to wait before calling {@code get}
* @param unit
* the {@link TimeUnit} of the {@code timeout} argument
* @param
* the type of object that the {@link Future} returns, and also the type of item to be emitted by
* the resulting Publisher
* @return a Flowable that emits the item from the source {@link Future}
* @see ReactiveX operators documentation: From
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable fromFuture(Future future, long timeout, TimeUnit unit) {
ObjectHelper.requireNonNull(future, "future is null");
ObjectHelper.requireNonNull(unit, "unit is null");
return RxJavaPlugins.onAssembly(new FlowableFromFuture(future, timeout, unit));
}
/**
* Converts a {@link Future} into a Publisher, with a timeout on the Future.
*
*
*
* You can convert any object that supports the {@link Future} interface into a Publisher that emits the
* return value of the {@link Future#get} method of that object, by passing the object into the {@code from}
* method.
*
* Unlike 1.x, cancelling the Flowable won't cancel the future. If necessary, one can use composition to achieve the
* cancellation effect: {@code futurePublisher.doOnCancel(() -> future.cancel(true));}.
*
* Important note: This Publisher is blocking; you cannot unsubscribe from it.
*
* - Backpressure:
* - The operator honors backpressure from downstream.
* - Scheduler:
* - {@code from} does not operate by default on a particular {@link Scheduler}.
*
*
* @param future
* the source {@link Future}
* @param timeout
* the maximum time to wait before calling {@code get}
* @param unit
* the {@link TimeUnit} of the {@code timeout} argument
* @param scheduler
* the {@link Scheduler} to wait for the Future on. Use a Scheduler such as
* {@link Schedulers#io()} that can block and wait on the Future
* @param
* the type of object that the {@link Future} returns, and also the type of item to be emitted by
* the resulting Publisher
* @return a Flowable that emits the item from the source {@link Future}
* @see ReactiveX operators documentation: From
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.CUSTOM)
public static Flowable fromFuture(Future future, long timeout, TimeUnit unit, Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return fromFuture(future, timeout, unit).subscribeOn(scheduler);
}
/**
* Converts a {@link Future}, operating on a specified {@link Scheduler}, into a Publisher.
*
* 
*
* You can convert any object that supports the {@link Future} interface into a Publisher that emits the
* return value of the {@link Future#get} method of that object, by passing the object into the {@code from}
* method.
*
* Unlike 1.x, cancelling the Flowable won't cancel the future. If necessary, one can use composition to achieve the
* cancellation effect: {@code futurePublisher.doOnCancel(() -> future.cancel(true));}.
*
* - Backpressure:
* - The operator honors backpressure from downstream.
* - Scheduler:
* - you specify which {@link Scheduler} this operator will use
*
*
* @param future
* the source {@link Future}
* @param scheduler
* the {@link Scheduler} to wait for the Future on. Use a Scheduler such as
* {@link Schedulers#io()} that can block and wait on the Future
* @param
* the type of object that the {@link Future} returns, and also the type of item to be emitted by
* the resulting Publisher
* @return a Flowable that emits the item from the source {@link Future}
* @see ReactiveX operators documentation: From
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.CUSTOM)
public static Flowable fromFuture(Future future, Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return fromFuture(future).subscribeOn(scheduler);
}
/**
* Converts an {@link Iterable} sequence into a Publisher that emits the items in the sequence.
*
*
*
* - Backpressure:
* - The operator honors backpressure from downstream and iterates the given {@code iterable}
* on demand (i.e., when requested).
* - Scheduler:
* - {@code from} does not operate by default on a particular {@link Scheduler}.
*
*
* @param source
* the source {@link Iterable} sequence
* @param
* the type of items in the {@link Iterable} sequence and the type of items to be emitted by the
* resulting Publisher
* @return a Flowable that emits each item in the source {@link Iterable} sequence
* @see ReactiveX operators documentation: From
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable fromIterable(Iterable source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new FlowableFromIterable(source));
}
/**
* Converts an arbitrary Reactive-Streams Publisher into a Flowable if not already a
* Flowable.
*
* - Backpressure:
* - The operator is a pass-through for backpressure and its behavior is determined by the
* backpressure behavior of the wrapped publisher.
* - Scheduler:
* - {@code fromPublisher} does not operate by default on a particular {@link Scheduler}.
*
* @param the value type of the flow
* @param source the Publisher to convert
* @return the new Flowable instance
* @throws NullPointerException if publisher is null
*/
@BackpressureSupport(BackpressureKind.PASS_THROUGH)
@SchedulerSupport(SchedulerSupport.NONE)
@SuppressWarnings("unchecked")
public static Flowable fromPublisher(final Publisher source) {
if (source instanceof Flowable) {
return RxJavaPlugins.onAssembly((Flowable)source);
}
ObjectHelper.requireNonNull(source, "publisher is null");
return RxJavaPlugins.onAssembly(new FlowableFromPublisher(source));
}
/**
* Returns a cold, synchronous, stateless and backpressure-aware generator of values.
*
*
* - Backpressure:
* - The operator honors downstream backpressure.
* - Scheduler:
* - {@code generator} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the generated value type
* @param generator the Consumer called whenever a particular downstream Subscriber has
* requested a value. The callback then should call {@code onNext}, {@code onError} or
* {@code onComplete} to signal a value or a terminal event. Signalling multiple {@code onNext}
* in a call will make the operator signal {@code IllegalStateException}.
* @return the new Flowable instance
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable generate(final Consumer> generator) {
ObjectHelper.requireNonNull(generator, "generator is null");
return generate(Functions.nullSupplier(),
FlowableInternalHelper.simpleGenerator(generator),
Functions.emptyConsumer());
}
/**
* Returns a cold, synchronous, stateful and backpressure-aware generator of values.
*
*
* - Backpressure:
* - The operator honors downstream backpressure.
* - Scheduler:
* - {@code generate} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the type of the per-Subscriber state
* @param the generated value type
* @param initialState the Callable to generate the initial state for each Subscriber
* @param generator the Consumer called with the current state whenever a particular downstream Subscriber has
* requested a value. The callback then should call {@code onNext}, {@code onError} or
* {@code onComplete} to signal a value or a terminal event. Signalling multiple {@code onNext}
* in a call will make the operator signal {@code IllegalStateException}.
* @return the new Flowable instance
*/
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public static Flowable generate(Callable initialState, final BiConsumer> generator) {
ObjectHelper.requireNonNull(generator, "generator is null");
return generate(initialState, FlowableInternalHelper.simpleBiGenerator(generator),
Functions.emptyConsumer());
}
/**
* Returns a cold, synchronous, stateful and backpressure-aware generator of values.
*