signature:
public class BehaviorSubject { /** * Creates a { @ link BehaviorSubject } that emits the last item it observed and all subsequent items to each
 * { @ link Observer } that subscribes to it .
 * @ param < T >
 * the type of item the Subject will emit
 * @ param defaultValue
 * the item that will be emitted first to any { @ link Observer } as long as the
 * { @ link BehaviorSubject } has not yet observed any items from its source { @ code Observable }
 * @ return the constructed { @ link BehaviorSubject } */
@ CheckReturnValue @ NonNull public static < T > BehaviorSubject < T > createDefault ( T defaultValue ) { } }

implementation: return new BehaviorSubject < T > ( defaultValue ) ;


signature:
public class BehaviorSubject { /** * Returns a single value the Subject currently has or null if no such value exists .
 * < p > The method is thread - safe .
 * @ return a single value the Subject currently has or null if no such value exists */
@ Nullable public T getValue ( ) { } }

implementation: Object o = value . get ( ) ; if ( NotificationLite . isComplete ( o ) || NotificationLite . isError ( o ) ) { return null ; } return NotificationLite . getValue ( o ) ;


signature:
public class BehaviorSubject { /** * Returns an Object array containing snapshot all values of the Subject .
 * < p > The method is thread - safe .
 * @ return the array containing the snapshot of all values of the Subject
 * @ deprecated in 2.1.14 ; put the result of { @ link # getValue ( ) } into an array manually , will be removed in 3 . x */
@ Deprecated public Object [ ] getValues ( ) { } }

implementation: @ SuppressWarnings ( "unchecked" ) T [ ] a = ( T [ ] ) EMPTY_ARRAY ; T [ ] b = getValues ( a ) ; if ( b == EMPTY_ARRAY ) { return new Object [ 0 ] ; } return b ;


signature:
public class BehaviorSubject { /** * Returns a typed array containing a snapshot of all values of the Subject .
 * < p > The method follows the conventions of Collection . toArray by setting the array element
 * after the last value to null ( if the capacity permits ) .
 * < p > The method is thread - safe .
 * @ param array the target array to copy values into if it fits
 * @ return the given array if the values fit into it or a new array containing all values
 * @ deprecated in 2.1.14 ; put the result of { @ link # getValue ( ) } into an array manually , will be removed in 3 . x */
@ Deprecated @ SuppressWarnings ( "unchecked" ) public T [ ] getValues ( T [ ] array ) { } }

implementation: Object o = value . get ( ) ; if ( o == null || NotificationLite . isComplete ( o ) || NotificationLite . isError ( o ) ) { if ( array . length != 0 ) { array [ 0 ] = null ; } return array ; } T v = NotificationLite . getValue ( o ) ; if ( array . length != 0 ) { array [ 0 ] = v ; if ( array . length != 1 ) { array [ 1 ] = null ; } } else { array = ( T [ ] ) Array . newInstance ( array . getClass ( ) . getComponentType ( ) , 1 ) ; array [ 0 ] = v ; } return array ;


signature:
public class BehaviorSubject { /** * Returns true if the subject has any value .
 * < p > The method is thread - safe .
 * @ return true if the subject has any value */
public boolean hasValue ( ) { } }

implementation: Object o = value . get ( ) ; return o != null && ! NotificationLite . isComplete ( o ) && ! NotificationLite . isError ( o ) ;


signature:
public class QueueDrainHelper { /** * Drain the queue but give up with an error if there aren ' t enough requests .
 * @ param < T > the queue value type
 * @ param < U > the emission value type
 * @ param q the queue
 * @ param a the subscriber
 * @ param delayError true if errors should be delayed after all normal items
 * @ param dispose the disposable to call when termination happens and cleanup is necessary
 * @ param qd the QueueDrain instance that gives status information to the drain logic */
public static < T , U > void drainMaxLoop ( SimplePlainQueue < T > q , Subscriber < ? super U > a , boolean delayError , Disposable dispose , QueueDrain < T , U > qd ) { } }

implementation: int missed = 1 ; for ( ; ; ) { for ( ; ; ) { boolean d = qd . done ( ) ; T v = q . poll ( ) ; boolean empty = v == null ; if ( checkTerminated ( d , empty , a , delayError , q , qd ) ) { if ( dispose != null ) { dispose . dispose ( ) ; } return ; } if ( empty ) { break ; } long r = qd . requested ( ) ; if ( r != 0L ) { if ( qd . accept ( a , v ) ) { if ( r != Long . MAX_VALUE ) { qd . produced ( 1 ) ; } } } else { q . clear ( ) ; if ( dispose != null ) { dispose . dispose ( ) ; } a . onError ( new MissingBackpressureException ( "Could not emit value due to lack of requests." ) ) ; return ; } } missed = qd . leave ( - missed ) ; if ( missed == 0 ) { break ; } }


signature:
public class QueueDrainHelper { /** * Creates a queue : spsc - array if capacityHint is positive and
 * spsc - linked - array if capacityHint is negative ; in both cases , the
 * capacity is the absolute value of prefetch .
 * @ param < T > the value type of the queue
 * @ param capacityHint the capacity hint , negative value will create an array - based SPSC queue
 * @ return the queue instance */
public static < T > SimpleQueue < T > createQueue ( int capacityHint ) { } }

implementation: if ( capacityHint < 0 ) { return new SpscLinkedArrayQueue < T > ( - capacityHint ) ; } return new SpscArrayQueue < T > ( capacityHint ) ;


signature:
public class QueueDrainHelper { /** * Requests Long . MAX _ VALUE if prefetch is negative or the exact
 * amount if prefetch is positive .
 * @ param s the Subscription to request from
 * @ param prefetch the prefetch value */
public static void request ( Subscription s , int prefetch ) { } }

implementation: s . request ( prefetch < 0 ? Long . MAX_VALUE : prefetch ) ;


signature:
public class QueueDrainHelper { /** * Accumulates requests ( not validated ) and handles the completed mode draining of the queue based on the requests .
 * Post - completion backpressure handles the case when a source produces values based on
 * requests when it is active but more values are available even after its completion .
 * In this case , the onComplete ( ) can ' t just emit the contents of the queue but has to
 * coordinate with the requested amounts . This requires two distinct modes : active and
 * completed . In active mode , requests flow through and the queue is not accessed but
 * in completed mode , requests no - longer reach the upstream but help in draining the queue .
 * @ param < T > the value type emitted
 * @ param n the request amount , positive ( not validated )
 * @ param actual the target Subscriber to send events to
 * @ param queue the queue to drain if in the post - complete state
 * @ param state holds the request amount and the post - completed flag
 * @ param isCancelled a supplier that returns true if the drain has been cancelled
 * @ return true if the state indicates a completion state . */
public static < T > boolean postCompleteRequest ( long n , Subscriber < ? super T > actual , Queue < T > queue , AtomicLong state , BooleanSupplier isCancelled ) { } }

implementation:
for ( ; ; ) { long r = state . get ( ) ; // extract the current request amount
long r0 = r & REQUESTED_MASK ; // preserve COMPLETED _ MASK and calculate new requested amount
long u = ( r & COMPLETED_MASK ) | BackpressureHelper . addCap ( r0 , n ) ; if ( state . compareAndSet ( r , u ) ) { // ( complete , 0 ) - > ( complete , n ) transition then replay
if ( r == COMPLETED_MASK ) { postCompleteDrain ( n | COMPLETED_MASK , actual , queue , state , isCancelled ) ; return true ; } // ( active , r ) - > ( active , r + n ) transition then continue with requesting from upstream
return false ; } }


signature:
public class QueueDrainHelper { /** * Drains the queue based on the outstanding requests in post - completed mode ( only ! ) .
 * @ param n the current request amount
 * @ param actual the target Subscriber to send events to
 * @ param queue the queue to drain if in the post - complete state
 * @ param state holds the request amount and the post - completed flag
 * @ param isCancelled a supplier that returns true if the drain has been cancelled
 * @ return true if the queue was completely drained or the drain process was cancelled */
static < T > boolean postCompleteDrain ( long n , Subscriber < ? super T > actual , Queue < T > queue , AtomicLong state , BooleanSupplier isCancelled ) { } }

