signature:
public class Completable { /** * Returns a Completable which given a Publisher and when this Completable emits an error , delivers
 * that error through a Flowable and the Publisher should signal a value indicating a retry in response
 * or a terminal event indicating a termination .
 * < img width = " 640 " height = " 586 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . retryWhen . png " alt = " " >
 * Note that the inner { @ code Publisher } returned by the handler function should signal
 * either { @ code onNext } , { @ code onError } or { @ code onComplete } in response to the received
 * { @ code Throwable } to indicate the operator should retry or terminate . If the upstream to
 * the operator is asynchronous , signalling onNext followed by onComplete immediately may
 * result in the sequence to be completed immediately . Similarly , if this inner
 * { @ code Publisher } signals { @ code onError } or { @ code onComplete } while the upstream is
 * active , the sequence is terminated with the same signal immediately .
 * The following example demonstrates how to retry an asynchronous source with a delay :
 * < pre > < code >
 * Completable . timer ( 1 , TimeUnit . SECONDS )
 * . doOnSubscribe ( s - & gt ; System . out . println ( " subscribing " ) )
 * . doOnComplete ( ( ) - & gt ; { throw new RuntimeException ( ) ; } )
 * . retryWhen ( errors - & gt ; {
 * AtomicInteger counter = new AtomicInteger ( ) ;
 * return errors
 * . takeWhile ( e - & gt ; counter . getAndIncrement ( ) ! = 3)
 * . flatMap ( e - & gt ; {
 * System . out . println ( " delay retry by " + counter . get ( ) + " second ( s ) " ) ;
 * return Flowable . timer ( counter . get ( ) , TimeUnit . SECONDS ) ;
 * . blockingAwait ( ) ;
 * < / code > < / pre >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code retryWhen } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param handler the handler that receives a Flowable delivering Throwables and should return a Publisher that
 * emits items to indicate retries or emits terminal events to indicate termination .
 * @ return the new Completable instance
 * @ throws NullPointerException if handler is null */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable retryWhen ( Function < ? super Flowable < Throwable > , ? extends Publisher < ? > > handler ) { } }

implementation: return fromPublisher ( toFlowable ( ) . retryWhen ( handler ) ) ;


signature:
public class Completable { /** * Returns an Observable which first delivers the events
 * of the other Observable then runs this CompletableConsumable .
 * < img width = " 640 " height = " 289 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . startWith . o . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code startWith } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type
 * @ param other the other Observable to run first
 * @ return the new Observable instance
 * @ throws NullPointerException if other is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final < T > Observable < T > startWith ( Observable < T > other ) { } }

implementation: ObjectHelper . requireNonNull ( other , "other is null" ) ; return other . concatWith ( this . < T > toObservable ( ) ) ;


signature:
public class Completable { /** * Subscribes to this CompletableConsumable and returns a Disposable which can be used to dispose
 * the subscription .
 * < img width = " 640 " height = " 352 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . subscribe . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code subscribe } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ return the Disposable that allows disposing the subscription */
@ SchedulerSupport ( SchedulerSupport . NONE ) public final Disposable subscribe ( ) { } }

implementation: EmptyCompletableObserver observer = new EmptyCompletableObserver ( ) ; subscribe ( observer ) ; return observer ;


signature:
public class Completable { /** * Subscribes a given CompletableObserver ( subclass ) to this Completable and returns the given
 * CompletableObserver as is .
 * < img width = " 640 " height = " 349 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . subscribeWith . png " alt = " " >
 * < p > Usage example :
 * < pre > < code >
 * Completable source = Completable . complete ( ) . delay ( 1 , TimeUnit . SECONDS ) ;
 * CompositeDisposable composite = new CompositeDisposable ( ) ;
 * DisposableCompletableObserver ds = new DisposableCompletableObserver ( ) {
 * composite . add ( source . subscribeWith ( ds ) ) ;
 * < / code > < / pre >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code subscribeWith } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < E > the type of the CompletableObserver to use and return
 * @ param observer the CompletableObserver ( subclass ) to use and return , not null
 * @ return the input { @ code observer }
 * @ throws NullPointerException if { @ code observer } is null
 * @ since 2.0 */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final < E extends CompletableObserver > E subscribeWith ( E observer ) { } }

implementation: subscribe ( observer ) ; return observer ;


signature:
public class Completable { /** * Returns a Completable which subscribes the child subscriber on the specified scheduler , making
 * sure the subscription side - effects happen on that specific thread of the scheduler .
 * < img width = " 640 " height = " 686 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . subscribeOn . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code subscribeOn } operates on a { @ link Scheduler } you specify . < / dd >
 * < / dl >
 * @ param scheduler the Scheduler to subscribe on
 * @ return the new Completable instance
 * @ throws NullPointerException if scheduler is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . CUSTOM ) public final Completable subscribeOn ( final Scheduler scheduler ) { } }

implementation: ObjectHelper . requireNonNull ( scheduler , "scheduler is null" ) ; return RxJavaPlugins . onAssembly ( new CompletableSubscribeOn ( this , scheduler ) ) ;


signature:
public class Completable { /** * Terminates the downstream if this or the other { @ code Completable }
 * terminates ( wins the termination race ) while disposing the connection to the losing source .
 * < img width = " 640 " height = " 468 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . takeuntil . c . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code takeUntil } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < dt > < b > Error handling : < / b > < / dt >
 * < dd > If both this and the other sources signal an error , only one of the errors
 * is signaled to the downstream and the other error is signaled to the global
 * error handler via { @ link RxJavaPlugins # onError ( Throwable ) } . < / dd >
 * < / dl >
 * < p > History : 2.1.17 - experimental
 * @ param other the other completable source to observe for the terminal signals
 * @ return the new Completable instance
 * @ since 2.2 */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable takeUntil ( CompletableSource other ) { } }

implementation: ObjectHelper . requireNonNull ( other , "other is null" ) ; return RxJavaPlugins . onAssembly ( new CompletableTakeUntilCompletable ( this , other ) ) ;


signature:
public class Completable { /** * Returns a Completable that runs this Completable and emits a TimeoutException in case
 * this Completable doesn ' t complete within the given time .
 * < img width = " 640 " height = " 348 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . timeout . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code timeout } signals the TimeoutException on the { @ code computation } { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param timeout the timeout value
 * @ param unit the timeout unit
 * @ return the new Completable instance
 * @ throws NullPointerException if unit is null */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . COMPUTATION ) public final Completable timeout ( long timeout , TimeUnit unit ) { } }

implementation: return timeout0 ( timeout , unit , Schedulers . computation ( ) , null ) ;


signature:
public class Completable { /** * Returns a Completable that runs this Completable and switches to the other Completable
 * in case this Completable doesn ' t complete within the given time .
 * < img width = " 640 " height = " 308 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . timeout . c . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code timeout } subscribes to the other CompletableSource on
 * the { @ code computation } { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param timeout the timeout value
 * @ param unit the timeout unit
 * @ param other the other Completable instance to switch to in case of a timeout
 * @ return the new Completable instance
 * @ throws NullPointerException if unit or other is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . COMPUTATION ) public final Completable timeout ( long timeout , TimeUnit unit , CompletableSource other ) { } }

implementation: ObjectHelper . requireNonNull ( other , "other is null" ) ; return timeout0 ( timeout , unit , Schedulers . computation ( ) , other ) ;


signature:
public class Completable { /** * Returns a Completable that runs this Completable and emits a TimeoutException in case
 * this Completable doesn ' t complete within the given time while " waiting " on the specified
 * Scheduler .
 * < img width = " 640 " height = " 348 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . timeout . s . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code timeout } signals the TimeoutException on the { @ link Scheduler } you specify . < / dd >
 * < / dl >
 * @ param timeout the timeout value
 * @ param unit the timeout unit
 * @ param scheduler the scheduler to use to wait for completion
 * @ return the new Completable instance
 * @ throws NullPointerException if unit or scheduler is null */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . CUSTOM ) public final Completable timeout ( long timeout , TimeUnit unit , Scheduler scheduler ) { } }

implementation: return timeout0 ( timeout , unit , scheduler , null ) ;


signature:
public class Completable { /** * Returns a Completable that runs this Completable and switches to the other Completable
 * in case this Completable doesn ' t complete within the given time while " waiting " on
 * the specified scheduler .
 * < img width = " 640 " height = " 308 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . timeout . sc . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code timeout } subscribes to the other CompletableSource on
 * the { @ link Scheduler } you specify . < / dd >
 * < / dl >
 * @ param timeout the timeout value
 * @ param unit the timeout unit
 * @ param scheduler the scheduler to use to wait for completion
 * @ param other the other Completable instance to switch to in case of a timeout
 * @ return the new Completable instance
 * @ throws NullPointerException if unit , scheduler or other is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . CUSTOM ) public final Completable timeout ( long timeout , TimeUnit unit , Scheduler scheduler , CompletableSource other ) { } }

implementation: ObjectHelper . requireNonNull ( other , "other is null" ) ; return timeout0 ( timeout , unit , scheduler , other ) ;


signature:
public class Completable { /** * Allows fluent conversion to another type via a function callback .
 * < img width = " 640 " height = " 751 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . to . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code to } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < U > the output type
 * @ param converter the function called with this which should return some other value .
 * @ return the converted value
 * @ throws NullPointerException if converter is null */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final < U > U to ( Function < ? super Completable , U > converter ) { } }

