/**
* 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.concurrent.Callable;
import io.reactivex.annotations.SchedulerSupport;
import io.reactivex.disposables.Disposable;
import io.reactivex.exceptions.Exceptions;
import io.reactivex.functions.*;
import io.reactivex.internal.functions.*;
import io.reactivex.internal.operators.maybe.*;
import io.reactivex.internal.util.ExceptionHelper;
import io.reactivex.plugins.RxJavaPlugins;
import io.reactivex.subscribers.TestSubscriber;
/**
* Represents a deferred computation and emission of a maybe value or exception.
*
* The main consumer type of Maybe is {@link MaybeObserver} whose methods are called
* in a sequential fashion following this protocol:
* {@code onSubscribe (onSuccess | onError | onComplete)?}.
*
* @param the value type
* @since 2.0
*/
public abstract class Maybe implements MaybeSource {
/**
* Provides an API (via a cold Maybe) that bridges the reactive world with the callback-style world.
*
* Example:
*
* Maybe.<Event>create(emitter -> {
* Callback listener = new Callback() {
* @Override
* public void onEvent(Event e) {
* if (e.isNothing()) {
* emitter.onComplete();
* } else {
* emitter.onSuccess(e);
* }
* }
*
* @Override
* public void onFailure(Exception e) {
* emitter.onError(e);
* }
* };
*
* AutoCloseable c = api.someMethod(listener);
*
* emitter.setCancellable(c::close);
*
* });
*
*
*
* - Scheduler:
* - {@code create} does not operate by default on a particular {@link Scheduler}.
*
* @param the value type
* @param onSubscribe the emitter that is called when a MaybeObserver subscribes to the returned {@code Flowable}
* @return the new Maybe instance
* @see MaybeOnSubscribe
* @see Cancellable
*/
public static Maybe create(MaybeOnSubscribe onSubscribe) {
ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");
return RxJavaPlugins.onAssembly(new MaybeCreate(onSubscribe));
}
/**
* Calls a Callable for each individual MaybeObserver to return the actual MaybeSource source to
* be subscribe to.
*
* - Scheduler:
* - {@code defer} does not operate by default on a particular {@link Scheduler}.
*
* @param the value type
* @param maybeSupplier the Callable that is called for each individual MaybeObserver and
* returns a MaybeSource instance to subscribe to
* @return the new Maybe instance
*/
public static Maybe defer(final Callable> maybeSupplier) {
ObjectHelper.requireNonNull(maybeSupplier, "maybeSupplier is null");
return RxJavaPlugins.onAssembly(new MaybeDefer(maybeSupplier));
}
/**
* Returns a (singleton) Maybe instance that calls {@link MaybeObserver#onComplete onComplete}
* immediately.
*
* 
*
* - Scheduler:
* - {@code empty} does not operate by default on a particular {@link Scheduler}.
*
* @param the value type
* @return the new Maybe instance
*/
@SuppressWarnings("unchecked")
public static Maybe empty() {
return RxJavaPlugins.onAssembly((Maybe)MaybeEmpty.INSTANCE);
}
/**
* Returns a Maybe that invokes a subscriber's {@link MaybeObserver#onError onError} method when the
* subscriber subscribes to it.
*
* 
*
* - Scheduler:
* - {@code error} does not operate by default on a particular {@link Scheduler}.
*
*
* @param exception
* the particular Throwable to pass to {@link MaybeObserver#onError onError}
* @param
* the type of the item (ostensibly) emitted by the Maybe
* @return a Maybe that invokes the subscriber's {@link MaybeObserver#onError onError} method when
* the subscriber subscribes to it
* @see ReactiveX operators documentation: Throw
*/
public static Maybe error(Throwable exception) {
ObjectHelper.requireNonNull(exception, "exception is null");
return RxJavaPlugins.onAssembly(new MaybeError(exception));
}
/**
* Returns a Maybe that invokes an {@link Observer}'s {@link MaybeObserver#onError onError} method when the
* Observer subscribes to it.
*
* 
*
* - 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 Maybe that invokes the {@link MaybeObserver}'s {@link MaybeObserver#onError onError} method when
* the MaybeObserver subscribes to it
* @see ReactiveX operators documentation: Throw
*/
public static Maybe error(Callable supplier) {
ObjectHelper.requireNonNull(supplier, "errorSupplier is null");
return RxJavaPlugins.onAssembly(new MaybeErrorCallable(supplier));
}
/**
* Returns a Maybe instance that runs the given Action for each subscriber and
* emits either its exception or simply completes.
*
* - Scheduler:
* - {@code fromAction} does not operate by default on a particular {@link Scheduler}.
*
* @param the target type
* @param run the runnable to run for each subscriber
* @return the new Maybe instance
* @throws NullPointerException if run is null
*/
@SchedulerSupport(SchedulerSupport.NONE)
public static Maybe fromAction(final Action run) {
ObjectHelper.requireNonNull(run, "run is null");
return RxJavaPlugins.onAssembly(new MaybeFromAction(run));
}
/**
* Returns a {@link Maybe} that invokes passed function and emits its result for each new MaybeObserver that subscribes
* while considering {@code null} value from the callable as indication for valueless completion.
*
* Allows you to defer execution of passed function until MaybeObserver subscribes to the {@link Maybe}.
* It makes passed function "lazy".
* Result of the function invocation will be emitted by the {@link Maybe}.
*
* - Scheduler:
* - {@code fromCallable} does not operate by default on a particular {@link Scheduler}.
*
*
* @param callable
* function which execution should be deferred, it will be invoked when MaybeObserver will subscribe to the {@link Maybe}.
* @param
* the type of the item emitted by the {@link Maybe}.
* @return a {@link Maybe} whose {@link MaybeObserver}s' subscriptions trigger an invocation of the given function.
*/
public static Maybe fromCallable(final Callable callable) {
ObjectHelper.requireNonNull(callable, "callable is null");
return RxJavaPlugins.onAssembly(new MaybeFromCallable(callable));
}
/**
* Returns a Maybe instance that runs the given Action for each subscriber and
* emits either its exception or simply completes.
*
* - Scheduler:
* - {@code fromRunnable} does not operate by default on a particular {@link Scheduler}.
*
* @param the target type
* @param run the runnable to run for each subscriber
* @return the new Maybe instance
* @throws NullPointerException if run is null
*/
@SchedulerSupport(SchedulerSupport.NONE)
public static Maybe fromRunnable(final Runnable run) {
ObjectHelper.requireNonNull(run, "run is null");
return RxJavaPlugins.onAssembly(new MaybeFromRunnable(run));
}
/**
* Returns a {@code Maybe} that emits a specified item.
*
* 
*
* To convert any object into a {@code Single} that emits that object, pass that object into the
* {@code just} method.
*
* - Scheduler:
* - {@code just} does not operate by default on a particular {@link Scheduler}.
*
*
* @param item
* the item to emit
* @param
* the type of that item
* @return a {@code Maybe} that emits {@code item}
* @see ReactiveX operators documentation: Just
*/
public static Maybe just(T item) {
ObjectHelper.requireNonNull(item, "item is null");
return RxJavaPlugins.onAssembly(new MaybeJust(item));
}
/**
* Returns a Maybe that never sends any items or notifications to an {@link MaybeObserver}.
*
* 
*
* This Publisher is useful primarily for testing purposes.
*
* - Scheduler:
* - {@code never} does not operate by default on a particular {@link Scheduler}.
*
*
* @param
* the type of items (not) emitted by the Publisher
* @return a Maybe that never emits any items or sends any notifications to an {@link MaybeObserver}
* @see ReactiveX operators documentation: Never
*/
@SuppressWarnings("unchecked")
public static Maybe never() {
return RxJavaPlugins.onAssembly((Maybe)MaybeNever.INSTANCE);
}
/**
* Advanced use only: creates a Maybe instance without
* any safeguards by using a callback that is called with a MaybeObserver.
*
* - Scheduler:
* - {@code unsafeCreate} does not operate by default on a particular {@link Scheduler}.
*
* @param the value type
* @param onSubscribe the function that is called with the subscribing MaybeObserver
* @return the new Maybe instance
*/
public static Maybe unsafeCreate(MaybeSource onSubscribe) {
if (onSubscribe instanceof Maybe) {
throw new IllegalArgumentException("unsafeCreate(Maybe) should be upgraded");
}
ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");
return RxJavaPlugins.onAssembly(new MaybeUnsafeCreate(onSubscribe));
}
/**
* Wraps a MaybeSource instance into a new Maybe instance if not already a Maybe
* instance.
*
* - Scheduler:
* - {@code wrap} does not operate by default on a particular {@link Scheduler}.
*
* @param the value type
* @param source the source to wrap
* @return the Maybe wrapper or the source cast to Maybe (if possible)
*/
public static Maybe wrap(MaybeSource source) {
if (source instanceof Maybe) {
return RxJavaPlugins.onAssembly((Maybe)source);
}
ObjectHelper.requireNonNull(source, "onSubscribe is null");
return RxJavaPlugins.onAssembly(new MaybeUnsafeCreate(source));
}
// ------------------------------------------------------------------
// Instance methods
// ------------------------------------------------------------------
/**
* Casts the success value of the current Maybe into the target type or signals a
* ClassCastException if not compatible.
*
* - Scheduler:
* - {@code cast} does not operate by default on a particular {@link Scheduler}.
*
* @param the target type
* @param clazz the type token to use for casting the success result from the current Maybe
* @return the new Maybe instance
* @since 2.0
*/
public final Maybe cast(final Class clazz) {
ObjectHelper.requireNonNull(clazz, "clazz is null");
return map(Functions.castFunction(clazz));
}
/**
* Transform a Maybe by applying a particular Transformer function to it.
*
* This method operates on the Maybe itself whereas {@link #lift} operates on the Maybe's MaybeObservers.
*
* If the operator you are creating is designed to act on the individual item emitted by a Maybe, use
* {@link #lift}. If your operator is designed to transform the source Maybe as a whole (for instance, by
* applying a particular set of existing RxJava operators to it) use {@code compose}.
*
* - Scheduler:
* - {@code compose} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the value type of the Maybe returned by the transformer function
* @param transformer
* implements the function that transforms the source Maybe
* @return a Maybe, transformed by the transformer function
* @see RxJava wiki: Implementing Your Own Operators
*/
public final Maybe compose(Function, ? extends MaybeSource> transformer) {
return wrap(to(transformer));
}
/**
* Registers an {@link Action} to be called when this Maybe invokes either
* {@link MaybeObserver#onComplete onSuccess},
* {@link MaybeObserver#onComplete onComplete} or {@link MaybeObserver#onError onError}.
*
* 
*
* - Scheduler:
* - {@code doAfterTerminate} does not operate by default on a particular {@link Scheduler}.
*
*
* @param onAfterTerminate
* an {@link Action} to be invoked when the source Publisher finishes
* @return a Maybe that emits the same items as the source Maybe, then invokes the
* {@link Action}
* @see ReactiveX operators documentation: Do
*/
@SchedulerSupport(SchedulerSupport.NONE)
public final Maybe doAfterTerminate(Action onAfterTerminate) {
return RxJavaPlugins.onAssembly(new MaybePeek(this,
Functions.emptyConsumer(), // onSubscribe
Functions.emptyConsumer(), // onSuccess
Functions.emptyConsumer(), // onError
Functions.EMPTY_ACTION, // onComplete
ObjectHelper.requireNonNull(onAfterTerminate, "onAfterTerminate is null"),
Functions.EMPTY_ACTION // dispose
));
}
/**
* Calls the shared runnable if a MaybeObserver subscribed to the current Single
* disposes the common Disposable it received via onSubscribe.
*
* - Scheduler:
* - {@code doOnSubscribe} does not operate by default on a particular {@link Scheduler}.
*
* @param onDispose the runnable called when the subscription is cancelled (disposed)
* @return the new Maybe instance
*/
public final Maybe doOnDispose(final Action onDispose) {
return RxJavaPlugins.onAssembly(new MaybePeek(this,
Functions.emptyConsumer(), // onSubscribe
Functions.emptyConsumer(), // onSuccess
Functions.emptyConsumer(), // onError
Functions.EMPTY_ACTION, // onComplete
Functions.EMPTY_ACTION, // (onSuccess | onError | onComplete) after
ObjectHelper.requireNonNull(onDispose, "onDispose is null")
));
}
/**
* Modifies the source Publisher so that it invokes an action when it calls {@code onCompleted}.
*
* 
*
* - Scheduler:
* - {@code doOnComplete} does not operate by default on a particular {@link Scheduler}.
*
*
* @param onComplete
* the action to invoke when the source Publisher calls {@code onCompleted}
* @return the new Maybe with the side-effecting behavior applied
* @see ReactiveX operators documentation: Do
*/
@SchedulerSupport(SchedulerSupport.NONE)
public final Maybe doOnComplete(Action onComplete) {
return RxJavaPlugins.onAssembly(new MaybePeek(this,
Functions.emptyConsumer(), // onSubscribe
Functions.emptyConsumer(), // onSuccess
Functions.emptyConsumer(), // onError
ObjectHelper.requireNonNull(onComplete, "onComplete is null"),
Functions.EMPTY_ACTION, // (onSuccess | onError | onComplete)
Functions.EMPTY_ACTION // dispose
));
}
/**
* Calls the shared consumer with the error sent via onError for each
* MaybeObserver that subscribes to the current Single.
*
* - Scheduler:
* - {@code doOnSubscribe} does not operate by default on a particular {@link Scheduler}.
*
* @param onError the consumer called with the success value of onError
* @return the new Maybe instance
*/
public final Maybe doOnError(final Consumer onError) {
return RxJavaPlugins.onAssembly(new MaybePeek(this,
Functions.emptyConsumer(), // onSubscribe
Functions.emptyConsumer(), // onSuccess
ObjectHelper.requireNonNull(onError, "onError is null"),
Functions.EMPTY_ACTION, // onComplete
Functions.EMPTY_ACTION, // (onSuccess | onError | onComplete)
Functions.EMPTY_ACTION // dispose
));
}
/**
* Calls the shared consumer with the Disposable sent through the onSubscribe for each
* MaybeObserver that subscribes to the current Single.
*
* - Scheduler:
* - {@code doOnSubscribe} does not operate by default on a particular {@link Scheduler}.
*
* @param onSubscribe the consumer called with the Disposable sent via onSubscribe
* @return the new Maybe instance
*/
public final Maybe doOnSubscribe(final Consumer onSubscribe) {
return RxJavaPlugins.onAssembly(new MaybePeek(this,
ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null"),
Functions.emptyConsumer(), // onSuccess
Functions.emptyConsumer(), // onError
Functions.EMPTY_ACTION, // onComplete
Functions.EMPTY_ACTION, // (onSuccess | onError | onComplete)
Functions.EMPTY_ACTION // dispose
));
}
/**
* Calls the shared consumer with the success value sent via onSuccess for each
* MaybeObserver that subscribes to the current Single.
*
* - Scheduler:
* - {@code doOnSubscribe} does not operate by default on a particular {@link Scheduler}.
*
* @param onSuccess the consumer called with the success value of onSuccess
* @return the new Maybe instance
*/
public final Maybe doOnSuccess(final Consumer onSuccess) {
return RxJavaPlugins.onAssembly(new MaybePeek(this,
Functions.emptyConsumer(), // onSubscribe
ObjectHelper.requireNonNull(onSuccess, "onSubscribe is null"),
Functions.emptyConsumer(), // onError
Functions.EMPTY_ACTION, // onComplete
Functions.EMPTY_ACTION, // (onSuccess | onError | onComplete)
Functions.EMPTY_ACTION // dispose
));
}
/**
* Filters the success item of the Maybe via a predicate function and emitting it if the predicate
* returns true, completing otherwise.
*
* 
*
* - Scheduler:
* - {@code filter} does not operate by default on a particular {@link Scheduler}.
*
*
* @param predicate
* a function that evaluates the item emitted by the source Maybe, returning {@code true}
* if it passes the filter
* @return a Maybe that emit the item emitted by the source Maybe that the filter
* evaluates as {@code true}
* @see ReactiveX operators documentation: Filter
*/
@SchedulerSupport(SchedulerSupport.NONE)
public final Maybe filter(Predicate predicate) {
ObjectHelper.requireNonNull(predicate, "predicate is null");
return RxJavaPlugins.onAssembly(new MaybeFilter(this, predicate));
}
/**
* Returns a Maybe that is based on applying a specified function to the item emitted by the source Single,
* where that function returns a MaybeSource.
*
* 
*
* - Scheduler:
* - {@code flatMap} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the result value type
* @param mapper
* a function that, when applied to the item emitted by the source Maybe, returns a MaybeSource
* @return the Maybe returned from {@code func} when applied to the item emitted by the source Maybe
* @see ReactiveX operators documentation: FlatMap
*/
public final Maybe flatMap(Function> mapper) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
return RxJavaPlugins.onAssembly(new MaybeFlatMap(this, mapper));
}
/**
* Maps the onSuccess, onError or onComplete signals of this Maybe into MaybeSource and emits that
* MaybeSource's signals
*
* 
*
* - Scheduler:
* - {@code flatMap} does not operate by default on a particular {@link Scheduler}.
*
*
* @param
* the result type
* @param onSuccessMapper
* a function that returns a MaybeSource to merge for the success item emitted by this Maybe
* @param onErrorMapper
* a function that returns a MaybeSource to merge for an onError notification from this Maybe
* @param onCompleteSupplier
* a function that returns a MaybeSource to merge for an onCompleted notification this Maybe
* @return the new Maybe instance
* @see ReactiveX operators documentation: FlatMap
*/
@SchedulerSupport(SchedulerSupport.NONE)
public final Maybe flatMap(
Function> onSuccessMapper,
Function> onErrorMapper,
Callable> onCompleteSupplier) {
ObjectHelper.requireNonNull(onSuccessMapper, "onNextMapper is null");
ObjectHelper.requireNonNull(onErrorMapper, "onErrorMapper is null");
ObjectHelper.requireNonNull(onCompleteSupplier, "onCompleteSupplier is null");
return new MaybeFlatMapNotification(this, onSuccessMapper, onErrorMapper, onCompleteSupplier);
}
/**
* Ignores the item emitted by the source Maybe and only calls {@code onCompleted} or {@code onError}.
*
* 
*
* - Scheduler:
* - {@code ignoreElement} does not operate by default on a particular {@link Scheduler}.
*
*
* @return an empty Maybe that only calls {@code onComplete} or {@code onError}, based on which one is
* called by the source Publisher
* @see ReactiveX operators documentation: IgnoreElements
*/
@SchedulerSupport(SchedulerSupport.NONE)
public final Maybe ignoreElement() {
return RxJavaPlugins.onAssembly(new MaybeIgnoreElement(this));
}
/**
* Lifts a function to the current Maybe and returns a new Maybe that when subscribed to will pass the
* values of the current Maybe through the MaybeOperator function.
*
* In other words, this allows chaining TaskExecutors together on a Maybe for acting on the values within
* the Maybe.
*
* {@code task.map(...).filter(...).lift(new OperatorA()).lift(new OperatorB(...)).subscribe() }
*
* If the operator you are creating is designed to act on the item emitted by a source Maybe, use
* {@code lift}. If your operator is designed to transform the source Maybe as a whole (for instance, by
* applying a particular set of existing RxJava operators to it) use {@link #compose}.
*
* - Scheduler:
* - {@code lift} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the downstream's value type (output)
* @param lift
* the MaybeOperator that implements the Maybe-operating function to be applied to the source Maybe
* @return a Maybe that is the result of applying the lifted Operator to the source Maybe
* @see RxJava wiki: Implementing Your Own Operators
*/
public final Maybe lift(final MaybeOperator lift) {
ObjectHelper.requireNonNull(lift, "onLift is null");
return RxJavaPlugins.onAssembly(new MaybeLift(this, lift));
}
/**
* Returns a Maybe that applies a specified function to the item emitted by the source Maybe and
* emits the result of this function application.
*
* 
*
* - Scheduler:
* - {@code map} does not operate by default on a particular {@link Scheduler}.
*
*
* @param the result value type
* @param mapper
* a function to apply to the item emitted by the Maybe
* @return a Maybe that emits the item from the source Maybe, transformed by the specified function
* @see ReactiveX operators documentation: Map
*/
public final Maybe map(Function mapper) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
return RxJavaPlugins.onAssembly(new MaybeMap(this, mapper));
}
/**
* Wraps a Maybe to emit its item (or notify of its error) on a specified {@link Scheduler},
* asynchronously.
*
* 
*
* - Scheduler:
* - you specify which {@link Scheduler} this operator will use
*
*
* @param scheduler
* the {@link Scheduler} to notify subscribers on
* @return the new Maybe instance that its subscribers are notified on the specified
* {@link Scheduler}
* @see ReactiveX operators documentation: ObserveOn
* @see RxJava Threading Examples
* @see #subscribeOn
*/
public final Maybe observeOn(final Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new MaybeObserveOn(this, scheduler));
}
/**
* Calls the specified converter function with the current Maybe instance
* during assembly time and returns its result.
*
* - Scheduler:
* - {@code to} does not operate by default on a particular {@link Scheduler}.
*
* @param the result type
* @param convert the function that is called with the current Maybe instance during
* assembly time that should return some value to be the result
*
* @return the value returned by the convert function
*/
public final R to(Function, R> convert) {
try {
return convert.apply(this);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
throw ExceptionHelper.wrapOrThrow(ex);
}
}
/**
* Converts this Maybe into an Completable instance composing cancellation
* through and dropping a success value if emitted.
*
* - Scheduler:
* - {@code create} does not operate by default on a particular {@link Scheduler}.
*
* @return the new Completable instance
*/
public final Completable toCompletable() {
return RxJavaPlugins.onAssembly(new MaybeToCompletable(this));
}
/**
* Converts this Maybe into a backpressure-aware Flowable instance composing cancellation
* through.
*
* - Scheduler:
* - {@code create} does not operate by default on a particular {@link Scheduler}.
*
* @return the new Flowable instance
*/
public final Flowable toFlowable() {
return RxJavaPlugins.onAssembly(new MaybeToFlowable(this));
}
/**
* Converts this Maybe into an Observable instance composing cancellation
* through.
*
* - Scheduler:
* - {@code create} does not operate by default on a particular {@link Scheduler}.
*
* @return the new Observable instance
*/
public final Observable toObservable() {
return RxJavaPlugins.onAssembly(new MaybeToObservable(this));
}
/**
* Converts this Maybe into an Single instance composing cancellation
* through and turing an empty Maybe into a signal of NoSuchElementException.
*
* - Scheduler:
* - {@code create} does not operate by default on a particular {@link Scheduler}.
*
* @return the new Single instance
*/
public final Single toSingle() {
return RxJavaPlugins.onAssembly(new MaybeToSingle(this));
}
@Override
public final void subscribe(MaybeObserver observer) {
ObjectHelper.requireNonNull(observer, "observer is null");
observer = RxJavaPlugins.onSubscribe(this, observer);
ObjectHelper.requireNonNull(observer, "observer returned by the RxJavaPlugins hook is null");
try {
subscribeActual(observer);
} catch (NullPointerException ex) {
throw ex;
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
NullPointerException npe = new NullPointerException("subscribeActual failed");
npe.initCause(ex);
throw npe;
}
}
/**
* Override this method in subclasses to handle the incoming MaybeObservers.
* @param observer the MaybeObserver to handle, not null
*/
protected abstract void subscribeActual(MaybeObserver observer);
/**
* Asynchronously subscribes subscribers to this Maybe on the specified {@link Scheduler}.
*
* 
*
* - Scheduler:
* - you specify which {@link Scheduler} this operator will use
*
*
* @param scheduler
* the {@link Scheduler} to perform subscription actions on
* @return the new Maybe instance that its subscriptions happen on the specified {@link Scheduler}
* @see ReactiveX operators documentation: SubscribeOn
* @see RxJava Threading Examples
* @see #observeOn
*/
public final Maybe subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new MaybeSubscribeOn(this, scheduler));
}
// ------------------------------------------------------------------
// Test helper
// ------------------------------------------------------------------
/**
* Creates a TestSubscriber and subscribes
* it to this Maybe.
* @return the new TestSubscriber instance
*/
public final TestSubscriber test() {
TestSubscriber ts = new TestSubscriber();
toFlowable().subscribe(ts);
return ts;
}
/**
* Creates a TestSubscriber optionally in cancelled state, then subscribes it to this Maybe.
* @param cancelled if true, the TestSubscriber will be cancelled before subscribing to this
* Maybe.
* @return the new TestSubscriber instance
*/
public final TestSubscriber test(boolean cancelled) {
TestSubscriber ts = new TestSubscriber();
if (cancelled) {
ts.cancel();
}
toFlowable().subscribe(ts);
return ts;
}
}