implementation:
// TODO enable fast - path
// if ( n = = - 1 | | n = = Long . MAX _ VALUE ) {
// for ( ; ; ) {
// if ( isCancelled . getAsBoolean ( ) ) {
// break ;
// T v = queue . poll ( ) ;
// if ( v = = null ) {
// actual . onComplete ( ) ;
// break ;
// actual . onNext ( v ) ;
// return true ;
long e = n & COMPLETED_MASK ; for ( ; ; ) { while ( e != n ) { if ( isCancelled ( isCancelled ) ) { return true ; } T t = queue . poll ( ) ; if ( t == null ) { actual . onComplete ( ) ; return true ; } actual . onNext ( t ) ; e ++ ; } if ( isCancelled ( isCancelled ) ) { return true ; } if ( queue . isEmpty ( ) ) { actual . onComplete ( ) ; return true ; } n = state . get ( ) ; if ( n == e ) { n = state . addAndGet ( - ( e & REQUESTED_MASK ) ) ; if ( ( n & REQUESTED_MASK ) == 0L ) { return false ; } e = n & COMPLETED_MASK ; } }


signature:
public class QueueDrainHelper { /** * Signals the completion of the main sequence and switches to post - completion replay mode .
 * Don ' t modify the queue after calling this method !
 * Post - completion backpressure handles the case when a source produces values based on
 * requests when it is active but more values are available even after its completion .
 * In this case , the onComplete ( ) can ' t just emit the contents of the queue but has to
 * coordinate with the requested amounts . This requires two distinct modes : active and
 * completed . In active mode , requests flow through and the queue is not accessed but
 * in completed mode , requests no - longer reach the upstream but help in draining the queue .
 * The algorithm utilizes the most significant bit ( bit 63 ) of a long value ( AtomicLong ) since
 * request amount only goes up to Long . MAX _ VALUE ( bits 0-62 ) and negative values aren ' t
 * allowed .
 * @ param < T > the value type emitted
 * @ param actual the target Subscriber to send events to
 * @ param queue the queue to drain if in the post - complete state
 * @ param state holds the request amount and the post - completed flag
 * @ param isCancelled a supplier that returns true if the drain has been cancelled */
public static < T > void postComplete ( Subscriber < ? super T > actual , Queue < T > queue , AtomicLong state , BooleanSupplier isCancelled ) { } }

implementation:
if ( queue . isEmpty ( ) ) { actual . onComplete ( ) ; return ; } if ( postCompleteDrain ( state . get ( ) , actual , queue , state , isCancelled ) ) { return ; } for ( ; ; ) { long r = state . get ( ) ; if ( ( r & COMPLETED_MASK ) != 0L ) { return ; } long u = r | COMPLETED_MASK ; // ( active , r ) - > ( complete , r ) transition
if ( state . compareAndSet ( r , u ) ) { // if the requested amount was non - zero , drain the queue
if ( r != 0L ) { postCompleteDrain ( u , actual , queue , state , isCancelled ) ; } return ; } }


signature:
public class ReplaySubject { /** * Creates an unbounded replay subject .
 * The internal buffer is backed by an { @ link ArrayList } and starts with an initial capacity of 16 . Once the
 * number of items reaches this capacity , it will grow as necessary ( usually by 50 % ) . However , as the
 * number of items grows , this causes frequent array reallocation and copying , and may hurt performance
 * and latency . This can be avoided with the { @ link # create ( int ) } overload which takes an initial capacity
 * parameter and can be tuned to reduce the array reallocation frequency as needed .
 * @ param < T >
 * the type of items observed and emitted by the Subject
 * @ return the created subject */
@ CheckReturnValue @ NonNull public static < T > ReplaySubject < T > create ( ) { } }

implementation: return new ReplaySubject < T > ( new UnboundedReplayBuffer < T > ( 16 ) ) ;


signature:
public class ReplaySubject { /** * Creates a size - bounded replay subject .
 * In this setting , the { @ code ReplaySubject } holds at most { @ code size } items in its internal buffer and
 * discards the oldest item .
 * When observers subscribe to a terminated { @ code ReplaySubject } , they are guaranteed to see at most
 * { @ code size } { @ code onNext } events followed by a termination event .
 * If an observer subscribes while the { @ code ReplaySubject } is active , it will observe all items in the
 * buffer at that point in time and each item observed afterwards , even if the buffer evicts items due to
 * the size constraint in the mean time . In other words , once an Observer subscribes , it will receive items
 * without gaps in the sequence .
 * @ param < T >
 * the type of items observed and emitted by the Subject
 * @ param maxSize
 * the maximum number of buffered items
 * @ return the created subject */
@ CheckReturnValue @ NonNull public static < T > ReplaySubject < T > createWithSize ( int maxSize ) { } }

implementation: return new ReplaySubject < T > ( new SizeBoundReplayBuffer < T > ( maxSize ) ) ;


signature:
public class ReplaySubject { /** * / * test */
static < T > ReplaySubject < T > createUnbounded ( ) { } }

implementation: return new ReplaySubject < T > ( new SizeBoundReplayBuffer < T > ( Integer . MAX_VALUE ) ) ;


signature:
public class SinglePostCompleteSubscriber { /** * Signals the given value and an onComplete if the downstream is ready to receive the final value .
 * @ param n the value to emit */
protected final void complete ( R n ) { } }

implementation: long p = produced ; if ( p != 0 ) { BackpressureHelper . produced ( this , p ) ; } for ( ; ; ) { long r = get ( ) ; if ( ( r & COMPLETE_MASK ) != 0 ) { onDrop ( n ) ; return ; } if ( ( r & REQUEST_MASK ) != 0 ) { lazySet ( COMPLETE_MASK + 1 ) ; downstream . onNext ( n ) ; downstream . onComplete ( ) ; return ; } value = n ; if ( compareAndSet ( 0 , COMPLETE_MASK ) ) { return ; } value = null ; }


signature:
public class AsyncProcessor { /** * Returns an Object array containing snapshot all values of this processor .
 * < p > The method is thread - safe .
 * @ return the array containing the snapshot of all values of this processor
 * @ deprecated in 2.1.14 ; put the result of { @ link # getValue ( ) } into an array manually , will be removed in 3 . x */
@ Deprecated public Object [ ] getValues ( ) { } }

implementation: T v = getValue ( ) ; return v != null ? new Object [ ] { v } : new Object [ 0 ] ;


signature:
public class AsyncProcessor { /** * Returns a typed array containing a snapshot of all values of this processor .
 * < p > The method follows the conventions of Collection . toArray by setting the array element
 * after the last value to null ( if the capacity permits ) .
 * < p > The method is thread - safe .
 * @ param array the target array to copy values into if it fits
 * @ return the given array if the values fit into it or a new array containing all values
 * @ deprecated in 2.1.14 ; put the result of { @ link # getValue ( ) } into an array manually , will be removed in 3 . x */
@ Deprecated public T [ ] getValues ( T [ ] array ) { } }

implementation: T v = getValue ( ) ; if ( v == null ) { if ( array . length != 0 ) { array [ 0 ] = null ; } return array ; } if ( array . length == 0 ) { array = Arrays . copyOf ( array , 1 ) ; } array [ 0 ] = v ; if ( array . length != 1 ) { array [ 1 ] = null ; } return array ;


signature:
public class Notification { /** * Returns the contained value if this notification is an onNext
 * signal , null otherwise .
 * @ return the value contained or null
 * @ see # isOnNext ( ) */
@ SuppressWarnings ( "unchecked" ) @ Nullable public T getValue ( ) { } }

implementation: Object o = value ; if ( o != null && ! NotificationLite . isError ( o ) ) { return ( T ) value ; } return null ;


signature:
public class Notification { /** * Returns the container Throwable error if this notification is an onError
 * signal , null otherwise .
 * @ return the Throwable error contained or null
 * @ see # isOnError ( ) */
@ Nullable public Throwable getError ( ) { } }

implementation: Object o = value ; if ( NotificationLite . isError ( o ) ) { return NotificationLite . getError ( o ) ; } return null ;


signature:
public class Notification { /** * Constructs an onNext notification containing the given value .
 * @ param < T > the value type
 * @ param value the value to carry around in the notification , not null
 * @ return the new Notification instance
 * @ throws NullPointerException if value is null */
@ NonNull public static < T > Notification < T > createOnNext ( @ NonNull T value ) { } }

implementation: ObjectHelper . requireNonNull ( value , "value is null" ) ; return new Notification < T > ( value ) ;