implementation: try { return ObjectHelper . requireNonNull ( converter , "converter is null" ) . apply ( this ) ; } catch ( Throwable ex ) { Exceptions . throwIfFatal ( ex ) ; throw ExceptionHelper . wrapOrThrow ( ex ) ; }


signature:
public class Completable { /** * Returns a Flowable which when subscribed to subscribes to this Completable and
 * relays the terminal events to the subscriber .
 * < img width = " 640 " height = " 585 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . toFlowable . png " alt = " " >
 * < dl >
 * < dt > < b > Backpressure : < / b > < / dt >
 * < dd > The returned { @ code Flowable } honors the backpressure of the downstream consumer . < / dd >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code toFlowable } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type
 * @ return the new Flowable instance */
@ CheckReturnValue @ SuppressWarnings ( "unchecked" ) @ BackpressureSupport ( BackpressureKind . FULL ) @ SchedulerSupport ( SchedulerSupport . NONE ) public final < T > Flowable < T > toFlowable ( ) { } }

implementation: if ( this instanceof FuseToFlowable ) { return ( ( FuseToFlowable < T > ) this ) . fuseToFlowable ( ) ; } return RxJavaPlugins . onAssembly ( new CompletableToFlowable < T > ( this ) ) ;


signature:
public class Completable { /** * Converts this Completable into a { @ link Maybe } .
 * < img width = " 640 " height = " 585 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . toMaybe . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code toMaybe } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type
 * @ return a { @ link Maybe } that only calls { @ code onComplete } or { @ code onError } , based on which one is
 * called by the source Completable . */
@ CheckReturnValue @ SuppressWarnings ( "unchecked" ) @ SchedulerSupport ( SchedulerSupport . NONE ) public final < T > Maybe < T > toMaybe ( ) { } }

implementation: if ( this instanceof FuseToMaybe ) { return ( ( FuseToMaybe < T > ) this ) . fuseToMaybe ( ) ; } return RxJavaPlugins . onAssembly ( new MaybeFromCompletable < T > ( this ) ) ;


signature:
public class ScalarXMapZHelper { /** * Try subscribing to a { @ link CompletableSource } mapped from
 * a scalar source ( which implements { @ link Callable } ) .
 * @ param < T > the upstream value type
 * @ param source the source reactive type ( { @ code Flowable } or { @ code Observable } )
 * possibly implementing { @ link Callable } .
 * @ param mapper the function that turns the scalar upstream value into a
 * { @ link CompletableSource }
 * @ param observer the consumer to subscribe to the mapped { @ link CompletableSource }
 * @ return true if a subscription did happen and the regular path should be skipped */
static < T > boolean tryAsCompletable ( Object source , Function < ? super T , ? extends CompletableSource > mapper , CompletableObserver observer ) { } }

implementation: if ( source instanceof Callable ) { @ SuppressWarnings ( "unchecked" ) Callable < T > call = ( Callable < T > ) source ; CompletableSource cs = null ; try { T item = call . call ( ) ; if ( item != null ) { cs = ObjectHelper . requireNonNull ( mapper . apply ( item ) , "The mapper returned a null CompletableSource" ) ; } } catch ( Throwable ex ) { Exceptions . throwIfFatal ( ex ) ; EmptyDisposable . error ( ex , observer ) ; return true ; } if ( cs == null ) { EmptyDisposable . complete ( observer ) ; } else { cs . subscribe ( observer ) ; } return true ; } return false ;


signature:
public class ScalarXMapZHelper { /** * Try subscribing to a { @ link MaybeSource } mapped from
 * a scalar source ( which implements { @ link Callable } ) .
 * @ param < T > the upstream value type
 * @ param source the source reactive type ( { @ code Flowable } or { @ code Observable } )
 * possibly implementing { @ link Callable } .
 * @ param mapper the function that turns the scalar upstream value into a
 * { @ link MaybeSource }
 * @ param observer the consumer to subscribe to the mapped { @ link MaybeSource }
 * @ return true if a subscription did happen and the regular path should be skipped */
static < T , R > boolean tryAsMaybe ( Object source , Function < ? super T , ? extends MaybeSource < ? extends R > > mapper , Observer < ? super R > observer ) { } }

implementation: if ( source instanceof Callable ) { @ SuppressWarnings ( "unchecked" ) Callable < T > call = ( Callable < T > ) source ; MaybeSource < ? extends R > cs = null ; try { T item = call . call ( ) ; if ( item != null ) { cs = ObjectHelper . requireNonNull ( mapper . apply ( item ) , "The mapper returned a null MaybeSource" ) ; } } catch ( Throwable ex ) { Exceptions . throwIfFatal ( ex ) ; EmptyDisposable . error ( ex , observer ) ; return true ; } if ( cs == null ) { EmptyDisposable . complete ( observer ) ; } else { cs . subscribe ( MaybeToObservable . create ( observer ) ) ; } return true ; } return false ;


signature:
public class LinkedArrayList { /** * Adds a new element to this list .
 * @ param o the object to add , nulls are accepted */
public void add ( Object o ) { } }

implementation:
// if no value yet , create the first array
if ( size == 0 ) { head = new Object [ capacityHint + 1 ] ; tail = head ; head [ 0 ] = o ; indexInTail = 1 ; size = 1 ; } else // if the tail is full , create a new tail and link
if ( indexInTail == capacityHint ) { Object [ ] t = new Object [ capacityHint + 1 ] ; t [ 0 ] = o ; tail [ capacityHint ] = t ; tail = t ; indexInTail = 1 ; size ++ ; } else { tail [ indexInTail ] = o ; indexInTail ++ ; size ++ ; }


signature:
public class SchedulerPoolFactory { /** * Stops the purge thread . */
public static void shutdown ( ) { } }

implementation: ScheduledExecutorService exec = PURGE_THREAD . getAndSet ( null ) ; if ( exec != null ) { exec . shutdownNow ( ) ; } POOLS . clear ( ) ;


signature:
public class SchedulerPoolFactory { /** * Creates a ScheduledExecutorService with the given factory .
 * @ param factory the thread factory
 * @ return the ScheduledExecutorService */
public static ScheduledExecutorService create ( ThreadFactory factory ) { } }

implementation: final ScheduledExecutorService exec = Executors . newScheduledThreadPool ( 1 , factory ) ; tryPutIntoPool ( PURGE_ENABLED , exec ) ; return exec ;


signature:
public class SubscriptionArbiter { /** * Atomically sets a new subscription .
 * @ param s the subscription to set , not null ( verified ) */
public final void setSubscription ( Subscription s ) { } }

implementation: if ( cancelled ) { s . cancel ( ) ; return ; } ObjectHelper . requireNonNull ( s , "s is null" ) ; if ( get ( ) == 0 && compareAndSet ( 0 , 1 ) ) { Subscription a = actual ; if ( a != null && cancelOnReplace ) { a . cancel ( ) ; } actual = s ; long r = requested ; if ( decrementAndGet ( ) != 0 ) { drainLoop ( ) ; } if ( r != 0L ) { s . request ( r ) ; } return ; } Subscription a = missedSubscription . getAndSet ( s ) ; if ( a != null && cancelOnReplace ) { a . cancel ( ) ; } drain ( ) ;


signature:
public class SerializedSubject { /** * Loops until all notifications in the queue has been processed . */
void emitLoop ( ) { } }

implementation: for ( ; ; ) { AppendOnlyLinkedArrayList < Object > q ; synchronized ( this ) { q = queue ; if ( q == null ) { emitting = false ; return ; } queue = null ; } q . forEachWhile ( this ) ; }


signature:
public class FlowableLastSingle { /** * TODO fuse back to Flowable */
@ Override protected void subscribeActual ( SingleObserver < ? super T > observer ) { } }

implementation: source . subscribe ( new LastSubscriber < T > ( observer , defaultItem ) ) ;


signature:
public class Exceptions { /** * Throws a particular { @ code Throwable } only if it belongs to a set of " fatal " error varieties . These
 * varieties are as follows :
 * < ul >
 * < li > { @ code VirtualMachineError } < / li >
 * < li > { @ code ThreadDeath } < / li >
 * < li > { @ code LinkageError } < / li >
 * < / ul >
 * This can be useful if you are writing an operator that calls user - supplied code , and you want to
 * notify subscribers of errors encountered in that code by calling their { @ code onError } methods , but only
 * if the errors are not so catastrophic that such a call would be futile , in which case you simply want to
 * rethrow the error .
 * @ param t
 * the { @ code Throwable } to test and perhaps throw
 * @ see < a href = " https : / / github . com / ReactiveX / RxJava / issues / 748 # issuecomment - 32471495 " > RxJava : StackOverflowError is swallowed ( Issue # 748 ) < / a > */
public static void throwIfFatal ( @ NonNull Throwable t ) { } }

implementation:
// values here derived from https : / / github . com / ReactiveX / RxJava / issues / 748 # issuecomment - 32471495
if ( t instanceof VirtualMachineError ) { throw ( VirtualMachineError ) t ; } else if ( t instanceof ThreadDeath ) { throw ( ThreadDeath ) t ; } else if ( t instanceof LinkageError ) { throw ( LinkageError ) t ; }


signature:
public class ExceptionHelper { /** * If the provided Throwable is an Error this method
 * throws it , otherwise returns a RuntimeException wrapping the error
 * if that error is a checked exception .
 * @ param error the error to wrap or throw
 * @ return the ( wrapped ) error */
public static RuntimeException wrapOrThrow ( Throwable error ) { } }

implementation: if ( error instanceof Error ) { throw ( Error ) error ; } if ( error instanceof RuntimeException ) { return ( RuntimeException ) error ; } return new RuntimeException ( error ) ;


signature:
public class ExceptionHelper { /** * Returns a flattened list of Throwables from tree - like CompositeException chain .
 * @ param t the starting throwable
 * @ return the list of Throwables flattened in a depth - first manner */
public static List < Throwable > flatten ( Throwable t ) { } }

implementation: List < Throwable > list = new ArrayList < Throwable > ( ) ; ArrayDeque < Throwable > deque = new ArrayDeque < Throwable > ( ) ; deque . offer ( t ) ; while ( ! deque . isEmpty ( ) ) { Throwable e = deque . removeFirst ( ) ; if ( e instanceof CompositeException ) { CompositeException ce = ( CompositeException ) e ; List < Throwable > exceptions = ce . getExceptions ( ) ; for ( int i = exceptions . size ( ) - 1 ; i >= 0 ; i -- ) { deque . offerFirst ( exceptions . get ( i ) ) ; } } else { list . add ( e ) ; } } return list ;


signature:
public class ExceptionHelper { /** * Workaround for Java 6 not supporting throwing a final Throwable from a catch block .
 * @ param < E > the generic exception type
 * @ param e the Throwable error to return or throw
 * @ return the Throwable e if it is a subclass of Exception
 * @ throws E the generic exception thrown */
@ SuppressWarnings ( "unchecked" ) public static < E extends Throwable > Exception throwIfThrowable ( Throwable e ) throws E { } }

implementation: if ( e instanceof Exception ) { return ( Exception ) e ; } throw ( E ) e ;


signature:
public class ObservableCache { /** * Atomically adds the consumer to the { @ link # observers } copy - on - write array
 * if the source has not yet terminated .
 * @ param consumer the consumer to add */
void add ( CacheDisposable < T > consumer ) { } }

implementation: for ( ; ; ) { CacheDisposable < T > [ ] current = observers . get ( ) ; if ( current == TERMINATED ) { return ; } int n = current . length ; @ SuppressWarnings ( "unchecked" ) CacheDisposable < T > [ ] next = new CacheDisposable [ n + 1 ] ; System . arraycopy ( current , 0 , next , 0 , n ) ; next [ n ] = consumer ; if ( observers . compareAndSet ( current , next ) ) { return ; } }


signature:
public class ObservableCache { /** * Atomically removes the consumer from the { @ link # observers } copy - on - write array .
 * @ param consumer the consumer to remove */
@ SuppressWarnings ( "unchecked" ) void remove ( CacheDisposable < T > consumer ) { } }

implementation: for ( ; ; ) { CacheDisposable < T > [ ] current = observers . get ( ) ; int n = current . length ; if ( n == 0 ) { return ; } int j = - 1 ; for ( int i = 0 ; i < n ; i ++ ) { if ( current [ i ] == consumer ) { j = i ; break ; } } if ( j < 0 ) { return ; } CacheDisposable < T > [ ] next ; if ( n == 1 ) { next = EMPTY ; } else { next = new CacheDisposable [ n - 1 ] ; System . arraycopy ( current , 0 , next , 0 , j ) ; System . arraycopy ( current , j + 1 , next , j , n - j - 1 ) ; } if ( observers . compareAndSet ( current , next ) ) { return ; } }


signature:
public class ObservableCache { /** * Replays the contents of this cache to the given consumer based on its
 * current state and number of items requested by it .
 * @ param consumer the consumer to continue replaying items to */
void replay ( CacheDisposable < T > consumer ) { } }

implementation:
// make sure there is only one replay going on at a time
if ( consumer . getAndIncrement ( ) != 0 ) { return ; } // see if there were more replay request in the meantime
int missed = 1 ; // read out state into locals upfront to avoid being re - read due to volatile reads
long index = consumer . index ; int offset = consumer . offset ; Node < T > node = consumer . node ; Observer < ? super T > downstream = consumer . downstream ; int capacity = capacityHint ; for ( ; ; ) { // if the consumer got disposed , clear the node and quit
if ( consumer . disposed ) { consumer . node = null ; return ; } // first see if the source has terminated , read order matters !
boolean sourceDone = done ; // and if the number of items is the same as this consumer has received
boolean empty = size == index ; // if the source is done and we have all items so far , terminate the consumer
if ( sourceDone && empty ) { // release the node object to avoid leaks through retained consumers
consumer . node = null ; // if error is not null then the source failed
Throwable ex = error ; if ( ex != null ) { downstream . onError ( ex ) ; } else { downstream . onComplete ( ) ; } return ; } // there are still items not sent to the consumer
if ( ! empty ) { // if the offset in the current node has reached the node capacity
if ( offset == capacity ) { // switch to the subsequent node
node = node . next ; // reset the in - node offset
offset = 0 ; } // emit the cached item
downstream . onNext ( node . values [ offset ] ) ; // move the node offset forward
offset ++ ; // move the total consumed item count forward
index ++ ; // retry for the next item / terminal event if any
continue ; } // commit the changed references back
consumer . index = index ; consumer . offset = offset ; consumer . node = node ; // release the changes and see if there were more replay request in the meantime
missed = consumer . addAndGet ( - missed ) ; if ( missed == 0 ) { break ; } }


signature:
public class OperatorMergePerf { /** * flatMap */
@ Benchmark public void oneStreamOfNthatMergesIn1 ( final InputMillion input ) throws InterruptedException { } }

implementation: Flowable < Flowable < Integer > > os = Flowable . range ( 1 , input . size ) . map ( new Function < Integer , Flowable < Integer > > ( ) { @ Override public Flowable < Integer > apply ( Integer v ) { return Flowable . just ( v ) ; } } ) ; PerfSubscriber o = input . newLatchedObserver ( ) ; Flowable . merge ( os ) . subscribe ( o ) ; if ( input . size == 1 ) { while ( o . latch . getCount ( ) != 0 ) { } } else { o . latch . await ( ) ; }


signature:
public class CompositeDisposable { /** * Adds a disposable to this container or disposes it if the
 * container has been disposed .
 * @ param disposable the disposable to add , not null
 * @ return true if successful , false if this container has been disposed
 * @ throws NullPointerException if { @ code disposable } is null */
@ Override public boolean add ( @ NonNull Disposable disposable ) { } }

implementation: ObjectHelper . requireNonNull ( disposable , "disposable is null" ) ; if ( ! disposed ) { synchronized ( this ) { if ( ! disposed ) { OpenHashSet < Disposable > set = resources ; if ( set == null ) { set = new OpenHashSet < Disposable > ( ) ; resources = set ; } set . add ( disposable ) ; return true ; } } } disposable . dispose ( ) ; return false ;


signature:
public class CompositeDisposable { /** * Atomically adds the given array of Disposables to the container or
 * disposes them all if the container has been disposed .
 * @ param disposables the array of Disposables
 * @ return true if the operation was successful , false if the container has been disposed
 * @ throws NullPointerException if { @ code disposables } or any of its array items is null */
public boolean addAll ( @ NonNull Disposable ... disposables ) { } }

implementation: ObjectHelper . requireNonNull ( disposables , "disposables is null" ) ; if ( ! disposed ) { synchronized ( this ) { if ( ! disposed ) { OpenHashSet < Disposable > set = resources ; if ( set == null ) { set = new OpenHashSet < Disposable > ( disposables . length + 1 ) ; resources = set ; } for ( Disposable d : disposables ) { ObjectHelper . requireNonNull ( d , "A Disposable in the disposables array is null" ) ; set . add ( d ) ; } return true ; } } } for ( Disposable d : disposables ) { d . dispose ( ) ; } return false ;


signature:
public class CompositeDisposable { /** * Removes and disposes the given disposable if it is part of this
 * container .
 * @ param disposable the disposable to remove and dispose , not null
 * @ return true if the operation was successful */
@ Override public boolean remove ( @ NonNull Disposable disposable ) { } }

implementation: if ( delete ( disposable ) ) { disposable . dispose ( ) ; return true ; } return false ;


signature:
public class CompositeDisposable { /** * Removes ( but does not dispose ) the given disposable if it is part of this
 * container .
 * @ param disposable the disposable to remove , not null
 * @ return true if the operation was successful
 * @ throws NullPointerException if { @ code disposable } is null */
@ Override public boolean delete ( @ NonNull Disposable disposable ) { } }

implementation: ObjectHelper . requireNonNull ( disposable , "disposables is null" ) ; if ( disposed ) { return false ; } synchronized ( this ) { if ( disposed ) { return false ; } OpenHashSet < Disposable > set = resources ; if ( set == null || ! set . remove ( disposable ) ) { return false ; } } return true ;


signature:
public class CompositeDisposable { /** * Atomically clears the container , then disposes all the previously contained Disposables . */
public void clear ( ) { } }

implementation: if ( disposed ) { return ; } OpenHashSet < Disposable > set ; synchronized ( this ) { if ( disposed ) { return ; } set = resources ; resources = null ; } dispose ( set ) ;


signature:
public class CompositeDisposable { /** * Returns the number of currently held Disposables .
 * @ return the number of currently held Disposables */
public int size ( ) { } }

implementation: if ( disposed ) { return 0 ; } synchronized ( this ) { if ( disposed ) { return 0 ; } OpenHashSet < Disposable > set = resources ; return set != null ? set . size ( ) : 0 ; }