signature:
public class Notification { /** * Constructs an onError notification containing the error .
 * @ param < T > the value type
 * @ param error the error Throwable to carry around in the notification , not null
 * @ return the new Notification instance
 * @ throws NullPointerException if error is null */
@ NonNull public static < T > Notification < T > createOnError ( @ NonNull Throwable error ) { } }

implementation: ObjectHelper . requireNonNull ( error , "error is null" ) ; return new Notification < T > ( NotificationLite . error ( error ) ) ;


signature:
public class EndConsumerHelper { /** * Ensures that the upstream Disposable is null and returns true , otherwise
 * disposes the next Disposable and if the upstream is not the shared
 * disposed instance , reports a ProtocolViolationException due to
 * multiple subscribe attempts .
 * @ param upstream the upstream current value
 * @ param next the Disposable to check for nullness and dispose if necessary
 * @ param observer the class of the consumer to have a personalized
 * error message if the upstream already contains a non - cancelled Disposable .
 * @ return true if successful , false if the upstream was non null */
public static boolean validate ( Disposable upstream , Disposable next , Class < ? > observer ) { } }

implementation: ObjectHelper . requireNonNull ( next , "next is null" ) ; if ( upstream != null ) { next . dispose ( ) ; if ( upstream != DisposableHelper . DISPOSED ) { reportDoubleSubscription ( observer ) ; } return false ; } return true ;


signature:
public class EndConsumerHelper { /** * Atomically updates the target upstream AtomicReference from null to the non - null
 * next Disposable , otherwise disposes next and reports a ProtocolViolationException
 * if the AtomicReference doesn ' t contain the shared disposed indicator .
 * @ param upstream the target AtomicReference to update
 * @ param next the Disposable to set on it atomically
 * @ param observer the class of the consumer to have a personalized
 * error message if the upstream already contains a non - cancelled Disposable .
 * @ return true if successful , false if the content of the AtomicReference was non null */
public static boolean setOnce ( AtomicReference < Disposable > upstream , Disposable next , Class < ? > observer ) { } }

implementation: ObjectHelper . requireNonNull ( next , "next is null" ) ; if ( ! upstream . compareAndSet ( null , next ) ) { next . dispose ( ) ; if ( upstream . get ( ) != DisposableHelper . DISPOSED ) { reportDoubleSubscription ( observer ) ; } return false ; } return true ;


signature:
public class EndConsumerHelper { /** * Ensures that the upstream Subscription is null and returns true , otherwise
 * cancels the next Subscription and if the upstream is not the shared
 * cancelled instance , reports a ProtocolViolationException due to
 * multiple subscribe attempts .
 * @ param upstream the upstream current value
 * @ param next the Subscription to check for nullness and cancel if necessary
 * @ param subscriber the class of the consumer to have a personalized
 * error message if the upstream already contains a non - cancelled Subscription .
 * @ return true if successful , false if the upstream was non null */
public static boolean validate ( Subscription upstream , Subscription next , Class < ? > subscriber ) { } }

implementation: ObjectHelper . requireNonNull ( next , "next is null" ) ; if ( upstream != null ) { next . cancel ( ) ; if ( upstream != SubscriptionHelper . CANCELLED ) { reportDoubleSubscription ( subscriber ) ; } return false ; } return true ;


signature:
public class EndConsumerHelper { /** * Atomically updates the target upstream AtomicReference from null to the non - null
 * next Subscription , otherwise cancels next and reports a ProtocolViolationException
 * if the AtomicReference doesn ' t contain the shared cancelled indicator .
 * @ param upstream the target AtomicReference to update
 * @ param next the Subscription to set on it atomically
 * @ param subscriber the class of the consumer to have a personalized
 * error message if the upstream already contains a non - cancelled Subscription .
 * @ return true if successful , false if the content of the AtomicReference was non null */
public static boolean setOnce ( AtomicReference < Subscription > upstream , Subscription next , Class < ? > subscriber ) { } }

implementation: ObjectHelper . requireNonNull ( next , "next is null" ) ; if ( ! upstream . compareAndSet ( null , next ) ) { next . cancel ( ) ; if ( upstream . get ( ) != SubscriptionHelper . CANCELLED ) { reportDoubleSubscription ( subscriber ) ; } return false ; } return true ;


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

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


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

implementation: for ( ; ; ) { CacheSubscription < T > [ ] current = subscribers . 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 ; } CacheSubscription < T > [ ] next ; if ( n == 1 ) { next = EMPTY ; } else { next = new CacheSubscription [ n - 1 ] ; System . arraycopy ( current , 0 , next , 0 , j ) ; System . arraycopy ( current , j + 1 , next , j , n - j - 1 ) ; } if ( subscribers . compareAndSet ( current , next ) ) { return ; } }


signature:
public class FlowableCache { /** * 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 ( CacheSubscription < 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 ; AtomicLong requested = consumer . requested ; Subscriber < ? super T > downstream = consumer . downstream ; int capacity = capacityHint ; for ( ; ; ) { // 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 ) { // see how many items the consumer has requested in total so far
long consumerRequested = requested . get ( ) ; // MIN _ VALUE indicates a cancelled consumer , we stop replaying
if ( consumerRequested == Long . MIN_VALUE ) { // release the node object to avoid leaks through retained consumers
consumer . node = null ; return ; } // if the consumer has requested more and there is more , we will emit an item
if ( consumerRequested != index ) { // 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 BackpressureHelper { /** * Adds two long values and caps the sum at Long . MAX _ VALUE .
 * @ param a the first value
 * @ param b the second value
 * @ return the sum capped at Long . MAX _ VALUE */
public static long addCap ( long a , long b ) { } }

implementation: long u = a + b ; if ( u < 0L ) { return Long . MAX_VALUE ; } return u ;


signature:
public class BackpressureHelper { /** * Multiplies two long values and caps the product at Long . MAX _ VALUE .
 * @ param a the first value
 * @ param b the second value
 * @ return the product capped at Long . MAX _ VALUE */
public static long multiplyCap ( long a , long b ) { } }

implementation: long u = a * b ; if ( ( ( a | b ) >>> 31 ) != 0 ) { if ( u / a != b ) { return Long . MAX_VALUE ; } } return u ;


signature:
public class BackpressureHelper { /** * Atomically adds the positive value n to the requested value in the AtomicLong and
 * caps the result at Long . MAX _ VALUE and returns the previous value .
 * @ param requested the AtomicLong holding the current requested value
 * @ param n the value to add , must be positive ( not verified )
 * @ return the original value before the add */
public static long add ( AtomicLong requested , long n ) { } }

implementation: for ( ; ; ) { long r = requested . get ( ) ; if ( r == Long . MAX_VALUE ) { return Long . MAX_VALUE ; } long u = addCap ( r , n ) ; if ( requested . compareAndSet ( r , u ) ) { return r ; } }


signature:
public class BackpressureHelper { /** * Atomically adds the positive value n to the requested value in the AtomicLong and
 * caps the result at Long . MAX _ VALUE and returns the previous value and
 * considers Long . MIN _ VALUE as a cancel indication ( no addition then ) .
 * @ param requested the AtomicLong holding the current requested value
 * @ param n the value to add , must be positive ( not verified )
 * @ return the original value before the add */
public static long addCancel ( AtomicLong requested , long n ) { } }

implementation: for ( ; ; ) { long r = requested . get ( ) ; if ( r == Long . MIN_VALUE ) { return Long . MIN_VALUE ; } if ( r == Long . MAX_VALUE ) { return Long . MAX_VALUE ; } long u = addCap ( r , n ) ; if ( requested . compareAndSet ( r , u ) ) { return r ; } }


signature:
public class BackpressureHelper { /** * Atomically subtract the given number ( positive , not validated ) from the target field unless it contains Long . MAX _ VALUE .
 * @ param requested the target field holding the current requested amount
 * @ param n the produced element count , positive ( not validated )
 * @ return the new amount */
public static long produced ( AtomicLong requested , long n ) { } }

implementation: for ( ; ; ) { long current = requested . get ( ) ; if ( current == Long . MAX_VALUE ) { return Long . MAX_VALUE ; } long update = current - n ; if ( update < 0L ) { RxJavaPlugins . onError ( new IllegalStateException ( "More produced than requested: " + update ) ) ; update = 0L ; } if ( requested . compareAndSet ( current , update ) ) { return update ; } }


signature:
public class ObservableBlockingSubscribe { /** * Subscribes to the source and calls the Observer methods on the current thread .
 * @ param o the source ObservableSource
 * The call to dispose ( ) is composed through .
 * @ param observer the subscriber to forward events and calls to in the current thread
 * @ param < T > the value type */
public static < T > void subscribe ( ObservableSource < ? extends T > o , Observer < ? super T > observer ) { } }

implementation: final BlockingQueue < Object > queue = new LinkedBlockingQueue < Object > ( ) ; BlockingObserver < T > bs = new BlockingObserver < T > ( queue ) ; observer . onSubscribe ( bs ) ; o . subscribe ( bs ) ; for ( ; ; ) { if ( bs . isDisposed ( ) ) { break ; } Object v = queue . poll ( ) ; if ( v == null ) { try { v = queue . take ( ) ; } catch ( InterruptedException ex ) { bs . dispose ( ) ; observer . onError ( ex ) ; return ; } } if ( bs . isDisposed ( ) || o == BlockingObserver . TERMINATED || NotificationLite . acceptFull ( v , observer ) ) { break ; } }


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

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


signature:
public class ObservableBlockingSubscribe { /** * Subscribes to the source and calls the given actions on the current thread .
 * @ param o the source ObservableSource
 * @ 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 ( ObservableSource < ? 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 LambdaObserver < T > ( onNext , onError , onComplete , Functions . emptyConsumer ( ) ) ) ;


signature:
public class Completable { /** * Returns a Completable which completes only when all sources complete , one after another .
 * < img width = " 640 " height = " 283 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . concatArray . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code concatArray } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param sources the sources to concatenate
 * @ return the Completable instance which completes only when all sources complete
 * @ throws NullPointerException if sources is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static Completable concatArray ( CompletableSource ... sources ) { } }

implementation: ObjectHelper . requireNonNull ( sources , "sources is null" ) ; if ( sources . length == 0 ) { return complete ( ) ; } else if ( sources . length == 1 ) { return wrap ( sources [ 0 ] ) ; } return RxJavaPlugins . onAssembly ( new CompletableConcatArray ( sources ) ) ;


signature:
public class Completable { /** * Returns a Completable which completes only when all sources complete , one after another .
 * < img width = " 640 " height = " 303 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . concat . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code concat } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param sources the sources to concatenate
 * @ return the Completable instance which completes only when all sources complete
 * @ throws NullPointerException if sources is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static Completable concat ( Iterable < ? extends CompletableSource > sources ) { } }

implementation: ObjectHelper . requireNonNull ( sources , "sources is null" ) ; return RxJavaPlugins . onAssembly ( new CompletableConcatIterable ( sources ) ) ;


signature:
public class Completable { /** * Returns a Completable which completes only when all sources complete , one after another .
 * < img width = " 640 " height = " 237 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . concat . p . png " alt = " " >
 * < dl >
 * < dt > < b > Backpressure : < / b > < / dt >
 * < dd > The returned { @ code Completable } honors the backpressure of the downstream consumer
 * and expects the other { @ code Publisher } to honor it as well . < / dd >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code concat } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param sources the sources to concatenate
 * @ return the Completable instance which completes only when all sources complete
 * @ throws NullPointerException if sources is null */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) @ BackpressureSupport ( BackpressureKind . FULL ) public static Completable concat ( Publisher < ? extends CompletableSource > sources ) { } }

implementation: return concat ( sources , 2 ) ;


signature:
public class Completable { /** * Returns a Completable which completes only when all sources complete , one after another .
 * < img width = " 640 " height = " 237 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . concat . pn . png " alt = " " >
 * < dl >
 * < dt > < b > Backpressure : < / b > < / dt >
 * < dd > The returned { @ code Completable } honors the backpressure of the downstream consumer
 * and expects the other { @ code Publisher } to honor it as well . < / dd >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code concat } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param sources the sources to concatenate
 * @ param prefetch the number of sources to prefetch from the sources
 * @ return the Completable instance which completes only when all sources complete
 * @ throws NullPointerException if sources is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) @ BackpressureSupport ( BackpressureKind . FULL ) public static Completable concat ( Publisher < ? extends CompletableSource > sources , int prefetch ) { } }

implementation: ObjectHelper . requireNonNull ( sources , "sources is null" ) ; ObjectHelper . verifyPositive ( prefetch , "prefetch" ) ; return RxJavaPlugins . onAssembly ( new CompletableConcat ( sources , prefetch ) ) ;


signature:
public class Completable { /** * Provides an API ( via a cold Completable ) that bridges the reactive world with the callback - style world .
 * < img width = " 640 " height = " 442 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . create . png " alt = " " >
 * Example :
 * < pre > < code >
 * Completable . create ( emitter - & gt ; {
 * Callback listener = new Callback ( ) {
 * & # 64 ; Override
 * public void onEvent ( Event e ) {
 * emitter . onComplete ( ) ;
 * & # 64 ; Override
 * public void onFailure ( Exception e ) {
 * emitter . onError ( e ) ;
 * AutoCloseable c = api . someMethod ( listener ) ;
 * emitter . setCancellable ( c : : close ) ;
 * < / code > < / pre >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code create } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param source the emitter that is called when a CompletableObserver subscribes to the returned { @ code Completable }
 * @ return the new Completable instance
 * @ see CompletableOnSubscribe
 * @ see Cancellable */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static Completable create ( CompletableOnSubscribe source ) { } }

implementation: ObjectHelper . requireNonNull ( source , "source is null" ) ; return RxJavaPlugins . onAssembly ( new CompletableCreate ( source ) ) ;


signature:
public class Completable { /** * Creates a Completable instance that emits the given Throwable exception to subscribers .
 * < img width = " 640 " height = " 462 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . error . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code error } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param error the Throwable instance to emit , not null
 * @ return the new Completable instance
 * @ throws NullPointerException if error is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static Completable error ( final Throwable error ) { } }

implementation: ObjectHelper . requireNonNull ( error , "error is null" ) ; return RxJavaPlugins . onAssembly ( new CompletableError ( error ) ) ;


signature:
public class Completable { /** * Returns a Completable instance that runs the given Action for each subscriber and
 * emits either an unchecked exception or simply completes .
 * < img width = " 640 " height = " 297 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . fromAction . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code fromAction } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < dt > < b > Error handling : < / b > < / dt >
 * < dd > If the { @ link Action } throws an exception , the respective { @ link Throwable } is
 * delivered to the downstream via { @ link CompletableObserver # onError ( Throwable ) } ,
 * except when the downstream has disposed this { @ code Completable } 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 run the runnable to run for each subscriber
 * @ return the new Completable instance
 * @ throws NullPointerException if run is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static Completable fromAction ( final Action run ) { } }

implementation: ObjectHelper . requireNonNull ( run , "run is null" ) ; return RxJavaPlugins . onAssembly ( new CompletableFromAction ( run ) ) ;


signature:
public class Completable { /** * Returns a Completable which when subscribed , executes the callable function , ignores its
 * normal result and emits onError or onComplete only .
 * < img width = " 640 " height = " 286 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . 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 CompletableObserver # onError ( Throwable ) } ,
 * except when the downstream has disposed this { @ code Completable } 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 the callable instance to execute for each subscriber
 * @ return the new Completable instance */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static Completable fromCallable ( final Callable < ? > callable ) { } }

implementation: ObjectHelper . requireNonNull ( callable , "callable is null" ) ; return RxJavaPlugins . onAssembly ( new CompletableFromCallable ( callable ) ) ;


signature:
public class Completable { /** * Returns a Completable instance that reacts to the termination of the given Future in a blocking fashion .
 * < img width = " 640 " height = " 628 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . fromFuture . png " alt = " " >
 * Note that if any of the observers to this Completable call dispose , this Completable will cancel the future .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code fromFuture } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param future the future to react to
 * @ return the new Completable instance */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static Completable fromFuture ( final Future < ? > future ) { } }

implementation: ObjectHelper . requireNonNull ( future , "future is null" ) ; return fromAction ( Functions . futureAction ( future ) ) ;


signature:
public class Completable { /** * Returns a Completable instance that when subscribed to , subscribes to the { @ code Maybe } instance and
 * emits a completion event if the maybe emits { @ code onSuccess } / { @ code onComplete } or forwards any
 * { @ code onError } events .
 * < img width = " 640 " height = " 235 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . fromMaybe . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code fromMaybe } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * < p > History : 2.1.17 - beta
 * @ param < T > the value type of the { @ link MaybeSource } element
 * @ param maybe the Maybe instance to subscribe to , not null
 * @ return the new Completable instance
 * @ throws NullPointerException if single is null
 * @ since 2.2 */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static < T > Completable fromMaybe ( final MaybeSource < T > maybe ) { } }

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