signature:
public class FlowableBlockingSubscribe { /** * Subscribes to the source and calls the Subscriber methods on the current thread .
 * @ param o the source publisher
 * The cancellation and backpressure is composed through .
 * @ param subscriber the subscriber to forward events and calls to in the current thread
 * @ param < T > the value type */
public static < T > void subscribe ( Publisher < ? extends T > o , Subscriber < ? super T > subscriber ) { } }

implementation: final BlockingQueue < Object > queue = new LinkedBlockingQueue < Object > ( ) ; BlockingSubscriber < T > bs = new BlockingSubscriber < T > ( queue ) ; o . subscribe ( bs ) ; try { for ( ; ; ) { if ( bs . isCancelled ( ) ) { break ; } Object v = queue . poll ( ) ; if ( v == null ) { if ( bs . isCancelled ( ) ) { break ; } BlockingHelper . verifyNonBlocking ( ) ; v = queue . take ( ) ; } if ( bs . isCancelled ( ) ) { break ; } if ( v == BlockingSubscriber . TERMINATED || NotificationLite . acceptFull ( v , subscriber ) ) { break ; } } } catch ( InterruptedException e ) { bs . cancel ( ) ; subscriber . onError ( e ) ; }


signature:
public class FlowableBlockingSubscribe { /** * Runs the source observable to a terminal event , ignoring any values and rethrowing any exception .
 * @ param o the source publisher
 * @ param < T > the value type */
public static < T > void subscribe ( Publisher < ? extends T > o ) { } }

implementation: BlockingIgnoringReceiver callback = new BlockingIgnoringReceiver ( ) ; LambdaSubscriber < T > ls = new LambdaSubscriber < T > ( Functions . emptyConsumer ( ) , callback , callback , Functions . REQUEST_MAX ) ; o . subscribe ( ls ) ; BlockingHelper . awaitForComplete ( callback , ls ) ; Throwable e = callback . error ; if ( e != null ) { throw ExceptionHelper . wrapOrThrow ( e ) ; }


signature:
public class FlowableBlockingSubscribe { /** * Subscribes to the source and calls the given actions on the current thread .
 * @ param o the source publisher
 * @ param onNext the callback action for each source value
 * @ param onError the callback action for an error event
 * @ param onComplete the callback action for the completion event .
 * @ param < T > the value type */
public static < T > void subscribe ( Publisher < ? extends T > o , final Consumer < ? super T > onNext , final Consumer < ? super Throwable > onError , final Action onComplete ) { } }

implementation: ObjectHelper . requireNonNull ( onNext , "onNext is null" ) ; ObjectHelper . requireNonNull ( onError , "onError is null" ) ; ObjectHelper . requireNonNull ( onComplete , "onComplete is null" ) ; subscribe ( o , new LambdaSubscriber < T > ( onNext , onError , onComplete , Functions . REQUEST_MAX ) ) ;


signature:
public class FlowableBlockingSubscribe { /** * Subscribes to the source and calls the given actions on the current thread .
 * @ param o the source publisher
 * @ param onNext the callback action for each source value
 * @ param onError the callback action for an error event
 * @ param onComplete the callback action for the completion event .
 * @ param bufferSize the number of elements to prefetch from the source Publisher
 * @ param < T > the value type */
public static < T > void subscribe ( Publisher < ? extends T > o , final Consumer < ? super T > onNext , final Consumer < ? super Throwable > onError , final Action onComplete , int bufferSize ) { } }

implementation: ObjectHelper . requireNonNull ( onNext , "onNext is null" ) ; ObjectHelper . requireNonNull ( onError , "onError is null" ) ; ObjectHelper . requireNonNull ( onComplete , "onComplete is null" ) ; ObjectHelper . verifyPositive ( bufferSize , "number > 0 required" ) ; subscribe ( o , new BoundedSubscriber < T > ( onNext , onError , onComplete , Functions . boundedConsumer ( bufferSize ) , bufferSize ) ) ;


signature:
public class ResourceSingleObserver { /** * Adds a resource to this ResourceObserver .
 * @ param resource the resource to add
 * @ throws NullPointerException if resource is null */
public final void add ( @ NonNull Disposable resource ) { } }

implementation: ObjectHelper . requireNonNull ( resource , "resource is null" ) ; resources . add ( resource ) ;


signature:
public class ObservableScalarXMap { /** * Tries to subscribe to a possibly Callable source ' s mapped ObservableSource .
 * @ param < T > the input value type
 * @ param < R > the output value type
 * @ param source the source ObservableSource
 * @ param observer the subscriber
 * @ param mapper the function mapping a scalar value into an ObservableSource
 * @ return true if successful , false if the caller should continue with the regular path . */
@ SuppressWarnings ( "unchecked" ) public static < T , R > boolean tryScalarXMapSubscribe ( ObservableSource < T > source , Observer < ? super R > observer , Function < ? super T , ? extends ObservableSource < ? extends R > > mapper ) { } }

implementation: if ( source instanceof Callable ) { T t ; try { t = ( ( Callable < T > ) source ) . call ( ) ; } catch ( Throwable ex ) { Exceptions . throwIfFatal ( ex ) ; EmptyDisposable . error ( ex , observer ) ; return true ; } if ( t == null ) { EmptyDisposable . complete ( observer ) ; return true ; } ObservableSource < ? extends R > r ; try { r = ObjectHelper . requireNonNull ( mapper . apply ( t ) , "The mapper returned a null ObservableSource" ) ; } catch ( Throwable ex ) { Exceptions . throwIfFatal ( ex ) ; EmptyDisposable . error ( ex , observer ) ; return true ; } if ( r instanceof Callable ) { R u ; try { u = ( ( Callable < R > ) r ) . call ( ) ; } catch ( Throwable ex ) { Exceptions . throwIfFatal ( ex ) ; EmptyDisposable . error ( ex , observer ) ; return true ; } if ( u == null ) { EmptyDisposable . complete ( observer ) ; return true ; } ScalarDisposable < R > sd = new ScalarDisposable < R > ( observer , u ) ; observer . onSubscribe ( sd ) ; sd . run ( ) ; } else { r . subscribe ( observer ) ; } return true ; } return false ;


signature:
public class ObservableScalarXMap { /** * Maps a scalar value into an Observable and emits its values .
 * @ param < T > the scalar value type
 * @ param < U > the output value type
 * @ param value the scalar value to map
 * @ param mapper the function that gets the scalar value and should return
 * an ObservableSource that gets streamed
 * @ return the new Observable instance */
public static < T , U > Observable < U > scalarXMap ( T value , Function < ? super T , ? extends ObservableSource < ? extends U > > mapper ) { } }

implementation: return RxJavaPlugins . onAssembly ( new ScalarXMapObservable < T , U > ( value , mapper ) ) ;


signature:
public class MaybeToObservable { /** * Creates a { @ link MaybeObserver } wrapper around a { @ link Observer } .
 * < p > History : 2.1.11 - experimental
 * @ param < T > the value type
 * @ param downstream the downstream { @ code Observer } to talk to
 * @ return the new MaybeObserver instance
 * @ since 2.2 */
public static < T > MaybeObserver < T > create ( Observer < ? super T > downstream ) { } }

implementation: return new MaybeToObservableObserver < T > ( downstream ) ;


signature:
public class Schedulers { /** * Wraps an { @ link Executor } into a new Scheduler instance and delegates { @ code schedule ( ) }
 * calls to it .
 * The tasks scheduled by the returned { @ link Scheduler } and its { @ link Scheduler . Worker }
 * can be optionally interrupted .
 * If the provided executor doesn ' t support any of the more specific standard Java executor
 * APIs , tasks scheduled with a time delay or periodically will use the
 * { @ link # single ( ) } scheduler for the timed waiting
 * before posting the actual task to the given executor .
 * If the provided executor supports the standard Java { @ link ExecutorService } API ,
 * canceling tasks scheduled by this scheduler can be cancelled / interrupted by calling
 * { @ link io . reactivex . disposables . Disposable # dispose ( ) } . In addition , tasks scheduled with
 * a time delay or periodically will use the { @ link # single ( ) } scheduler for the timed waiting
 * before posting the actual task to the given executor .
 * If the provided executor supports the standard Java { @ link ScheduledExecutorService } API ,
 * canceling tasks scheduled by this scheduler can be cancelled / interrupted by calling
 * { @ link io . reactivex . disposables . Disposable # dispose ( ) } . In addition , tasks scheduled with
 * a time delay or periodically will use the provided executor . Note , however , if the provided
 * { @ code ScheduledExecutorService } instance is not single threaded , tasks scheduled
 * with a time delay close to each other may end up executing in different order than
 * the original schedule ( ) call was issued . This limitation may be lifted in a future patch .
 * Starting , stopping and restarting this scheduler is not supported ( no - op ) and the provided
 * executor ' s lifecycle must be managed externally :
 * < pre > < code >
 * ExecutorService exec = Executors . newSingleThreadedExecutor ( ) ;
 * try {
 * Scheduler scheduler = Schedulers . from ( exec , true ) ;
 * Flowable . just ( 1)
 * . subscribeOn ( scheduler )
 * . map ( v - & gt ; v + 1)
 * . observeOn ( scheduler )
 * . blockingSubscribe ( System . out : : println ) ;
 * } finally {
 * exec . shutdown ( ) ;
 * < / code > < / pre >
 * This type of scheduler is less sensitive to leaking { @ link Scheduler . Worker } instances , although
 * not disposing a worker that has timed / delayed tasks not cancelled by other means may leak resources and / or
 * execute those tasks " unexpectedly " .
 * Note that this method returns a new { @ link Scheduler } instance , even for the same { @ link Executor } instance .
 * @ param executor
 * the executor to wrap
 * @ param interruptibleWorker if { @ code true } the tasks submitted to the { @ link Scheduler . Worker } will
 * be interrupted when the task is disposed .
 * @ return the new Scheduler wrapping the Executor
 * @ since 2.2.6 - experimental */
@ NonNull @ Experimental public static Scheduler from ( @ NonNull Executor executor , boolean interruptibleWorker ) { } }