signature:
public class Completable { /** * Returns a Completable instance that runs the given Runnable for each subscriber and
 * emits either its exception or simply completes .
 * < img width = " 640 " height = " 297 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . fromRunnable . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code fromRunnable } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < dt > < b > Error handling : < / b > < / dt >
 * < dd > If the { @ link Runnable } throws an exception , the respective { @ link Throwable } is
 * delivered to the downstream via { @ link CompletableObserver # onError ( Throwable ) } ,
 * except when the downstream has disposed this { @ code Completable } 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 run the runnable to run for each subscriber
 * @ return the new Completable instance
 * @ throws NullPointerException if run is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static Completable fromRunnable ( final Runnable run ) { } }

implementation: ObjectHelper . requireNonNull ( run , "run is null" ) ; return RxJavaPlugins . onAssembly ( new CompletableFromRunnable ( run ) ) ;


signature:
public class Completable { /** * Returns a Completable instance that when subscribed to , subscribes to the Single instance and
 * emits a completion event if the single emits onSuccess or forwards any onError events .
 * < img width = " 640 " height = " 356 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . fromSingle . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code fromSingle } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type of the Single
 * @ param single the Single instance to subscribe to , not null
 * @ return the new Completable instance
 * @ throws NullPointerException if single is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static < T > Completable fromSingle ( final SingleSource < T > single ) { } }

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


signature:
public class Completable { /** * Returns a Completable instance that subscribes to all sources at once and
 * completes only when all source Completables complete or one of them emits an error .
 * < img width = " 640 " height = " 270 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . mergeArray . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code mergeArray } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < dt > < b > Error handling : < / b > < / dt >
 * < dd > If any of the source { @ code CompletableSource } s signal a { @ code Throwable } via { @ code onError } , the resulting
 * { @ code Completable } terminates with that { @ code Throwable } and all other source { @ code CompletableSource } s are disposed .
 * If more than one { @ code CompletableSource } signals an error , the resulting { @ code Completable } may terminate with the
 * first one ' s error or , depending on the concurrency of the sources , may terminate with a
 * { @ code CompositeException } containing two or more of the various error signals .
 * { @ code Throwable } s that didn ' t make into the composite will be sent ( individually ) to the global error handler via
 * { @ link RxJavaPlugins # onError ( Throwable ) } method as { @ code UndeliverableException } errors . Similarly , { @ code Throwable } s
 * signaled by source ( s ) after the returned { @ code Completable } has been disposed or terminated with a
 * ( composite ) error will be sent to the same global error handler .
 * Use { @ link # mergeArrayDelayError ( CompletableSource . . . ) } to merge sources and terminate only when all source { @ code CompletableSource } s
 * have completed or failed with an error .
 * < / dd >
 * < / dl >
 * @ param sources the iterable sequence of sources .
 * @ return the new Completable instance
 * @ throws NullPointerException if sources is null
 * @ see # mergeArrayDelayError ( CompletableSource . . . ) */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static Completable mergeArray ( CompletableSource ... sources ) { } }

implementation: ObjectHelper . requireNonNull ( sources , "sources is null" ) ; if ( sources . length == 0 ) { return complete ( ) ; } else if ( sources . length == 1 ) { return wrap ( sources [ 0 ] ) ; } return RxJavaPlugins . onAssembly ( new CompletableMergeArray ( sources ) ) ;


signature:
public class Completable { /** * Returns a Completable instance that subscribes to all sources at once and
 * completes only when all source Completables complete or one of them emits an error .
 * < img width = " 640 " height = " 311 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . merge . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code merge } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < dt > < b > Error handling : < / b > < / dt >
 * < dd > If any of the source { @ code CompletableSource } s signal a { @ code Throwable } via { @ code onError } , the resulting
 * { @ code Completable } terminates with that { @ code Throwable } and all other source { @ code CompletableSource } s are disposed .
 * If more than one { @ code CompletableSource } signals an error , the resulting { @ code Completable } may terminate with the
 * first one ' s error or , depending on the concurrency of the sources , may terminate with a
 * { @ code CompositeException } containing two or more of the various error signals .
 * { @ code Throwable } s that didn ' t make into the composite will be sent ( individually ) to the global error handler via
 * { @ link RxJavaPlugins # onError ( Throwable ) } method as { @ code UndeliverableException } errors . Similarly , { @ code Throwable } s
 * signaled by source ( s ) after the returned { @ code Completable } has been disposed or terminated with a
 * ( composite ) error will be sent to the same global error handler .
 * Use { @ link # mergeDelayError ( Iterable ) } to merge sources and terminate only when all source { @ code CompletableSource } s
 * have completed or failed with an error .
 * < / dd >
 * < / dl >
 * @ param sources the iterable sequence of sources .
 * @ return the new Completable instance
 * @ throws NullPointerException if sources is null
 * @ see # mergeDelayError ( Iterable ) */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static Completable merge ( Iterable < ? extends CompletableSource > sources ) { } }

implementation: ObjectHelper . requireNonNull ( sources , "sources is null" ) ; return RxJavaPlugins . onAssembly ( new CompletableMergeIterable ( sources ) ) ;


signature:
public class Completable { /** * Returns a Completable instance that subscribes to all sources at once and
 * completes only when all source Completables complete or one of them emits an error .
 * < img width = " 640 " height = " 336 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . merge . p . png " alt = " " >
 * < dl >
 * < dt > < b > Backpressure : < / b > < / dt >
 * < dd > The returned { @ code Completable } honors the backpressure of the downstream consumer
 * and expects the other { @ code Publisher } to honor it as well . < / dd >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code merge } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < dt > < b > Error handling : < / b > < / dt >
 * < dd > If any of the source { @ code CompletableSource } s signal a { @ code Throwable } via { @ code onError } , the resulting
 * { @ code Completable } terminates with that { @ code Throwable } and all other source { @ code CompletableSource } s are disposed .
 * If more than one { @ code CompletableSource } signals an error , the resulting { @ code Completable } may terminate with the
 * first one ' s error or , depending on the concurrency of the sources , may terminate with a
 * { @ code CompositeException } containing two or more of the various error signals .
 * { @ code Throwable } s that didn ' t make into the composite will be sent ( individually ) to the global error handler via
 * { @ link RxJavaPlugins # onError ( Throwable ) } method as { @ code UndeliverableException } errors . Similarly , { @ code Throwable } s
 * signaled by source ( s ) after the returned { @ code Completable } has been disposed or terminated with a
 * ( composite ) error will be sent to the same global error handler .
 * Use { @ link # mergeDelayError ( Publisher ) } to merge sources and terminate only when all source { @ code CompletableSource } s
 * have completed or failed with an error .
 * < / dd >
 * < / dl >
 * @ param sources the iterable sequence of sources .
 * @ return the new Completable instance
 * @ throws NullPointerException if sources is null
 * @ see # mergeDelayError ( Publisher ) */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) @ BackpressureSupport ( BackpressureKind . UNBOUNDED_IN ) public static Completable merge ( Publisher < ? extends CompletableSource > sources ) { } }

implementation: return merge0 ( sources , Integer . MAX_VALUE , false ) ;


signature:
public class Completable { /** * Returns a Completable instance that keeps subscriptions to a limited number of sources at once and
 * completes only when all source Completables complete or one of them emits an error .
 * < img width = " 640 " height = " 269 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . merge . pn . png " alt = " " >
 * < dl >
 * < dt > < b > Backpressure : < / b > < / dt >
 * < dd > The returned { @ code Completable } honors the backpressure of the downstream consumer
 * and expects the other { @ code Publisher } to honor it as well . < / dd >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code merge } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < dt > < b > Error handling : < / b > < / dt >
 * < dd > If any of the source { @ code CompletableSource } s signal a { @ code Throwable } via { @ code onError } , the resulting
 * { @ code Completable } terminates with that { @ code Throwable } and all other source { @ code CompletableSource } s are disposed .
 * If more than one { @ code CompletableSource } signals an error , the resulting { @ code Completable } may terminate with the
 * first one ' s error or , depending on the concurrency of the sources , may terminate with a
 * { @ code CompositeException } containing two or more of the various error signals .
 * { @ code Throwable } s that didn ' t make into the composite will be sent ( individually ) to the global error handler via
 * { @ link RxJavaPlugins # onError ( Throwable ) } method as { @ code UndeliverableException } errors . Similarly , { @ code Throwable } s
 * signaled by source ( s ) after the returned { @ code Completable } has been disposed or terminated with a
 * ( composite ) error will be sent to the same global error handler .
 * Use { @ link # mergeDelayError ( Publisher , int ) } to merge sources and terminate only when all source { @ code CompletableSource } s
 * have completed or failed with an error .
 * < / dd >
 * < / dl >
 * @ param sources the iterable sequence of sources .
 * @ param maxConcurrency the maximum number of concurrent subscriptions
 * @ return the new Completable instance
 * @ throws NullPointerException if sources is null
 * @ throws IllegalArgumentException if maxConcurrency is less than 1
 * @ see # mergeDelayError ( Publisher , int ) */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) @ BackpressureSupport ( BackpressureKind . FULL ) public static Completable merge ( Publisher < ? extends CompletableSource > sources , int maxConcurrency ) { } }

implementation: return merge0 ( sources , maxConcurrency , false ) ;


signature:
public class Completable { /** * Returns a Completable that subscribes to all Completables in the source sequence and delays
 * any error emitted by either the sources observable or any of the inner Completables until all of
 * them terminate in a way or another .
 * < img width = " 640 " height = " 466 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . mergeDelayError . p . png " alt = " " >
 * < dl >
 * < dt > < b > Backpressure : < / b > < / dt >
 * < dd > The returned { @ code Completable } honors the backpressure of the downstream consumer
 * and expects the other { @ code Publisher } to honor it as well . < / dd >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code mergeDelayError } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param sources the sequence of Completables
 * @ return the new Completable instance
 * @ throws NullPointerException if sources is null */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) @ BackpressureSupport ( BackpressureKind . UNBOUNDED_IN ) public static Completable mergeDelayError ( Publisher < ? extends CompletableSource > sources ) { } }

implementation: return merge0 ( sources , Integer . MAX_VALUE , true ) ;


signature:
public class Completable { /** * Returns a Completable that subscribes to a limited number of inner Completables at once in
 * the source sequence and delays any error emitted by either the sources
 * observable or any of the inner Completables until all of
 * them terminate in a way or another .
 * < img width = " 640 " height = " 440 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . mergeDelayError . pn . png " alt = " " >
 * < dl >
 * < dt > < b > Backpressure : < / b > < / dt >
 * < dd > The returned { @ code Completable } honors the backpressure of the downstream consumer
 * and expects the other { @ code Publisher } to honor it as well . < / dd >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code mergeDelayError } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param sources the sequence of Completables
 * @ param maxConcurrency the maximum number of concurrent subscriptions to Completables
 * @ return the new Completable instance
 * @ throws NullPointerException if sources is null */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) @ BackpressureSupport ( BackpressureKind . FULL ) public static Completable mergeDelayError ( Publisher < ? extends CompletableSource > sources , int maxConcurrency ) { } }

implementation: return merge0 ( sources , maxConcurrency , true ) ;


signature:
public class Completable { /** * Returns a Completable that never calls onError or onComplete .
 * < img width = " 640 " height = " 512 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . never . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code never } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ return the singleton instance that never calls onError or onComplete */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public static Completable never ( ) { } }

implementation: return RxJavaPlugins . onAssembly ( CompletableNever . INSTANCE ) ;


signature:
public class Completable { /** * Returns a Completable instance that fires its onComplete event after the given delay elapsed .
 * < img width = " 640 " height = " 413 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . timer . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code timer } does operate by default on the { @ code computation } { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param delay the delay time
 * @ param unit the delay unit
 * @ return the new Completable instance */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . COMPUTATION ) public static Completable timer ( long delay , TimeUnit unit ) { } }

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


signature:
public class Completable { /** * Creates a NullPointerException instance and sets the given Throwable as its initial cause .
 * @ param ex the Throwable instance to use as cause , not null ( not verified )
 * @ return the created NullPointerException */
private static NullPointerException toNpe ( Throwable ex ) { } }

implementation: NullPointerException npe = new NullPointerException ( "Actually not, but can't pass out an exception otherwise..." ) ; npe . initCause ( ex ) ; return npe ;


signature:
public class Completable { /** * Returns a Completable instance which manages a resource along
 * with a custom Completable instance while the subscription is active .
 * < img width = " 640 " height = " 388 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . using . png " alt = " " >
 * This overload disposes eagerly before the terminal event is emitted .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code using } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < R > the resource type
 * @ param resourceSupplier the supplier that returns a resource to be managed .
 * @ param completableFunction the function that given a resource returns a Completable instance that will be subscribed to
 * @ param disposer the consumer that disposes the resource created by the resource supplier
 * @ return the new Completable instance */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public static < R > Completable using ( Callable < R > resourceSupplier , Function < ? super R , ? extends CompletableSource > completableFunction , Consumer < ? super R > disposer ) { } }

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


signature:
public class Completable { /** * Wraps the given CompletableSource into a Completable
 * if not already Completable .
 * < img width = " 640 " height = " 354 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . wrap . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code wrap } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param source the source to wrap
 * @ return the source or its wrapper Completable
 * @ throws NullPointerException if source is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public static Completable wrap ( CompletableSource source ) { } }

implementation: ObjectHelper . requireNonNull ( source , "source is null" ) ; if ( source instanceof Completable ) { return RxJavaPlugins . onAssembly ( ( Completable ) source ) ; } return RxJavaPlugins . onAssembly ( new CompletableFromUnsafeSource ( source ) ) ;


signature:
public class Completable { /** * Returns a Completable that emits the a terminated event of either this Completable
 * or the other Completable whichever fires first .
 * < img width = " 640 " height = " 484 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . ambWith . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code ambWith } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param other the other Completable , not null . A subscription to this provided source will occur after subscribing
 * to the current source .
 * @ return the new Completable instance
 * @ throws NullPointerException if other is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable ambWith ( CompletableSource other ) { } }

implementation: ObjectHelper . requireNonNull ( other , "other is null" ) ; return ambArray ( this , other ) ;


signature:
public class Completable { /** * Returns an Observable which will subscribe to this Completable and once that is completed then
 * will subscribe to the { @ code next } ObservableSource . An error event from this Completable will be
 * propagated to the downstream subscriber and will result in skipping the subscription of the
 * Observable .
 * < img width = " 640 " height = " 278 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . andThen . o . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code andThen } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type of the next ObservableSource
 * @ param next the Observable to subscribe after this Completable is completed , not null
 * @ return Observable that composes this Completable and next
 * @ throws NullPointerException if next is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final < T > Observable < T > andThen ( ObservableSource < T > next ) { } }

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


signature:
public class Completable { /** * Returns a Flowable which will subscribe to this Completable and once that is completed then
 * will subscribe to the { @ code next } Flowable . An error event from this Completable will be
 * propagated to the downstream subscriber and will result in skipping the subscription of the
 * Publisher .
 * < img width = " 640 " height = " 249 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . andThen . p . png " alt = " " >
 * < dl >
 * < dt > < b > Backpressure : < / b > < / dt >
 * < dd > The returned { @ code Flowable } honors the backpressure of the downstream consumer
 * and expects the other { @ code Publisher } to honor it as well . < / dd >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code andThen } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type of the next Publisher
 * @ param next the Publisher to subscribe after this Completable is completed , not null
 * @ return Flowable that composes this Completable and next
 * @ throws NullPointerException if next is null */
@ CheckReturnValue @ NonNull @ BackpressureSupport ( BackpressureKind . FULL ) @ SchedulerSupport ( SchedulerSupport . NONE ) public final < T > Flowable < T > andThen ( Publisher < T > next ) { } }

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


signature:
public class Completable { /** * Returns a Single which will subscribe to this Completable and once that is completed then
 * will subscribe to the { @ code next } SingleSource . An error event from this Completable will be
 * propagated to the downstream subscriber and will result in skipping the subscription of the
 * Single .
 * < img width = " 640 " height = " 437 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . andThen . s . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code andThen } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type of the next SingleSource
 * @ param next the Single to subscribe after this Completable is completed , not null
 * @ return Single that composes this Completable and next */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final < T > Single < T > andThen ( SingleSource < T > next ) { } }

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