implementation: return new ExecutorScheduler ( executor , interruptibleWorker ) ;


signature:
public class Schedulers { /** * Shuts down the standard Schedulers .
 * < p > The operation is idempotent and thread - safe . */
public static void shutdown ( ) { } }

implementation: computation ( ) . shutdown ( ) ; io ( ) . shutdown ( ) ; newThread ( ) . shutdown ( ) ; single ( ) . shutdown ( ) ; trampoline ( ) . shutdown ( ) ; SchedulerPoolFactory . shutdown ( ) ;


signature:
public class AppendOnlyLinkedArrayList { /** * Append a non - null value to the list .
 * < p > Don ' t add null to the list !
 * @ param value the value to append */
public void add ( T value ) { } }

implementation: final int c = capacity ; int o = offset ; if ( o == c ) { Object [ ] next = new Object [ c + 1 ] ; tail [ c ] = next ; tail = next ; o = 0 ; } tail [ o ] = value ; offset = o + 1 ;


signature:
public class AppendOnlyLinkedArrayList { /** * Loops over all elements of the array until a null element is encountered or
 * the given predicate returns true .
 * @ param consumer the consumer of values that returns true if the forEach should terminate */
@ SuppressWarnings ( "unchecked" ) public void forEachWhile ( NonThrowingPredicate < ? super T > consumer ) { } }

implementation: Object [ ] a = head ; final int c = capacity ; while ( a != null ) { for ( int i = 0 ; i < c ; i ++ ) { Object o = a [ i ] ; if ( o == null ) { break ; } if ( consumer . test ( ( T ) o ) ) { return ; } } a = ( Object [ ] ) a [ c ] ; }


signature:
public class AppendOnlyLinkedArrayList { /** * Interprets the contents as NotificationLite objects and calls
 * the appropriate Subscriber method .
 * @ param < U > the target type
 * @ param subscriber the subscriber to emit the events to
 * @ return true if a terminal event has been reached */
public < U > boolean accept ( Subscriber < ? super U > subscriber ) { } }

implementation: Object [ ] a = head ; final int c = capacity ; while ( a != null ) { for ( int i = 0 ; i < c ; i ++ ) { Object o = a [ i ] ; if ( o == null ) { break ; } if ( NotificationLite . acceptFull ( o , subscriber ) ) { return true ; } } a = ( Object [ ] ) a [ c ] ; } return false ;


signature:
public class AppendOnlyLinkedArrayList { /** * Interprets the contents as NotificationLite objects and calls
 * the appropriate Observer method .
 * @ param < U > the target type
 * @ param observer the observer to emit the events to
 * @ return true if a terminal event has been reached */
public < U > boolean accept ( Observer < ? super U > observer ) { } }

implementation: Object [ ] a = head ; final int c = capacity ; while ( a != null ) { for ( int i = 0 ; i < c ; i ++ ) { Object o = a [ i ] ; if ( o == null ) { break ; } if ( NotificationLite . acceptFull ( o , observer ) ) { return true ; } } a = ( Object [ ] ) a [ c ] ; } return false ;


signature:
public class AppendOnlyLinkedArrayList { /** * Loops over all elements of the array until a null element is encountered or
 * the given predicate returns true .
 * @ param < S > the extra state type
 * @ param state the extra state passed into the consumer
 * @ param consumer the consumer of values that returns true if the forEach should terminate
 * @ throws Exception if the predicate throws */
@ SuppressWarnings ( "unchecked" ) public < S > void forEachWhile ( S state , BiPredicate < ? super S , ? super T > consumer ) throws Exception { } }

implementation: Object [ ] a = head ; final int c = capacity ; for ( ; ; ) { for ( int i = 0 ; i < c ; i ++ ) { Object o = a [ i ] ; if ( o == null ) { return ; } if ( consumer . test ( state , ( T ) o ) ) { return ; } } a = ( Object [ ] ) a [ c ] ; }


signature:
public class Single { /** * Runs multiple SingleSources and signals the events of the first one that signals ( disposing
 * the rest ) .
 * < img width = " 640 " height = " 515 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . amb . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code amb } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type
 * @ param sources the Iterable sequence of sources . A subscription to each source will
 * occur in the same order as in this Iterable .
 * @ return the new Single instance
 * @ since 2.0 */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static < T > Single < T > amb ( final Iterable < ? extends SingleSource < ? extends T > > sources ) { } }

implementation: ObjectHelper . requireNonNull ( sources , "sources is null" ) ; return RxJavaPlugins . onAssembly ( new SingleAmb < T > ( null , sources ) ) ;


signature:
public class Single { /** * Runs multiple SingleSources and signals the events of the first one that signals ( disposing
 * the rest ) .
 * < img width = " 640 " height = " 515 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . ambArray . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code ambArray } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type
 * @ param sources the array of sources . A subscription to each source will
 * occur in the same order as in this array .
 * @ return the new Single instance
 * @ since 2.0 */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) @ SuppressWarnings ( "unchecked" ) public static < T > Single < T > ambArray ( final SingleSource < ? extends T > ... sources ) { } }

implementation: if ( sources . length == 0 ) { return error ( SingleInternalHelper . < T > emptyThrower ( ) ) ; } if ( sources . length == 1 ) { return wrap ( ( SingleSource < T > ) sources [ 0 ] ) ; } return RxJavaPlugins . onAssembly ( new SingleAmb < T > ( sources , null ) ) ;


signature:
public class Single { /** * Concatenate the single values , in a non - overlapping fashion , of the SingleSources provided in
 * an array .
 * < img width = " 640 " height = " 319 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . concatArray . png " alt = " " >
 * < dl >
 * < dt > < b > Backpressure : < / b > < / dt >
 * < dd > The returned { @ code Flowable } honors the backpressure of the downstream consumer . < / dd >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code concatArray } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type
 * @ param sources the array of SingleSource instances
 * @ return the new Flowable instance
 * @ since 2.0 */
@ CheckReturnValue @ NonNull @ BackpressureSupport ( BackpressureKind . FULL ) @ SchedulerSupport ( SchedulerSupport . NONE ) @ SuppressWarnings ( { } }

implementation: "unchecked" , "rawtypes" } ) public static < T > Flowable < T > concatArray ( SingleSource < ? extends T > ... sources ) { return RxJavaPlugins . onAssembly ( new FlowableConcatMap ( Flowable . fromArray ( sources ) , SingleInternalHelper . toFlowable ( ) , 2 , ErrorMode . BOUNDARY ) ) ;


signature:
public class Single { /** * Calls a { @ link Callable } for each individual { @ link SingleObserver } to return the actual { @ link SingleSource } to
 * be subscribed to .
 * < img width = " 640 " height = " 515 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . defer . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code defer } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type
 * @ param singleSupplier the { @ code Callable } that is called for each individual { @ code SingleObserver } and
 * returns a SingleSource instance to subscribe to
 * @ return the new Single instance */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static < T > Single < T > defer ( final Callable < ? extends SingleSource < ? extends T > > singleSupplier ) { } }

implementation: ObjectHelper . requireNonNull ( singleSupplier , "singleSupplier is null" ) ; return RxJavaPlugins . onAssembly ( new SingleDefer < T > ( singleSupplier ) ) ;


signature:
public class Single { /** * Signals a Throwable returned by the callback function for each individual SingleObserver .
 * < img width = " 640 " height = " 283 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . error . c . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code error } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type
 * @ param errorSupplier the callable that is called for each individual SingleObserver and
 * returns a Throwable instance to be emitted .
 * @ return the new Single instance */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static < T > Single < T > error ( final Callable < ? extends Throwable > errorSupplier ) { } }

implementation: ObjectHelper . requireNonNull ( errorSupplier , "errorSupplier is null" ) ; return RxJavaPlugins . onAssembly ( new SingleError < T > ( errorSupplier ) ) ;


signature:
public class Single { /** * Returns a { @ link Single } that invokes passed function and emits its result for each new SingleObserver that subscribes .
 * Allows you to defer execution of passed function until SingleObserver subscribes to the { @ link Single } .
 * It makes passed function " lazy " .
 * Result of the function invocation will be emitted by the { @ link Single } .
 * < img width = " 640 " height = " 467 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . fromCallable . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code fromCallable } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < dt > < b > Error handling : < / b > < / dt >
 * < dd > If the { @ link Callable } throws an exception , the respective { @ link Throwable } is
 * delivered to the downstream via { @ link SingleObserver # onError ( Throwable ) } ,
 * except when the downstream has disposed this { @ code Single } source .
 * In this latter case , the { @ code Throwable } is delivered to the global error handler via
 * { @ link RxJavaPlugins # onError ( Throwable ) } as an { @ link io . reactivex . exceptions . UndeliverableException UndeliverableException } .
 * < / dd >
 * < / dl >
 * @ param callable
 * function which execution should be deferred , it will be invoked when SingleObserver will subscribe to the { @ link Single } .
 * @ param < T >
 * the type of the item emitted by the { @ link Single } .
 * @ return a { @ link Single } whose { @ link SingleObserver } s ' subscriptions trigger an invocation of the given function . */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static < T > Single < T > fromCallable ( final Callable < ? extends T > callable ) { } }

implementation: ObjectHelper . requireNonNull ( callable , "callable is null" ) ; return RxJavaPlugins . onAssembly ( new SingleFromCallable < T > ( callable ) ) ;


signature:
public class Single { /** * Converts a { @ link Future } into a { @ code Single } .
 * < img width = " 640 " height = " 315 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . from . Future . png " alt = " " >
 * You can convert any object that supports the { @ link Future } interface into a Single that emits the return
 * value of the { @ link Future # get } method of that object , by passing the object into the { @ code from }
 * method .
 * < em > Important note : < / em > This Single is blocking ; you cannot dispose it .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code fromFuture } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param future
 * the source { @ link Future }
 * @ param < T >
 * the type of object that the { @ link Future } returns , and also the type of item to be emitted by
 * the resulting { @ code Single }
 * @ return a { @ code Single } that emits the item from the source { @ link Future }
 * @ see < a href = " http : / / reactivex . io / documentation / operators / from . html " > ReactiveX operators documentation : From < / a > */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public static < T > Single < T > fromFuture ( Future < ? extends T > future ) { } }

implementation: return toSingle ( Flowable . < T > fromFuture ( future ) ) ;


signature:
public class Single { /** * Converts a { @ link Future } into a { @ code Single } , with a timeout on the Future .
 * < img width = " 640 " height = " 315 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . from . Future . png " alt = " " >
 * You can convert any object that supports the { @ link Future } interface into a { @ code Single } that emits
 * the return value of the { @ link Future # get } method of that object , by passing the object into the
 * { @ code from } method .
 * < em > Important note : < / em > This { @ code Single } is blocking ; you cannot dispose it .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code fromFuture } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ 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 < T >
 * the type of object that the { @ link Future } returns , and also the type of item to be emitted by
 * the resulting { @ code Single }
 * @ return a { @ code Single } that emits the item from the source { @ link Future }
 * @ see < a href = " http : / / reactivex . io / documentation / operators / from . html " > ReactiveX operators documentation : From < / a > */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public static < T > Single < T > fromFuture ( Future < ? extends T > future , long timeout , TimeUnit unit ) { } }

implementation: return toSingle ( Flowable . < T > fromFuture ( future , timeout , unit ) ) ;


signature:
public class Single { /** * Converts a { @ link Future } into a { @ code Single } , with a timeout on the Future .
 * < img width = " 640 " height = " 315 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . from . Future . png " alt = " " >
 * You can convert any object that supports the { @ link Future } interface into a { @ code Single } that emits
 * the return value of the { @ link Future # get } method of that object , by passing the object into the
 * { @ code from } method .
 * < em > Important note : < / em > This { @ code Single } is blocking ; you cannot dispose it .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > You specify the { @ link Scheduler } where the blocking wait will happen . < / dd >
 * < / dl >
 * @ 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 Scheduler to use for the blocking wait
 * @ param < T >
 * the type of object that the { @ link Future } returns , and also the type of item to be emitted by
 * the resulting { @ code Single }
 * @ return a { @ code Single } that emits the item from the source { @ link Future }
 * @ see < a href = " http : / / reactivex . io / documentation / operators / from . html " > ReactiveX operators documentation : From < / a > */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . CUSTOM ) public static < T > Single < T > fromFuture ( Future < ? extends T > future , long timeout , TimeUnit unit , Scheduler scheduler ) { } }

implementation: return toSingle ( Flowable . < T > fromFuture ( future , timeout , unit , scheduler ) ) ;


signature:
public class Single { /** * Flattens a { @ code Single } that emits a { @ code Single } into a single { @ code Single } that emits the item
 * emitted by the nested { @ code Single } , without any transformation .
 * < img width = " 640 " height = " 412 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . merge . oo . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code merge } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < dd > The resulting { @ code Single } emits the outer source ' s or the inner { @ code SingleSource } ' s { @ code Throwable } as is .
 * Unlike the other { @ code merge ( ) } operators , this operator won ' t and can ' t produce a { @ code CompositeException } because there is
 * only one possibility for the outer or the inner { @ code SingleSource } to emit an { @ code onError } signal .
 * Therefore , there is no need for a { @ code mergeDelayError ( SingleSource < SingleSource < T > > ) } operator .
 * < / dd >
 * < / dl >
 * @ param < T > the value type of the sources and the output
 * @ param source
 * a { @ code Single } that emits a { @ code Single }
 * @ return a { @ code Single } that emits the item that is the result of flattening the { @ code Single } emitted
 * by { @ code source }
 * @ see < a href = " http : / / reactivex . io / documentation / operators / merge . html " > ReactiveX operators documentation : Merge < / a > */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) @ SuppressWarnings ( { } }

implementation: "unchecked" , "rawtypes" } ) public static < T > Single < T > merge ( SingleSource < ? extends SingleSource < ? extends T > > source ) { ObjectHelper . requireNonNull ( source , "source is null" ) ; return RxJavaPlugins . onAssembly ( new SingleFlatMap < SingleSource < ? extends T > , T > ( source , ( Function ) Functions . identity ( ) ) ) ;


signature:
public class Single { /** * Merges a Flowable sequence of SingleSource instances into a single Flowable sequence ,
 * running all SingleSources at once and delaying any error ( s ) until all sources succeed or fail .
 * < dl >
 * < dt > < b > Backpressure : < / b > < / dt >
 * < dd > The returned { @ code Flowable } honors the backpressure of the downstream consumer . < / dd >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code mergeDelayError } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * < p > History : 2.1.9 - experimental
 * @ param < T > the common and resulting value type
 * @ param sources the Flowable sequence of SingleSource sources
 * @ return the new Flowable instance
 * @ see # merge ( Publisher )
 * @ since 2.2 */
@ CheckReturnValue @ NonNull @ BackpressureSupport ( BackpressureKind . FULL ) @ SchedulerSupport ( SchedulerSupport . NONE ) @ SuppressWarnings ( { } }

implementation: "unchecked" , "rawtypes" } ) public static < T > Flowable < T > mergeDelayError ( Publisher < ? extends SingleSource < ? extends T > > sources ) { ObjectHelper . requireNonNull ( sources , "sources is null" ) ; return RxJavaPlugins . onAssembly ( new FlowableFlatMapPublisher ( sources , SingleInternalHelper . toFlowable ( ) , true , Integer . MAX_VALUE , Flowable . bufferSize ( ) ) ) ;


signature:
public class Single { /** * Returns a singleton instance of a never - signalling Single ( only calls onSubscribe ) .
 * < img width = " 640 " height = " 244 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . never . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code never } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the target value type
 * @ return the singleton never instance
 * @ since 2.0 */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) @ SuppressWarnings ( "unchecked" ) public static < T > Single < T > never ( ) { } }

implementation: return RxJavaPlugins . onAssembly ( ( Single < T > ) SingleNever . INSTANCE ) ;


signature:
public class Single { /** * Signals success with 0L value after the given delay for each SingleObserver .
 * < img width = " 640 " height = " 292 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . timer . s . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > you specify the { @ link Scheduler } to signal on . < / dd >
 * < / dl >
 * @ param delay the delay amount
 * @ param unit the time unit of the delay
 * @ param scheduler the scheduler where the single 0L will be emitted
 * @ return the new Single instance
 * @ throws NullPointerException
 * if unit is null , or
 * if scheduler is null
 * @ since 2.0 */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . CUSTOM ) public static Single < Long > timer ( final long delay , final TimeUnit unit , final Scheduler scheduler ) { } }

implementation: ObjectHelper . requireNonNull ( unit , "unit is null" ) ; ObjectHelper . requireNonNull ( scheduler , "scheduler is null" ) ; return RxJavaPlugins . onAssembly ( new SingleTimer ( delay , unit , scheduler ) ) ;


signature:
public class Single { /** * < strong > Advanced use only : < / strong > creates a Single instance without
 * any safeguards by using a callback that is called with a SingleObserver .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code unsafeCreate } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type
 * @ param onSubscribe the function that is called with the subscribing SingleObserver
 * @ return the new Single instance
 * @ throws IllegalArgumentException if { @ code source } is a subclass of { @ code Single } ; such
 * instances don ' t need conversion and is possibly a port remnant from 1 . x or one should use { @ link # hide ( ) }
 * instead .
 * @ since 2.0 */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static < T > Single < T > unsafeCreate ( SingleSource < T > onSubscribe ) { } }

implementation: ObjectHelper . requireNonNull ( onSubscribe , "onSubscribe is null" ) ; if ( onSubscribe instanceof Single ) { throw new IllegalArgumentException ( "unsafeCreate(Single) should be upgraded" ) ; } return RxJavaPlugins . onAssembly ( new SingleFromUnsafeSource < T > ( onSubscribe ) ) ;


signature:
public class Single { /** * Allows using and disposing a resource while running a SingleSource instance generated from
 * that resource ( similar to a try - with - resources ) .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code using } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type of the SingleSource generated
 * @ param < U > the resource type
 * @ param resourceSupplier the Callable called for each SingleObserver to generate a resource Object
 * @ param singleFunction the function called with the returned resource
 * Object from { @ code resourceSupplier } and should return a SingleSource instance
 * to be run by the operator
 * @ param disposer the consumer of the generated resource that is called exactly once for
 * that particular resource when the generated SingleSource terminates
 * ( successfully or with an error ) or gets disposed .
 * @ return the new Single instance
 * @ since 2.0 */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public static < T , U > Single < T > using ( Callable < U > resourceSupplier , Function < ? super U , ? extends SingleSource < ? extends T > > singleFunction , Consumer < ? super U > disposer ) { } }