signature:
public class Completable { /** * Returns a { @ link Maybe } which will subscribe to this Completable and once that is completed then
 * will subscribe to the { @ code next } MaybeSource . An error event from this Completable will be
 * propagated to the downstream subscriber and will result in skipping the subscription of the
 * Maybe .
 * < img width = " 640 " height = " 280 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . andThen . m . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code andThen } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param < T > the value type of the next MaybeSource
 * @ param next the Maybe to subscribe after this Completable is completed , not null
 * @ return Maybe that composes this Completable and next */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final < T > Maybe < T > andThen ( MaybeSource < T > next ) { } }

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


signature:
public class Completable { /** * Returns a Completable that first runs this Completable
 * and then the other completable .
 * < img width = " 640 " height = " 437 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . andThen . c . png " alt = " " >
 * This is an alias for { @ link # concatWith ( CompletableSource ) } .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code andThen } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param next the other Completable , not null
 * @ return the new Completable instance
 * @ throws NullPointerException if other is null */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable andThen ( CompletableSource next ) { } }

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


signature:
public class Completable { /** * Calls the specified converter function during assembly time and returns its resulting value .
 * < img width = " 640 " height = " 751 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . 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 Completable 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 CompletableConverter < ? extends R > converter ) { } }

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


signature:
public class Completable { /** * Subscribes to and awaits the termination of this Completable instance in a blocking manner and
 * rethrows any exception emitted .
 * < img width = " 640 " height = " 432 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . blockingAwait . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code blockingAwait } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < dt > < b > Error handling : < / b > < / dt >
 * < dd > If the source signals an error , the operator wraps a checked { @ link Exception }
 * into { @ link RuntimeException } and throws that . Otherwise , { @ code RuntimeException } s and
 * { @ link Error } s are rethrown as they are . < / dd >
 * < / dl >
 * @ throws RuntimeException wrapping an InterruptedException if the current thread is interrupted */
@ SchedulerSupport ( SchedulerSupport . NONE ) public final void blockingAwait ( ) { } }

implementation: BlockingMultiObserver < Void > observer = new BlockingMultiObserver < Void > ( ) ; subscribe ( observer ) ; observer . blockingGet ( ) ;


signature:
public class Completable { /** * Calls the given transformer function with this instance and returns the function ' s resulting
 * Completable .
 * < img width = " 640 " height = " 625 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . compose . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code compose } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param transformer the transformer function , not null
 * @ return the Completable returned by the function
 * @ throws NullPointerException if transformer is null */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable compose ( CompletableTransformer transformer ) { } }

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


signature:
public class Completable { /** * Concatenates this Completable with another Completable .
 * < img width = " 640 " height = " 317 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . concatWith . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code concatWith } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param other the other Completable , not null
 * @ return the new Completable which subscribes to this and then the other Completable
 * @ throws NullPointerException if other is null
 * @ see # andThen ( MaybeSource )
 * @ see # andThen ( ObservableSource )
 * @ see # andThen ( SingleSource )
 * @ see # andThen ( Publisher ) */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable concatWith ( CompletableSource other ) { } }

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


signature:
public class Completable { /** * Returns a Completable which delays the emission of the completion event by the given time while
 * running on the specified scheduler .
 * < img width = " 640 " height = " 313 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . delay . s . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code delay } operates on the { @ link Scheduler } you specify . < / dd >
 * < / dl >
 * @ param delay the delay time
 * @ param unit the delay unit
 * @ param scheduler the scheduler to run the delayed completion on
 * @ return the new Completable instance
 * @ throws NullPointerException if unit or scheduler is null */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . CUSTOM ) public final Completable delay ( long delay , TimeUnit unit , Scheduler scheduler ) { } }

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


signature:
public class Completable { /** * Returns a Completable which delays the emission of the completion event , and optionally the error as well , by the given time while
 * running on the specified scheduler .
 * < img width = " 640 " height = " 253 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . delay . sb . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code delay } operates on the { @ link Scheduler } you specify . < / dd >
 * < / dl >
 * @ param delay the delay time
 * @ param unit the delay unit
 * @ param scheduler the scheduler to run the delayed completion on
 * @ param delayError delay the error emission as well ?
 * @ return the new Completable instance
 * @ throws NullPointerException if unit or scheduler is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . CUSTOM ) public final Completable delay ( final long delay , final TimeUnit unit , final Scheduler scheduler , final boolean delayError ) { } }

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


signature:
public class Completable { /** * Returns a Completable that delays the subscription to the source CompletableSource by a given amount of time .
 * < img width = " 640 " height = " 475 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . delaySubscription . t . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > This version of { @ code delaySubscription } operates by default on the { @ code computation } { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param delay the time to delay the subscription
 * @ param unit the time unit of { @ code delay }
 * @ return a Completable that delays the subscription to the source CompletableSource by the given amount
 * @ since 2.2.3 - experimental
 * @ see < a href = " http : / / reactivex . io / documentation / operators / delay . html " > ReactiveX operators documentation : Delay < / a > */
@ CheckReturnValue @ Experimental @ SchedulerSupport ( SchedulerSupport . COMPUTATION ) public final Completable delaySubscription ( long delay , TimeUnit unit ) { } }

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


signature:
public class Completable { /** * Returns a Completable that delays the subscription to the source CompletableSource by a given amount of time ,
 * both waiting and subscribing on a given Scheduler .
 * < img width = " 640 " height = " 420 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . delaySubscription . ts . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > You specify which { @ link Scheduler } this operator will use . < / dd >
 * < / dl >
 * @ param delay the time to delay the subscription
 * @ param unit the time unit of { @ code delay }
 * @ param scheduler the Scheduler on which the waiting and subscription will happen
 * @ return a Completable that delays the subscription to the source CompletableSource by a given
 * amount , waiting and subscribing on the given Scheduler
 * @ since 2.2.3 - experimental
 * @ see < a href = " http : / / reactivex . io / documentation / operators / delay . html " > ReactiveX operators documentation : Delay < / a > */
@ CheckReturnValue @ Experimental @ SchedulerSupport ( SchedulerSupport . CUSTOM ) public final Completable delaySubscription ( long delay , TimeUnit unit , Scheduler scheduler ) { } }

implementation: return Completable . timer ( delay , unit , scheduler ) . andThen ( this ) ;


signature:
public class Completable { /** * Returns a Completable which calls the given onComplete callback if this Completable completes .
 * < img width = " 640 " height = " 304 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . doOnComplete . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code doOnComplete } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param onComplete the callback to call when this emits an onComplete event
 * @ return the new Completable instance
 * @ throws NullPointerException if onComplete is null
 * @ see # doFinally ( Action ) */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable doOnComplete ( Action onComplete ) { } }

implementation: return doOnLifecycle ( Functions . emptyConsumer ( ) , Functions . emptyConsumer ( ) , onComplete , Functions . EMPTY_ACTION , Functions . EMPTY_ACTION , Functions . EMPTY_ACTION ) ;


signature:
public class Completable { /** * Calls the shared { @ code Action } if a CompletableObserver subscribed to the current
 * Completable disposes the common Disposable it received via onSubscribe .
 * < img width = " 640 " height = " 589 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . 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 to call when the child subscriber disposes the subscription
 * @ return the new Completable instance
 * @ throws NullPointerException if onDispose is null */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable doOnDispose ( Action onDispose ) { } }

implementation: return doOnLifecycle ( Functions . emptyConsumer ( ) , Functions . emptyConsumer ( ) , Functions . EMPTY_ACTION , Functions . EMPTY_ACTION , Functions . EMPTY_ACTION , onDispose ) ;


signature:
public class Completable { /** * Returns a Completable which calls the given onError callback if this Completable emits an error .
 * < img width = " 640 " height = " 304 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . doOnError . 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 error callback
 * @ return the new Completable instance
 * @ throws NullPointerException if onError is null
 * @ see # doFinally ( Action ) */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable doOnError ( Consumer < ? super Throwable > onError ) { } }

implementation: return doOnLifecycle ( Functions . emptyConsumer ( ) , onError , Functions . EMPTY_ACTION , Functions . EMPTY_ACTION , Functions . EMPTY_ACTION , Functions . EMPTY_ACTION ) ;


signature:
public class Completable { /** * Returns a Completable instance that calls the various callbacks on the specific
 * lifecycle events .
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code doOnLifecycle } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param onSubscribe the consumer called when a CompletableSubscriber subscribes .
 * @ param onError the consumer called when this emits an onError event
 * @ param onComplete the runnable called just before when this Completable completes normally
 * @ param onAfterTerminate the runnable called after this Completable completes normally
 * @ param onDispose the runnable called when the child disposes the subscription
 * @ return the new Completable instance */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) private Completable doOnLifecycle ( final Consumer < ? super Disposable > onSubscribe , final Consumer < ? super Throwable > onError , final Action onComplete , final Action onTerminate , final Action onAfterTerminate , final Action onDispose ) { } }