implementation: return using ( resourceSupplier , singleFunction , disposer , true ) ;


signature:
public class Single { /** * Calls the specified converter function during assembly time and returns its resulting value .
 * < img width = " 640 " height = " 553 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . as . png " alt = " " >
 * This allows fluent conversion to any other type .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code as } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * < p > History : 2.1.7 - experimental
 * @ param < R > the resulting object type
 * @ param converter the function that receives the current Single instance and returns a value
 * @ return the converted value
 * @ throws NullPointerException if converter is null
 * @ since 2.2 */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final < R > R as ( @ NonNull SingleConverter < T , ? extends R > converter ) { } }

implementation: return ObjectHelper . requireNonNull ( converter , "converter is null" ) . apply ( this ) ;


signature:
public class Single { /** * Hides the identity of the current Single , including the Disposable that is sent
 * to the downstream via { @ code onSubscribe ( ) } .
 * < img width = " 640 " height = " 458 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . hide . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code hide } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ return the new Single instance
 * @ since 2.0 */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Single < T > hide ( ) { } }

implementation: return RxJavaPlugins . onAssembly ( new SingleHide < T > ( this ) ) ;


signature:
public class Single { /** * Transform a Single by applying a particular Transformer function to it .
 * < img width = " 640 " height = " 612 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . compose . png " alt = " " >
 * This method operates on the Single itself whereas { @ link # lift } operates on the Single ' s SingleObservers .
 * If the operator you are creating is designed to act on the individual item emitted by a Single , use
 * { @ link # lift } . If your operator is designed to transform the source Single as a whole ( for instance , by
 * applying a particular set of existing RxJava operators to it ) use { @ code compose } .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code compose } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < R > the value type of the single returned by the transformer function
 * @ param transformer the transformer function , not null
 * @ return the source Single , transformed by the transformer function
 * @ see < a href = " https : / / github . com / ReactiveX / RxJava / wiki / Implementing - Your - Own - Operators " > RxJava wiki : Implementing Your Own Operators < / a > */
@ SuppressWarnings ( "unchecked" ) @ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final < R > Single < R > compose ( SingleTransformer < ? super T , ? extends R > transformer ) { } }

implementation: return wrap ( ( ( SingleTransformer < T , R > ) ObjectHelper . requireNonNull ( transformer , "transformer is null" ) ) . apply ( this ) ) ;


signature:
public class Single { /** * Stores the success value or exception from the current Single and replays it to late SingleObservers .
 * The returned Single subscribes to the current Single when the first SingleObserver subscribes .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code cache } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ return the new Single instance
 * @ since 2.0 */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Single < T > cache ( ) { } }

implementation: return RxJavaPlugins . onAssembly ( new SingleCache < T > ( this ) ) ;


signature:
public class Single { /** * Returns a Flowable that emits the item emitted by the source Single , then the item emitted by the
 * specified Single .
 * < img width = " 640 " height = " 335 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . concatWith . png " alt = " " >
 * < dl >
 * < dt > < b > Backpressure : < / b > < / dt >
 * < dd > The returned { @ code Flowable } honors the backpressure of the downstream consumer . < / dd >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code concatWith } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param other
 * a Single to be concatenated after the current
 * @ return a Flowable that emits the item emitted by the source Single , followed by the item emitted by
 * { @ code t1}
 * @ see < a href = " http : / / reactivex . io / documentation / operators / concat . html " > ReactiveX operators documentation : Concat < / a > */
@ BackpressureSupport ( BackpressureKind . FULL ) @ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Flowable < T > concatWith ( SingleSource < ? extends T > other ) { } }

implementation: return concat ( this , other ) ;


signature:
public class Single { /** * Delays the emission of the success signal from the current Single by the specified amount .
 * An error signal will not be delayed .
 * < img width = " 640 " height = " 457 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . delay . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code delay } operates by default on the { @ code computation } { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param time the amount of time the success signal should be delayed for
 * @ param unit the time unit
 * @ return the new Single instance
 * @ since 2.0 */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . COMPUTATION ) public final Single < T > delay ( long time , TimeUnit unit ) { } }

implementation: return delay ( time , unit , Schedulers . computation ( ) , false ) ;


signature:
public class Single { /** * Delays the emission of the success signal from the current Single by the specified amount .
 * An error signal will not be delayed .
 * < img width = " 640 " height = " 457 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . delay . s . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > you specify the { @ link Scheduler } where the non - blocking wait and emission happens < / dd >
 * < / dl >
 * @ param time the amount of time the success signal should be delayed for
 * @ param unit the time unit
 * @ param scheduler the target scheduler to use for the non - blocking wait and emission
 * @ return the new Single instance
 * @ throws NullPointerException
 * if unit is null , or
 * if scheduler is null
 * @ since 2.0 */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . CUSTOM ) public final Single < T > delay ( final long time , final TimeUnit unit , final Scheduler scheduler ) { } }

implementation: return delay ( time , unit , scheduler , false ) ;


signature:
public class Single { /** * Delays the emission of the success or error signal from the current Single by the specified amount .
 * < img width = " 640 " height = " 457 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . delay . se . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > you specify the { @ link Scheduler } where the non - blocking wait and emission happens < / dd >
 * < / dl >
 * < p > History : 2.1.5 - experimental
 * @ param time the amount of time the success or error signal should be delayed for
 * @ param unit the time unit
 * @ param scheduler the target scheduler to use for the non - blocking wait and emission
 * @ param delayError if true , both success and error signals are delayed . if false , only success signals are delayed .
 * @ return the new Single instance
 * @ throws NullPointerException
 * if unit is null , or
 * if scheduler is null
 * @ since 2.2 */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . CUSTOM ) public final Single < T > delay ( final long time , final TimeUnit unit , final Scheduler scheduler , boolean delayError ) { } }

implementation: ObjectHelper . requireNonNull ( unit , "unit is null" ) ; ObjectHelper . requireNonNull ( scheduler , "scheduler is null" ) ; return RxJavaPlugins . onAssembly ( new SingleDelay < T > ( this , time , unit , scheduler , delayError ) ) ;


signature:
public class Single { /** * Delays the actual subscription to the current Single until the given other CompletableSource
 * completes .
 * < p > If the delaying source signals an error , that error is re - emitted and no subscription
 * to the current Single happens .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code delaySubscription } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param other the CompletableSource that has to complete before the subscription to the
 * current Single happens
 * @ return the new Single instance
 * @ since 2.0 */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final Single < T > delaySubscription ( CompletableSource other ) { } }

implementation: ObjectHelper . requireNonNull ( other , "other is null" ) ; return RxJavaPlugins . onAssembly ( new SingleDelayWithCompletable < T > ( this , other ) ) ;


signature:
public class Single { /** * Delays the actual subscription to the current Single until the given time delay elapsed .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code delaySubscription } does by default subscribe to the current Single
 * on the { @ code computation } { @ link Scheduler } after the delay . < / dd >
 * < / dl >
 * @ param time the time amount to wait with the subscription
 * @ param unit the time unit of the waiting
 * @ return the new Single instance
 * @ since 2.0 */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . COMPUTATION ) public final Single < T > delaySubscription ( long time , TimeUnit unit ) { } }

implementation: return delaySubscription ( time , unit , Schedulers . computation ( ) ) ;


signature:
public class Single { /** * Delays the actual subscription to the current Single until the given time delay elapsed .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code delaySubscription } does by default subscribe to the current Single
 * on the { @ link Scheduler } you provided , after the delay . < / dd >
 * < / dl >
 * @ param time the time amount to wait with the subscription
 * @ param unit the time unit of the waiting
 * @ param scheduler the scheduler to wait on and subscribe on to the current Single
 * @ return the new Single instance
 * @ since 2.0 */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . CUSTOM ) public final Single < T > delaySubscription ( long time , TimeUnit unit , Scheduler scheduler ) { } }

implementation: return delaySubscription ( Observable . timer ( time , unit , scheduler ) ) ;


signature:
public class Single { /** * Maps the { @ link Notification } success value of this Single back into normal
 * { @ code onSuccess } , { @ code onError } or { @ code onComplete } signals as a
 * { @ link Maybe } source .
 * The intended use of the { @ code selector } function is to perform a
 * type - safe identity mapping ( see example ) on a source that is already of type
 * { @ code Notification < T > } . The Java language doesn ' t allow
 * limiting instance methods to a certain generic argument shape , therefore ,
 * a function is used to ensure the conversion remains type safe .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code dematerialize } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * Example :
 * < pre > < code >
 * Single . just ( Notification . createOnNext ( 1 ) )
 * . dematerialize ( notification - & gt ; notification )
 * . test ( )
 * . assertResult ( 1 ) ;
 * < / code > < / pre >
 * @ param < R > the result type
 * @ param selector the function called with the success item and should
 * return a { @ link Notification } instance .
 * @ return the new Maybe instance
 * @ since 2.2.4 - experimental
 * @ see # materialize ( ) */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) @ Experimental public final < R > Maybe < R > dematerialize ( Function < ? super T , Notification < R > > selector ) { } }