implementation: ObjectHelper . requireNonNull ( onSubscribe , "onSubscribe is null" ) ; ObjectHelper . requireNonNull ( onError , "onError is null" ) ; ObjectHelper . requireNonNull ( onComplete , "onComplete is null" ) ; ObjectHelper . requireNonNull ( onTerminate , "onTerminate is null" ) ; ObjectHelper . requireNonNull ( onAfterTerminate , "onAfterTerminate is null" ) ; ObjectHelper . requireNonNull ( onDispose , "onDispose is null" ) ; return RxJavaPlugins . onAssembly ( new CompletablePeek ( this , onSubscribe , onError , onComplete , onTerminate , onAfterTerminate , onDispose ) ) ;


signature:
public class Completable { /** * Returns a Completable instance that calls the given onSubscribe callback with the disposable
 * that child subscribers receive on subscription .
 * < img width = " 640 " height = " 304 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . 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 callback called when a child subscriber subscribes
 * @ return the new Completable instance
 * @ throws NullPointerException if onSubscribe is null */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable doOnSubscribe ( Consumer < ? super Disposable > onSubscribe ) { } }

implementation: return doOnLifecycle ( onSubscribe , Functions . emptyConsumer ( ) , Functions . EMPTY_ACTION , Functions . EMPTY_ACTION , Functions . EMPTY_ACTION , Functions . EMPTY_ACTION ) ;


signature:
public class Completable { /** * Returns a Completable instance that calls the given onTerminate callback just before this Completable
 * completes normally or with an exception .
 * < img width = " 640 " height = " 304 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . doOnTerminate . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code doOnTerminate } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param onTerminate the callback to call just before this Completable terminates
 * @ return the new Completable instance
 * @ see # doFinally ( Action ) */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable doOnTerminate ( final Action onTerminate ) { } }

implementation: return doOnLifecycle ( Functions . emptyConsumer ( ) , Functions . emptyConsumer ( ) , Functions . EMPTY_ACTION , onTerminate , Functions . EMPTY_ACTION , Functions . EMPTY_ACTION ) ;


signature:
public class Completable { /** * Returns a Completable instance that calls the given onTerminate callback after this Completable
 * completes normally or with an exception .
 * < img width = " 640 " height = " 304 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . doAfterTerminate . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code doAfterTerminate } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param onAfterTerminate the callback to call after this Completable terminates
 * @ return the new Completable instance
 * @ see # doFinally ( Action ) */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable doAfterTerminate ( final Action onAfterTerminate ) { } }

implementation: return doOnLifecycle ( Functions . emptyConsumer ( ) , Functions . emptyConsumer ( ) , Functions . EMPTY_ACTION , Functions . EMPTY_ACTION , onAfterTerminate , Functions . EMPTY_ACTION ) ;


signature:
public class Completable { /** * Returns a Completable which emits the terminal events from the thread of the specified scheduler .
 * < img width = " 640 " height = " 523 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . observeOn . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code observeOn } operates on a { @ link Scheduler } you specify . < / dd >
 * < / dl >
 * @ param scheduler the scheduler to emit terminal events on
 * @ return the new Completable instance
 * @ throws NullPointerException if scheduler is null */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . CUSTOM ) public final Completable observeOn ( final Scheduler scheduler ) { } }

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


signature:
public class Completable { /** * Returns a Completable instance that if this Completable emits an error and the predicate returns
 * true , it will emit an onComplete and swallow the throwable .
 * < img width = " 640 " height = " 283 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . onErrorComplete . f . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code onErrorComplete } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param predicate the predicate to call when an Throwable is emitted which should return true
 * if the Throwable should be swallowed and replaced with an onComplete .
 * @ return the new Completable instance */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable onErrorComplete ( final Predicate < ? super Throwable > predicate ) { } }

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


signature:
public class Completable { /** * Returns a Completable instance that when encounters an error from this Completable , calls the
 * specified mapper function that returns another Completable instance for it and resumes the
 * execution with it .
 * < img width = " 640 " height = " 426 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . onErrorResumeNext . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code onErrorResumeNext } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param errorMapper the mapper function that takes the error and should return a Completable as
 * continuation .
 * @ return the new Completable instance */
@ CheckReturnValue @ NonNull @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable onErrorResumeNext ( final Function < ? super Throwable , ? extends CompletableSource > errorMapper ) { } }

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


signature:
public class Completable { /** * Returns a Completable that subscribes repeatedly at most the given times to this Completable .
 * < img width = " 640 " height = " 408 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . repeat . n . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code repeat } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param times the number of times the resubscription should happen
 * @ return the new Completable instance
 * @ throws IllegalArgumentException if times is less than zero */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable repeat ( long times ) { } }

implementation: return fromPublisher ( toFlowable ( ) . repeat ( times ) ) ;


signature:
public class Completable { /** * Returns a Completable that repeatedly subscribes to this Completable so long as the given
 * stop supplier returns false .
 * < img width = " 640 " height = " 381 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . repeatUntil . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code repeatUntil } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param stop the supplier that should return true to stop resubscribing .
 * @ return the new Completable instance
 * @ throws NullPointerException if stop is null */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable repeatUntil ( BooleanSupplier stop ) { } }

implementation: return fromPublisher ( toFlowable ( ) . repeatUntil ( stop ) ) ;


signature:
public class Completable { /** * Returns a Completable instance that repeats when the Publisher returned by the handler
 * emits an item or completes when this Publisher emits a completed event .
 * < img width = " 640 " height = " 586 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . repeatWhen . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code repeatWhen } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param handler the function that transforms the stream of values indicating the completion of
 * this Completable and returns a Publisher that emits items for repeating or completes to indicate the
 * repetition should stop
 * @ return the new Completable instance
 * @ throws NullPointerException if stop is null */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable repeatWhen ( Function < ? super Flowable < Object > , ? extends Publisher < ? > > handler ) { } }

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


signature:
public class Completable { /** * Returns a Completable that retries this Completable in case of an error as long as the predicate
 * returns true .
 * < img width = " 640 " height = " 325 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . retry . ff . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code retry } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param predicate the predicate called when this emits an error with the repeat count and the latest exception
 * and should return true to retry .
 * @ return the new Completable instance */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable retry ( BiPredicate < ? super Integer , ? super Throwable > predicate ) { } }

implementation: return fromPublisher ( toFlowable ( ) . retry ( predicate ) ) ;


signature:
public class Completable { /** * Returns a Completable that when this Completable emits an error , retries at most the given
 * number of times before giving up and emitting the last error .
 * < img width = " 640 " height = " 451 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . retry . n . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code retry } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * @ param times the number of times to resubscribe if the current Completable fails
 * @ return the new Completable instance
 * @ throws IllegalArgumentException if times is negative */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable retry ( long times ) { } }

implementation: return fromPublisher ( toFlowable ( ) . retry ( times ) ) ;


signature:
public class Completable { /** * Returns a Completable that when this Completable emits an error , retries at most times
 * or until the predicate returns false , whichever happens first and emitting the last error .
 * < img width = " 640 " height = " 361 " src = " https : / / raw . github . com / wiki / ReactiveX / RxJava / images / rx - operators / Completable . retry . nf . png " alt = " " >
 * < dl >
 * < dt > < b > Scheduler : < / b > < / dt >
 * < dd > { @ code retry } does not operate by default on a particular { @ link Scheduler } . < / dd >
 * < / dl >
 * < p > History : 2.1.8 - experimental
 * @ param times the number of times to resubscribe if the current Completable fails
 * @ param predicate the predicate that is called with the latest throwable and should return
 * true to indicate the returned Completable should resubscribe to this Completable .
 * @ return the new Completable instance
 * @ throws NullPointerException if predicate is null
 * @ throws IllegalArgumentException if times is negative
 * @ since 2.2 */
@ CheckReturnValue @ SchedulerSupport ( SchedulerSupport . NONE ) public final Completable retry ( long times , Predicate < ? super Throwable > predicate ) { } }

implementation: return fromPublisher ( toFlowable ( ) . retry ( times , predicate ) ) ;