implementation: ObjectHelper . requireNonNull ( selector , "selector is null" ) ; return RxJavaPlugins . onAssembly ( new SingleDematerialize < T , R > ( this , selector ) ) ;


signature:
public class Single { /** * Calls the specified consumer with the success item after this item has been emitted to the downstream .
 * < img width = " 640 " height = " 460 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . doAfterSuccess . png " alt = " " >
 * Note that the { @ code doAfterSuccess } action is shared between subscriptions and as such
 * should be thread - safe .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code doAfterSuccess } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * < p > History : 2.0.1 - experimental
 * @ param onAfterSuccess the Consumer that will be called after emitting an item from upstream to the downstream
 * @ return the new Single instance
 * @ since 2.1 */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final Single < T > doAfterSuccess ( Consumer < ? super T > onAfterSuccess ) { } }

implementation: ObjectHelper . requireNonNull ( onAfterSuccess , "onAfterSuccess is null" ) ; return RxJavaPlugins . onAssembly ( new SingleDoAfterSuccess < T > ( this , onAfterSuccess ) ) ;


signature:
public class Single { /** * Registers an { @ link Action } to be called after this Single invokes either onSuccess or onError .
 * < img width = " 640 " height = " 460 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . doAfterTerminate . png " alt = " " >
 * Note that the { @ code doAfterTerminate } action is shared between subscriptions and as such
 * should be thread - safe . < / p >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code doAfterTerminate } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * < p > History : 2.0.6 - experimental
 * @ param onAfterTerminate
 * an { @ link Action } to be invoked when the source Single finishes
 * @ return a Single that emits the same items as the source Single , then invokes the
 * { @ link Action }
 * @ see < a href = " http : / / reactivex . io / documentation / operators / do . html " > ReactiveX operators documentation : Do < / a >
 * @ since 2.1 */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final Single < T > doAfterTerminate ( Action onAfterTerminate ) { } }

implementation: ObjectHelper . requireNonNull ( onAfterTerminate , "onAfterTerminate is null" ) ; return RxJavaPlugins . onAssembly ( new SingleDoAfterTerminate < T > ( this , onAfterTerminate ) ) ;


signature:
public class Single { /** * Calls the specified action after this Single signals onSuccess or onError or gets disposed by
 * the downstream .
 * < p > In case of a race between a terminal event and a dispose call , the provided { @ code onFinally } action
 * is executed once per subscription .
 * < p > Note that the { @ code onFinally } action is shared between subscriptions and as such
 * should be thread - safe .
 * < img width = " 640 " height = " 291 " src = " https : / / raw . githubusercontent . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . doFinally . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code doFinally } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * < p > History : 2.0.1 - experimental
 * @ param onFinally the action called when this Single terminates or gets disposed
 * @ return the new Single instance
 * @ since 2.1 */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final Single < T > doFinally ( Action onFinally ) { } }

implementation: ObjectHelper . requireNonNull ( onFinally , "onFinally is null" ) ; return RxJavaPlugins . onAssembly ( new SingleDoFinally < T > ( this , onFinally ) ) ;


signature:
public class Single { /** * Calls the shared consumer with the Disposable sent through the onSubscribe for each
 * SingleObserver that subscribes to the current Single .
 * < img width = " 640 " height = " 347 " src = " https : / / raw . githubusercontent . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . doOnSubscribe . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code doOnSubscribe } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param onSubscribe the consumer called with the Disposable sent via onSubscribe
 * @ return the new Single instance
 * @ since 2.0 */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final Single < T > doOnSubscribe ( final Consumer < ? super Disposable > onSubscribe ) { } }

implementation: ObjectHelper . requireNonNull ( onSubscribe , "onSubscribe is null" ) ; return RxJavaPlugins . onAssembly ( new SingleDoOnSubscribe < T > ( this , onSubscribe ) ) ;


signature:
public class Single { /** * Returns a Single instance that calls the given onTerminate callback
 * just before this Single completes normally or with an exception .
 * < img width = " 640 " height = " 305 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / doOnTerminate . png " alt = " " >
 * This differs from { @ code doAfterTerminate } in that this happens < em > before < / em > the { @ code onComplete } or
 * { @ code onError } notification .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code doOnTerminate } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param onTerminate the action to invoke when the consumer calls { @ code onComplete } or { @ code onError }
 * @ return the new Single instance
 * @ see < a href = " http : / / reactivex . io / documentation / operators / do . html " > ReactiveX operators documentation : Do < / a >
 * @ see # doOnTerminate ( Action )
 * @ since 2.2.7 - experimental */
@ Experimental @ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final Single < T > doOnTerminate ( final Action onTerminate ) { } }

implementation: ObjectHelper . requireNonNull ( onTerminate , "onTerminate is null" ) ; return RxJavaPlugins . onAssembly ( new SingleDoOnTerminate < T > ( this , onTerminate ) ) ;


signature:
public class Single { /** * Calls the shared consumer with the success value sent via onSuccess for each
 * SingleObserver that subscribes to the current Single .
 * < img width = " 640 " height = " 347 " src = " https : / / raw . githubusercontent . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . doOnSuccess . 2 . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code doOnSuccess } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param onSuccess the consumer called with the success value of onSuccess
 * @ return the new Single instance
 * @ since 2.0 */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final Single < T > doOnSuccess ( final Consumer < ? super T > onSuccess ) { } }

implementation: ObjectHelper . requireNonNull ( onSuccess , "onSuccess is null" ) ; return RxJavaPlugins . onAssembly ( new SingleDoOnSuccess < T > ( this , onSuccess ) ) ;


signature:
public class Single { /** * Calls the shared consumer with the error sent via onError for each
 * SingleObserver that subscribes to the current Single .
 * < img width = " 640 " height = " 349 " src = " https : / / raw . githubusercontent . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . doOnError . 2 . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code doOnError } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param onError the consumer called with the success value of onError
 * @ return the new Single instance
 * @ since 2.0 */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final Single < T > doOnError ( final Consumer < ? super Throwable > onError ) { } }

implementation: ObjectHelper . requireNonNull ( onError , "onError is null" ) ; return RxJavaPlugins . onAssembly ( new SingleDoOnError < T > ( this , onError ) ) ;


signature:
public class Single { /** * Calls the shared { @ code Action } if a SingleObserver subscribed to the current Single
 * disposes the common Disposable it received via onSubscribe .
 * < img width = " 640 " height = " 332 " src = " https : / / raw . githubusercontent . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . doOnDispose . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code doOnDispose } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param onDispose the action called when the subscription is disposed
 * @ return the new Single instance
 * @ throws NullPointerException if onDispose is null
 * @ since 2.0 */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final Single < T > doOnDispose ( final Action onDispose ) { } }

implementation: ObjectHelper . requireNonNull ( onDispose , "onDispose is null" ) ; return RxJavaPlugins . onAssembly ( new SingleDoOnDispose < T > ( this , onDispose ) ) ;


signature:
public class Single { /** * Returns a Single that is based on applying a specified function to the item emitted by the source Single ,
 * where that function returns a SingleSource .
 * < img width = " 640 " height = " 300 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . flatMap . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code flatMap } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < R > the result value type
 * @ param mapper
 * a function that , when applied to the item emitted by the source Single , returns a SingleSource
 * @ return the Single returned from { @ code mapper } when applied to the item emitted by the source Single
 * @ see < a href = " http : / / reactivex . io / documentation / operators / flatmap . html " > ReactiveX operators documentation : FlatMap < / a > */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final < R > Single < R > flatMap ( Function < ? super T , ? extends SingleSource < ? extends R > > mapper ) { } }

implementation: ObjectHelper . requireNonNull ( mapper , "mapper is null" ) ; return RxJavaPlugins . onAssembly ( new SingleFlatMap < T , R > ( this , mapper ) ) ;


signature:
public class Single { /** * 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 .
 * < img width = " 640 " height = " 191 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . flatMapMaybe . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code flatMapMaybe } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < R > the result value type
 * @ param mapper
 * a function that , when applied to the item emitted by the source Single , returns a MaybeSource
 * @ return the Maybe returned from { @ code mapper } when applied to the item emitted by the source Single
 * @ see < a href = " http : / / reactivex . io / documentation / operators / flatmap . html " > ReactiveX operators documentation : FlatMap < / a > */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final < R > Maybe < R > flatMapMaybe ( final Function < ? super T , ? extends MaybeSource < ? extends R > > mapper ) { } }

implementation: ObjectHelper . requireNonNull ( mapper , "mapper is null" ) ; return RxJavaPlugins . onAssembly ( new SingleFlatMapMaybe < T , R > ( this , mapper ) ) ;


signature:
public class Single { /** * Returns an Observable that is based on applying a specified function to the item emitted by the source Single ,
 * where that function returns an ObservableSource .
 * < img width = " 640 " height = " 300 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Single . flatMapObservable . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code flatMapObservable } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < R > the result value type
 * @ param mapper
 * a function that , when applied to the item emitted by the source Single , returns an ObservableSource
 * @ return the Observable returned from { @ code func } when applied to the item emitted by the source Single
 * @ see < a href = " http : / / reactivex . io / documentation / operators / flatmap . html " > ReactiveX operators documentation : FlatMap < / a > */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final < R > Observable < R > flatMapObservable ( Function < ? super T , ? extends ObservableSource < ? extends R > > mapper ) { } }

implementation: ObjectHelper . requireNonNull ( mapper , "mapper is null" ) ; return RxJavaPlugins . onAssembly ( new SingleFlatMapObservable < T , R > ( this , mapper ) ) ;