signature:
public class Operators { /** * Concurrent addition bound to Long . MAX _ VALUE .
 * Any concurrent write will " happen before " this operation .
 * @ param < T > the parent instance type
 * @ param updater current field updater
 * @ param instance current instance to update
 * @ param toAdd delta to add
 * @ return value before addition or Long . MAX _ VALUE */
public static < T > long addCap ( AtomicLongFieldUpdater < T > updater , T instance , long toAdd ) { } }

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


signature:
public class Operators { /** * Returns the subscription as QueueSubscription if possible or null .
 * @ param < T > the value type of the QueueSubscription .
 * @ param s the source subscription to try to convert .
 * @ return the QueueSubscription instance or null */
@ SuppressWarnings ( "unchecked" ) @ Nullable public static < T > QueueSubscription < T > as ( Subscription s ) { } }

implementation: if ( s instanceof QueueSubscription ) { return ( QueueSubscription < T > ) s ; } return null ;


signature:
public class Operators { /** * Calls onSubscribe on the target Subscriber with the empty instance followed by a call to onError with the
 * supplied error .
 * @ param s target Subscriber to error
 * @ param e the actual error */
public static void error ( Subscriber < ? > s , Throwable e ) { } }

implementation: s . onSubscribe ( EmptySubscription . INSTANCE ) ; s . onError ( e ) ;


signature:
public class Operators { /** * Create a function that can be used to support a custom operator via
 * { @ link CoreSubscriber } decoration . The function is compatible with
 * { @ link Flux # transform ( Function ) } , { @ link Mono # transform ( Function ) } ,
 * { @ link Hooks # onEachOperator ( Function ) } and { @ link Hooks # onLastOperator ( Function ) } ,
 * but requires that the original { @ link Publisher } be { @ link Scannable } .
 * This variant attempts to expose the { @ link Publisher } as a { @ link Scannable } for
 * convenience of introspection . You should however avoid instanceof checks or any
 * other processing that depends on identity of the { @ link Publisher } , as it might
 * get hidden if { @ link Scannable # isScanAvailable ( ) } returns { @ code false } .
 * Use { @ link # liftPublisher ( BiFunction ) } instead for that kind of use case .
 * @ param lifter the bifunction taking { @ link Scannable } from the enclosing
 * publisher ( assuming it is compatible ) and consuming { @ link CoreSubscriber } .
 * It must return a receiving { @ link CoreSubscriber } that will immediately subscribe
 * to the applied { @ link Publisher } .
 * @ param < I > the input type
 * @ param < O > the output type
 * @ return a new { @ link Function }
 * @ see # liftPublisher ( BiFunction ) */
public static < I , O > Function < ? super Publisher < I > , ? extends Publisher < O > > lift ( BiFunction < Scannable , ? super CoreSubscriber < ? super O > , ? extends CoreSubscriber < ? super I > > lifter ) { } }

implementation: return LiftFunction . liftScannable ( null , lifter ) ;


signature:
public class Operators { /** * Create a function that can be used to support a custom operator via
 * { @ link CoreSubscriber } decoration . The function is compatible with
 * { @ link Flux # transform ( Function ) } , { @ link Mono # transform ( Function ) } ,
 * { @ link Hooks # onEachOperator ( Function ) } and { @ link Hooks # onLastOperator ( Function ) } ,
 * but requires that the original { @ link Publisher } be { @ link Scannable } .
 * This variant attempts to expose the { @ link Publisher } as a { @ link Scannable } for
 * convenience of introspection . You should however avoid instanceof checks or any
 * other processing that depends on identity of the { @ link Publisher } , as it might
 * get hidden if { @ link Scannable # isScanAvailable ( ) } returns { @ code false } .
 * Use { @ link # liftPublisher ( Predicate , BiFunction ) } instead for that kind of use case .
 * The function will be invoked only if the passed { @ link Predicate } matches .
 * Therefore the transformed type O must be the same than the input type since
 * unmatched predicate will return the applied { @ link Publisher } .
 * @ param filter the predicate to match taking { @ link Scannable } from the applied
 * publisher to operate on . Assumes original is scan - compatible .
 * @ param lifter the bifunction taking { @ link Scannable } from the enclosing
 * publisher and consuming { @ link CoreSubscriber } . It must return a receiving
 * { @ link CoreSubscriber } that will immediately subscribe to the applied
 * { @ link Publisher } . Assumes the original is scan - compatible .
 * @ param < O > the input and output type
 * @ return a new { @ link Function }
 * @ see # liftPublisher ( Predicate , BiFunction ) */
public static < O > Function < ? super Publisher < O > , ? extends Publisher < O > > lift ( Predicate < Scannable > filter , BiFunction < Scannable , ? super CoreSubscriber < ? super O > , ? extends CoreSubscriber < ? super O > > lifter ) { } }

implementation: return LiftFunction . liftScannable ( filter , lifter ) ;


signature:
public class Operators { /** * Create a function that can be used to support a custom operator via
 * { @ link CoreSubscriber } decoration . The function is compatible with
 * { @ link Flux # transform ( Function ) } , { @ link Mono # transform ( Function ) } ,
 * { @ link Hooks # onEachOperator ( Function ) } and { @ link Hooks # onLastOperator ( Function ) } ,
 * and works with the raw { @ link Publisher } as input , which is useful if you need to
 * detect the precise type of the source ( eg . instanceof checks to detect Mono , Flux ,
 * true Scannable , etc . . . ) .
 * @ param lifter the bifunction taking the raw { @ link Publisher } and
 * { @ link CoreSubscriber } . The publisher can be double - checked ( including with
 * { @ code instanceof } , and the function must return a receiving { @ link CoreSubscriber }
 * that will immediately subscribe to the { @ link Publisher } .
 * @ param < I > the input type
 * @ param < O > the output type
 * @ return a new { @ link Function } */
public static < I , O > Function < ? super Publisher < I > , ? extends Publisher < O > > liftPublisher ( BiFunction < Publisher , ? super CoreSubscriber < ? super O > , ? extends CoreSubscriber < ? super I > > lifter ) { } }

implementation: return LiftFunction . liftPublisher ( null , lifter ) ;


signature:
public class Operators { /** * Create a function that can be used to support a custom operator via
 * { @ link CoreSubscriber } decoration . The function is compatible with
 * { @ link Flux # transform ( Function ) } , { @ link Mono # transform ( Function ) } ,
 * { @ link Hooks # onEachOperator ( Function ) } and { @ link Hooks # onLastOperator ( Function ) } ,
 * and works with the raw { @ link Publisher } as input , which is useful if you need to
 * detect the precise type of the source ( eg . instanceof checks to detect Mono , Flux ,
 * true Scannable , etc . . . ) .
 * The function will be invoked only if the passed { @ link Predicate } matches .
 * Therefore the transformed type O must be the same than the input type since
 * unmatched predicate will return the applied { @ link Publisher } .
 * @ param filter the { @ link Predicate } that the raw { @ link Publisher } must pass for
 * the transformation to occur
 * @ param lifter the { @ link BiFunction } taking the raw { @ link Publisher } and
 * { @ link CoreSubscriber } . The publisher can be double - checked ( including with
 * { @ code instanceof } , and the function must return a receiving { @ link CoreSubscriber }
 * that will immediately subscribe to the { @ link Publisher } .
 * @ param < O > the input and output type
 * @ return a new { @ link Function } */
public static < O > Function < ? super Publisher < O > , ? extends Publisher < O > > liftPublisher ( Predicate < Publisher > filter , BiFunction < Publisher , ? super CoreSubscriber < ? super O > , ? extends CoreSubscriber < ? super O > > lifter ) { } }

implementation: return LiftFunction . liftPublisher ( filter , lifter ) ;


signature:
public class Operators { /** * Create an adapter for local onDiscard hooks that check the element
 * being discarded is of a given { @ link Class } . The resulting { @ link Function } adds the
 * hook to the { @ link Context } , potentially chaining it to an existing hook in the { @ link Context } .
 * @ param type the type of elements to take into account
 * @ param discardHook the discarding handler for this type of elements
 * @ param < R > element type
 * @ return a { @ link Function } that can be used to modify a { @ link Context } , adding or
 * updating a context - local discard hook . */
static final < R > Function < Context , Context > discardLocalAdapter ( Class < R > type , Consumer < ? super R > discardHook ) { } }

implementation: Objects . requireNonNull ( type , "onDiscard must be based on a type" ) ; Objects . requireNonNull ( discardHook , "onDiscard must be provided a discardHook Consumer" ) ; final Consumer < Object > safeConsumer = obj -> { if ( type . isInstance ( obj ) ) { discardHook . accept ( type . cast ( obj ) ) ; } } ; return ctx -> { Consumer < Object > consumer = ctx . getOrDefault ( Hooks . KEY_ON_DISCARD , null ) ; if ( consumer == null ) { return ctx . put ( Hooks . KEY_ON_DISCARD , safeConsumer ) ; } else { return ctx . put ( Hooks . KEY_ON_DISCARD , safeConsumer . andThen ( consumer ) ) ; } } ;


signature:
public class Operators { /** * Utility method to activate the onDiscard feature ( see { @ link Flux # doOnDiscard ( Class , Consumer ) } )
 * in a target { @ link Context } . Prefer using the { @ link Flux } API , and reserve this for
 * testing purposes .
 * @ param target the original { @ link Context }
 * @ param discardConsumer the consumer that will be used to cleanup discarded elements
 * @ return a new { @ link Context } that holds ( potentially combined ) cleanup { @ link Consumer } */
public static final Context enableOnDiscard ( @ Nullable Context target , Consumer < ? > discardConsumer ) { } }

implementation: Objects . requireNonNull ( discardConsumer , "discardConsumer must be provided" ) ; if ( target == null ) { return Context . of ( Hooks . KEY_ON_DISCARD , discardConsumer ) ; } return target . put ( Hooks . KEY_ON_DISCARD , discardConsumer ) ;


signature:
public class Operators { /** * Invoke a ( local or global ) hook that processes elements that get discarded . This
 * includes elements that are dropped ( for malformed sources ) , but also filtered out
 * ( eg . not passing a { @ code filter ( ) } predicate ) .
 * For elements that are buffered or enqueued , but subsequently discarded due to
 * cancellation or error , see { @ link # onDiscardMultiple ( Stream , Context ) } and
 * { @ link # onDiscardQueueWithClear ( Queue , Context , Function ) } .
 * @ param element the element that is being discarded
 * @ param context the context in which to look for a local hook
 * @ param < T > the type of the element
 * @ see # onDiscardMultiple ( Stream , Context )
 * @ see # onDiscardMultiple ( Collection , Context )
 * @ see # onDiscardQueueWithClear ( Queue , Context , Function ) */
public static < T > void onDiscard ( @ Nullable T element , Context context ) { } }

implementation: Consumer < Object > hook = context . getOrDefault ( Hooks . KEY_ON_DISCARD , null ) ; if ( element != null && hook != null ) { try { hook . accept ( element ) ; } catch ( Throwable t ) { log . warn ( "Error in discard hook" , t ) ; } }


signature:
public class Operators { /** * Invoke a ( local or global ) hook that processes elements that get discarded
 * en masse after having been enqueued , due to cancellation or error . This method
 * also empties the { @ link Queue } ( either by repeated { @ link Queue # poll ( ) } calls if
 * a hook is defined , or by { @ link Queue # clear ( ) } as a shortcut if no hook is defined ) .
 * @ param queue the queue that is being discarded and cleared
 * @ param context the context in which to look for a local hook
 * @ param extract an optional extractor method for cases where the queue doesn ' t
 * directly contain the elements to discard
 * @ param < T > the type of the element
 * @ see # onDiscardMultiple ( Stream , Context )
 * @ see # onDiscardMultiple ( Collection , Context )
 * @ see # onDiscard ( Object , Context ) */
public static < T > void onDiscardQueueWithClear ( @ Nullable Queue < T > queue , Context context , @ Nullable Function < T , Stream < ? > > extract ) { } }

implementation: if ( queue == null ) { return ; } Consumer < Object > hook = context . getOrDefault ( Hooks . KEY_ON_DISCARD , null ) ; if ( hook == null ) { queue . clear ( ) ; return ; } try { for ( ; ; ) { T toDiscard = queue . poll ( ) ; if ( toDiscard == null ) { break ; } if ( extract != null ) { extract . apply ( toDiscard ) . forEach ( hook ) ; } else { hook . accept ( toDiscard ) ; } } } catch ( Throwable t ) { log . warn ( "Error in discard hook while discarding and clearing a queue" , t ) ; }


signature:
public class Operators { /** * Invoke a ( local or global ) hook that processes elements that get discarded en masse .
 * This includes elements that are buffered but subsequently discarded due to
 * cancellation or error .
 * @ param multiple the collection of elements to discard ( possibly extracted from other
 * collections / arrays / queues )
 * @ param context the { @ link Context } in which to look for local hook
 * @ see # onDiscard ( Object , Context )
 * @ see # onDiscardMultiple ( Collection , Context )
 * @ see # onDiscardQueueWithClear ( Queue , Context , Function ) */
public static void onDiscardMultiple ( Stream < ? > multiple , Context context ) { } }

implementation: Consumer < Object > hook = context . getOrDefault ( Hooks . KEY_ON_DISCARD , null ) ; if ( hook != null ) { try { multiple . forEach ( hook ) ; } catch ( Throwable t ) { log . warn ( "Error in discard hook while discarding multiple values" , t ) ; } }


signature:
public class Operators { /** * An unexpected exception is about to be dropped .
 * If no hook is registered for { @ link Hooks # onErrorDropped ( Consumer ) } , the dropped
 * error is logged at ERROR level and thrown ( via { @ link Exceptions # bubble ( Throwable ) } .
 * @ param e the dropped exception
 * @ param context a context that might hold a local error consumer */
public static void onErrorDropped ( Throwable e , Context context ) { } }

implementation: Consumer < ? super Throwable > hook = context . getOrDefault ( Hooks . KEY_ON_ERROR_DROPPED , null ) ; if ( hook == null ) { hook = Hooks . onErrorDroppedHook ; } if ( hook == null ) { log . error ( "Operator called default onErrorDropped" , e ) ; throw Exceptions . bubble ( e ) ; } hook . accept ( e ) ;


signature:
public class Operators { /** * An unexpected event is about to be dropped .
 * If no hook is registered for { @ link Hooks # onNextDropped ( Consumer ) } , the dropped
 * element is just logged at DEBUG level .
 * @ param < T > the dropped value type
 * @ param t the dropped data
 * @ param context a context that might hold a local next consumer */
public static < T > void onNextDropped ( T t , Context context ) { } }

implementation: Objects . requireNonNull ( t , "onNext" ) ; Objects . requireNonNull ( context , "context" ) ; Consumer < Object > hook = context . getOrDefault ( Hooks . KEY_ON_NEXT_DROPPED , null ) ; if ( hook == null ) { hook = Hooks . onNextDroppedHook ; } if ( hook != null ) { hook . accept ( t ) ; } else if ( log . isDebugEnabled ( ) ) { log . debug ( "onNextDropped: " + t ) ; }


signature:
public class Operators { /** * Map an " operator " error . The
 * result error will be passed via onError to the operator downstream after
 * checking for fatal error via
 * { @ link Exceptions # throwIfFatal ( Throwable ) } .
 * @ param error the callback or operator error
 * @ param context a context that might hold a local error consumer
 * @ return mapped { @ link Throwable } */
public static Throwable onOperatorError ( Throwable error , Context context ) { } }

implementation: return onOperatorError ( null , error , context ) ;


signature:
public class Operators { /** * Map an " operator " error given an operator parent { @ link Subscription } . The
 * result error will be passed via onError to the operator downstream .
 * { @ link Subscription } will be cancelled after checking for fatal error via
 * { @ link Exceptions # throwIfFatal ( Throwable ) } .
 * @ param subscription the linked operator parent { @ link Subscription }
 * @ param error the callback or operator error
 * @ param context a context that might hold a local error consumer
 * @ return mapped { @ link Throwable } */
public static Throwable onOperatorError ( @ Nullable Subscription subscription , Throwable error , Context context ) { } }

implementation: return onOperatorError ( subscription , error , null , context ) ;


signature:
public class Operators { /** * Return a wrapped { @ link RejectedExecutionException } which can be thrown by the
 * operator . This exception denotes that an execution was rejected by a
 * { @ link reactor . core . scheduler . Scheduler } , notably when it was already disposed .
 * Wrapping is done by calling both { @ link Exceptions # bubble ( Throwable ) } and
 * { @ link # onOperatorError ( Subscription , Throwable , Object , Context ) } .
 * @ param original the original execution error
 * @ param context a context that might hold a local error consumer */
public static RuntimeException onRejectedExecution ( Throwable original , Context context ) { } }

implementation: return onRejectedExecution ( original , null , null , null , context ) ;


signature:
public class Operators { /** * Find the { @ link OnNextFailureStrategy } to apply to the calling operator ( which could be a local
 * error mode defined in the { @ link Context } ) and apply it . For poll ( ) , prefer
 * { @ link # onNextPollError ( Object , Throwable , Context ) } as it returns a { @ link RuntimeException } .
 * Cancels the { @ link Subscription } and return a { @ link Throwable } if errors are
 * fatal for the error mode , in which case the operator should call onError with the
 * returned error . On the contrary , if the error mode allows the sequence to
 * continue , does not cancel the Subscription and returns { @ code null } .
 * Typical usage pattern differs depending on the calling method :
 * < ul >
 * < li > { @ code onNext } : check for a throwable return value and call
 * { @ link Subscriber # onError ( Throwable ) } if not null , otherwise perform a direct
 * { @ link Subscription # request ( long ) request ( 1 ) } on the upstream . < / li >
 * < li > { @ code tryOnNext } : check for a throwable return value and call
 * { @ link Subscriber # onError ( Throwable ) } if not null , otherwise
 * return { @ code false } to indicate value was not consumed and more must be
 * tried . < / li >
 * < li > { @ code poll } : use { @ link # onNextPollError ( Object , Throwable , Context ) } instead . < / li >
 * < / ul >
 * @ param value The onNext value that caused an error .
 * @ param error The error .
 * @ param context The most significant { @ link Context } in which to look for an { @ link OnNextFailureStrategy } .
 * @ param subscriptionForCancel The { @ link Subscription } that should be cancelled if the
 * strategy is terminal . Not null , use { @ link # onNextPollError ( Object , Throwable , Context ) }
 * when unsubscribing is undesirable ( eg . for poll ( ) ) .
 * @ param < T > The type of the value causing the error .
 * @ return a { @ link Throwable } to propagate through onError if the strategy is
 * terminal and cancelled the subscription , null if not . */
@ Nullable public static < T > Throwable onNextError ( @ Nullable T value , Throwable error , Context context , Subscription subscriptionForCancel ) { } }

implementation:
error = unwrapOnNextError ( error ) ; OnNextFailureStrategy strategy = onNextErrorStrategy ( context ) ; if ( strategy . test ( error , value ) ) { // some strategies could still return an exception , eg . if the consumer throws
Throwable t = strategy . process ( error , value , context ) ; if ( t != null ) { subscriptionForCancel . cancel ( ) ; } return t ; } else { // falls back to operator errors
return onOperatorError ( subscriptionForCancel , error , value , context ) ; }


signature:
public class Operators { /** * Find the { @ link OnNextFailureStrategy } to apply to the calling operator ( which could be a local
 * error mode defined in the { @ link Context } ) and apply it .
 * @ param error The error .
 * @ param context The most significant { @ link Context } in which to look for an { @ link OnNextFailureStrategy } .
 * @ param subscriptionForCancel The { @ link Subscription } that should be cancelled if the
 * strategy is terminal . Null to ignore ( for poll , use { @ link # onNextPollError ( Object , Throwable , Context ) }
 * rather than passing null ) .
 * @ param < T > The type of the value causing the error .
 * @ return a { @ link Throwable } to propagate through onError if the strategy is
 * terminal and cancelled the subscription , null if not . */
public static < T > Throwable onNextInnerError ( Throwable error , Context context , Subscription subscriptionForCancel ) { } }

implementation:
error = unwrapOnNextError ( error ) ; OnNextFailureStrategy strategy = onNextErrorStrategy ( context ) ; if ( strategy . test ( error , null ) ) { // some strategies could still return an exception , eg . if the consumer throws
Throwable t = strategy . process ( error , null , context ) ; if ( t != null ) { subscriptionForCancel . cancel ( ) ; } return t ; } else { return error ; }


signature:
public class Operators { /** * Find the { @ link OnNextFailureStrategy } to apply to the calling async operator ( which could be
 * a local error mode defined in the { @ link Context } ) and apply it .
 * Returns a { @ link RuntimeException } if errors are fatal for the error mode , in which
 * case the operator poll should throw the returned error . On the contrary if the
 * error mode allows the sequence to continue , returns { @ code null } in which case
 * the operator should retry the { @ link Queue # poll ( ) poll ( ) } .
 * @ param value The onNext value that caused an error .
 * @ param error The error .
 * @ param context The most significant { @ link Context } in which to look for an { @ link OnNextFailureStrategy } .
 * @ param < T > The type of the value causing the error .
 * @ return a { @ link Throwable } to propagate through onError if the strategy is
 * terminal and cancelled the subscription , null if not . */
@ Nullable public static < T > RuntimeException onNextPollError ( T value , Throwable error , Context context ) { } }

implementation:
error = unwrapOnNextError ( error ) ; OnNextFailureStrategy strategy = onNextErrorStrategy ( context ) ; if ( strategy . test ( error , value ) ) { // some strategies could still return an exception , eg . if the consumer throws
Throwable t = strategy . process ( error , value , context ) ; if ( t != null ) return Exceptions . propagate ( t ) ; return null ; } else { Throwable t = onOperatorError ( null , error , value , context ) ; return Exceptions . propagate ( t ) ; }


signature:
public class Operators { /** * Applies the hooks registered with { @ link Hooks # onLastOperator } and returns
 * { @ link CorePublisher } ready to be subscribed on .
 * @ param source the original { @ link CorePublisher } .
 * @ param < T > the type of the value .
 * @ return a { @ link CorePublisher } to subscribe on . */
@ SuppressWarnings ( "unchecked" ) public static < T > CorePublisher < T > onLastAssembly ( CorePublisher < T > source ) { } }

implementation: Function < Publisher , Publisher > hook = Hooks . onLastOperatorHook ; final Publisher < T > publisher ; if ( hook == null ) { publisher = source ; } else { publisher = Objects . requireNonNull ( hook . apply ( source ) , "LastOperator hook returned null" ) ; } if ( publisher instanceof CorePublisher ) { return ( CorePublisher < T > ) publisher ; } else { return new CorePublisherAdapter < > ( publisher ) ; }


signature:
public class Operators { /** * Return a wrapped { @ link RejectedExecutionException } which can be thrown by the
 * operator . This exception denotes that an execution was rejected by a
 * { @ link reactor . core . scheduler . Scheduler } , notably when it was already disposed .
 * Wrapping is done by calling both { @ link Exceptions # failWithRejected ( Throwable ) } and
 * { @ link # onOperatorError ( Subscription , Throwable , Object , Context ) } ( with the passed
 * { @ link Subscription } ) .
 * @ param original the original execution error
 * @ param subscription the subscription to pass to onOperatorError .
 * @ param suppressed a Throwable to be suppressed by the { @ link RejectedExecutionException } ( or null if not relevant )
 * @ param dataSignal a value to be passed to { @ link # onOperatorError ( Subscription , Throwable , Object , Context ) } ( or null if not relevant )
 * @ param context a context that might hold a local error consumer */
public static RuntimeException onRejectedExecution ( Throwable original , @ Nullable Subscription subscription , @ Nullable Throwable suppressed , @ Nullable Object dataSignal , Context context ) { } }

implementation:
// we " cheat " to apply the special key for onRejectedExecution in onOperatorError
if ( context . hasKey ( Hooks . KEY_ON_REJECTED_EXECUTION ) ) { context = context . put ( Hooks . KEY_ON_OPERATOR_ERROR , context . get ( Hooks . KEY_ON_REJECTED_EXECUTION ) ) ; } // don ' t create REE if original is a reactor - produced REE ( not including singletons )
RejectedExecutionException ree = Exceptions . failWithRejected ( original ) ; if ( suppressed != null ) { ree . addSuppressed ( suppressed ) ; } if ( dataSignal != null ) { return Exceptions . propagate ( Operators . onOperatorError ( subscription , ree , dataSignal , context ) ) ; } return Exceptions . propagate ( Operators . onOperatorError ( subscription , ree , context ) ) ;


signature:
public class Operators { /** * Concurrent subtraction bound to 0 , mostly used to decrement a request tracker by
 * the amount produced by the operator . Any concurrent write will " happen before "
 * this operation .
 * @ param < T > the parent instance type
 * @ param updater current field updater
 * @ param instance current instance to update
 * @ param toSub delta to subtract
 * @ return value after subtraction or zero */
public static < T > long produced ( AtomicLongFieldUpdater < T > updater , T instance , long toSub ) { } }

implementation: long r , u ; do { r = updater . get ( instance ) ; if ( r == 0 || r == Long . MAX_VALUE ) { return r ; } u = subOrZero ( r , toSub ) ; } while ( ! updater . compareAndSet ( instance , r , u ) ) ; return u ;


signature:
public class Operators { /** * A generic utility to atomically replace a subscription or cancel the replacement
 * if the current subscription is marked as already cancelled ( as in
 * { @ link # cancelledSubscription ( ) } ) .
 * @ param field The Atomic container
 * @ param instance the instance reference
 * @ param s the subscription
 * @ param < F > the instance type
 * @ return true if replaced */
public static < F > boolean replace ( AtomicReferenceFieldUpdater < F , Subscription > field , F instance , Subscription s ) { } }

implementation: for ( ; ; ) { Subscription a = field . get ( instance ) ; if ( a == CancelledSubscription . INSTANCE ) { s . cancel ( ) ; return false ; } if ( field . compareAndSet ( instance , a , s ) ) { return true ; } }


signature:
public class Operators { /** * Represents a fuseable Subscription that emits a single constant value synchronously
 * to a Subscriber or consumer .
 * @ param subscriber the delegate { @ link Subscriber } that will be requesting the value
 * @ param value the single value to be emitted
 * @ param < T > the value type
 * @ return a new scalar { @ link Subscription } */
public static < T > Subscription scalarSubscription ( CoreSubscriber < ? super T > subscriber , T value ) { } }

implementation: return new ScalarSubscription < > ( subscriber , value ) ;


signature:
public class Operators { /** * Sets the given subscription once and returns true if successful , false
 * if the field has a subscription already or has been cancelled .
 * If the field already has a subscription , it is cancelled and the duplicate
 * subscription is reported ( see { @ link # reportSubscriptionSet ( ) } ) .
 * @ param < F > the instance type containing the field
 * @ param field the field accessor
 * @ param instance the parent instance
 * @ param s the subscription to set once
 * @ return true if successful , false if the target was not empty or has been cancelled */
public static < F > boolean setOnce ( AtomicReferenceFieldUpdater < F , Subscription > field , F instance , Subscription s ) { } }

implementation: Objects . requireNonNull ( s , "subscription" ) ; Subscription a = field . get ( instance ) ; if ( a == CancelledSubscription . INSTANCE ) { s . cancel ( ) ; return false ; } if ( a != null ) { s . cancel ( ) ; reportSubscriptionSet ( ) ; return false ; } if ( field . compareAndSet ( instance , null , s ) ) { return true ; } a = field . get ( instance ) ; if ( a == CancelledSubscription . INSTANCE ) { s . cancel ( ) ; return false ; } s . cancel ( ) ; reportSubscriptionSet ( ) ; return false ;


signature:
public class Operators { /** * Atomically terminates the subscription if it is not already a
 * { @ link # cancelledSubscription ( ) } , cancelling the subscription and setting the field
 * to the singleton { @ link # cancelledSubscription ( ) } .
 * @ param < F > the instance type containing the field
 * @ param field the field accessor
 * @ param instance the parent instance
 * @ return true if terminated , false if the subscription was already terminated */
public static < F > boolean terminate ( AtomicReferenceFieldUpdater < F , Subscription > field , F instance ) { } }

implementation: Subscription a = field . get ( instance ) ; if ( a != CancelledSubscription . INSTANCE ) { a = field . getAndSet ( instance , CancelledSubscription . INSTANCE ) ; if ( a != null && a != CancelledSubscription . INSTANCE ) { a . cancel ( ) ; return true ; } } return false ;


signature:
public class Operators { /** * Check Subscription current state and cancel new Subscription if current is set ,
 * or return true if ready to subscribe .
 * @ param current current Subscription , expected to be null
 * @ param next new Subscription
 * @ return true if Subscription can be used */
public static boolean validate ( @ Nullable Subscription current , Subscription next ) { } }

implementation:
Objects . requireNonNull ( next , "Subscription cannot be null" ) ; if ( current != null ) { next . cancel ( ) ; // reportSubscriptionSet ( ) ;
return false ; } return true ;


signature:
public class Operators { /** * If the actual { @ link Subscriber } is not a { @ link CoreSubscriber } , it will apply
 * safe strict wrapping to apply all reactive streams rules including the ones
 * relaxed by internal operators based on { @ link CoreSubscriber } .
 * @ param < T > passed subscriber type
 * @ param actual the { @ link Subscriber } to apply hook on
 * @ return an eventually transformed { @ link Subscriber } */
@ SuppressWarnings ( "unchecked" ) public static < T > CoreSubscriber < ? super T > toCoreSubscriber ( Subscriber < ? super T > actual ) { } }

implementation: Objects . requireNonNull ( actual , "actual" ) ; CoreSubscriber < ? super T > _actual ; if ( actual instanceof CoreSubscriber ) { _actual = ( CoreSubscriber < ? super T > ) actual ; } else { _actual = new StrictSubscriber < > ( actual ) ; } return _actual ;


signature:
public class Operators { /** * If the actual { @ link CoreSubscriber } is not { @ link Fuseable . ConditionalSubscriber } ,
 * it will apply an adapter which directly maps all
 * { @ link Fuseable . ConditionalSubscriber # tryOnNext ( T ) } to { @ link CoreSubscriber # onNext ( T ) }
 * and always returns true as the result
 * @ param < T > passed subscriber type
 * @ param actual the { @ link Subscriber } to adapt
 * @ return a potentially adapted { @ link Fuseable . ConditionalSubscriber } */
@ SuppressWarnings ( "unchecked" ) public static < T > Fuseable . ConditionalSubscriber < ? super T > toConditionalSubscriber ( CoreSubscriber < ? super T > actual ) { } }

implementation: Objects . requireNonNull ( actual , "actual" ) ; Fuseable . ConditionalSubscriber < ? super T > _actual ; if ( actual instanceof Fuseable . ConditionalSubscriber ) { _actual = ( Fuseable . ConditionalSubscriber < ? super T > ) actual ; } else { _actual = new ConditionalSubscriberAdapter < > ( actual ) ; } return _actual ;


signature:
public class Operators { /** * Add the amount { @ code n } to the given field , capped to { @ link Long # MAX _ VALUE } ,
 * unless the field is already at { @ link Long # MAX _ VALUE } OR { @ link Long # MIN _ VALUE } .
 * Return the value before the update .
 * @ param updater the field to update
 * @ param instance the instance bearing the field
 * @ param n the value to add
 * @ param < T > the type of the field - bearing instance
 * @ return the old value of the field , before update . */
static < T > long addCapCancellable ( AtomicLongFieldUpdater < T > updater , T instance , long n ) { } }

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


signature:
public class Loggers { /** * Force the usage of Console - based { @ link Logger Loggers } , even if SLF4J is available
 * on the classpath . Console loggers will output { @ link Logger # error ( String ) ERROR } and
 * { @ link Logger # warn ( String ) WARN } levels to { @ link System # err } and levels below to
 * { @ link System # out } . All levels < strong > except TRACE and DEBUG < / strong > are
 * considered enabled . TRACE and DEBUG are omitted .
 * The previously active logger factory is simply replaced without
 * any particular clean - up . */
public static final void useConsoleLoggers ( ) { } }

implementation: String name = LoggerFactory . class . getName ( ) ; LoggerFactory loggerFactory = new ConsoleLoggerFactory ( false ) ; loggerFactory . getLogger ( name ) . debug ( "Using Console logging" ) ; LOGGER_FACTORY = loggerFactory ;


signature:
public class Loggers { /** * Force the usage of Console - based { @ link Logger Loggers } , even if SLF4J is available
 * on the classpath . Console loggers will output { @ link Logger # error ( String ) ERROR } and
 * { @ link Logger # warn ( String ) WARN } levels to { @ link System # err } and levels below to
 * { @ link System # out } . All levels ( including TRACE and DEBUG ) are considered enabled .
 * The previously active logger factory is simply replaced without
 * any particular clean - up . */
public static final void useVerboseConsoleLoggers ( ) { } }

implementation: String name = LoggerFactory . class . getName ( ) ; LoggerFactory loggerFactory = new ConsoleLoggerFactory ( true ) ; loggerFactory . getLogger ( name ) . debug ( "Using Verbose Console logging" ) ; LOGGER_FACTORY = loggerFactory ;


signature:
public class Loggers { /** * Use a custom type of { @ link Logger } created through the provided { @ link Function } ,
 * which takes a logger name as input .
 * The previously active logger factory is simply replaced without
 * any particular clean - up .
 * @ param loggerFactory the { @ link Function } that provides a ( possibly cached ) { @ link Logger }
 * given a name . */
public static final void useCustomLoggers ( final Function < String , ? extends Logger > loggerFactory ) { } }

implementation: String name = LoggerFactory . class . getName ( ) ; loggerFactory . apply ( name ) . debug ( "Using custom logging" ) ; LOGGER_FACTORY = loggerFactory :: apply ;


signature:
public class Loggers { /** * Force the usage of JDK - based { @ link Logger Loggers } , even if SLF4J is available
 * on the classpath .
 * The previously active logger factory is simply replaced without
 * any particular clean - up . */
public static final void useJdkLoggers ( ) { } }

implementation: String name = LoggerFactory . class . getName ( ) ; LoggerFactory loggerFactory = new JdkLoggerFactory ( ) ; loggerFactory . getLogger ( name ) . debug ( "Using JDK logging framework" ) ; LOGGER_FACTORY = loggerFactory ;


signature:
public class Loggers { /** * Force the usage of SL4J - based { @ link Logger Loggers } , throwing an exception if
 * SLF4J isn ' t available on the classpath . Prefer using { @ link # resetLoggerFactory ( ) }
 * as it will fallback in the later case .
 * The previously active logger factory is simply replaced without
 * any particular clean - up . */
public static final void useSl4jLoggers ( ) { } }

implementation: String name = LoggerFactory . class . getName ( ) ; LoggerFactory loggerFactory = new Slf4JLoggerFactory ( ) ; loggerFactory . getLogger ( name ) . debug ( "Using Slf4j logging framework" ) ; LOGGER_FACTORY = loggerFactory ;


signature:
public class MonoIgnoreThen { /** * Shifts the current last Mono into the ignore array and sets up a new last Mono instance .
 * @ param < U > the new last value type
 * @ param newLast the new last Mono instance
 * @ return the new operator set up */
< U > MonoIgnoreThen < U > shift ( Mono < U > newLast ) { } }

implementation: Objects . requireNonNull ( newLast , "newLast" ) ; Publisher < ? > [ ] a = ignore ; int n = a . length ; Publisher < ? > [ ] b = new Publisher [ n + 1 ] ; System . arraycopy ( a , 0 , b , 0 , n ) ; b [ n ] = last ; return new MonoIgnoreThen < > ( b , newLast ) ;


signature:
public class WaitStrategy { /** * < p > Phased wait strategy for waiting consumers on a barrier . < / p >
 * < p > This strategy can be used when throughput and low - latency are not as important as CPU resource . Spins , then
 * yields , then waits using the configured fallback WaitStrategy . < / p >
 * @ param spinTimeout the spin timeout
 * @ param yieldTimeout the yield timeout
 * @ param units the time unit
 * @ param delegate the target wait strategy to fallback on
 * @ return the wait strategy */
public static WaitStrategy phasedOff ( long spinTimeout , long yieldTimeout , TimeUnit units , WaitStrategy delegate ) { } }

implementation: return new PhasedOff ( spinTimeout , yieldTimeout , units , delegate ) ;


signature:
public class WaitStrategy { /** * Block with wait / notifyAll semantics
 * @ param spinTimeout the spin timeout
 * @ param yieldTimeout the yield timeout
 * @ param units the time unit
 * @ return the wait strategy */
public static WaitStrategy phasedOffLiteLock ( long spinTimeout , long yieldTimeout , TimeUnit units ) { } }

implementation: return phasedOff ( spinTimeout , yieldTimeout , units , liteBlocking ( ) ) ;


signature:
public class WaitStrategy { /** * Block with wait / notifyAll semantics
 * @ param spinTimeout the spin timeout
 * @ param yieldTimeout the yield timeout
 * @ param units the time unit
 * @ return the wait strategy */
public static WaitStrategy phasedOffLock ( long spinTimeout , long yieldTimeout , TimeUnit units ) { } }

implementation: return phasedOff ( spinTimeout , yieldTimeout , units , blocking ( ) ) ;


signature:
public class WaitStrategy { /** * Block by parking in a loop
 * @ param spinTimeout the spin timeout
 * @ param yieldTimeout the yield timeout
 * @ param units the time unit
 * @ return the wait strategy */
public static WaitStrategy phasedOffSleep ( long spinTimeout , long yieldTimeout , TimeUnit units ) { } }

implementation: return phasedOff ( spinTimeout , yieldTimeout , units , parking ( 0 ) ) ;


signature:
public class FluxFirstEmitting { /** * Returns a new instance which has the additional source to be amb ' d together with
 * the current array of sources .
 * This operation doesn ' t change the current FluxFirstEmitting instance .
 * @ param source the new source to merge with the others
 * @ return the new FluxFirstEmitting instance or null if the Amb runs with an Iterable */
@ Nullable FluxFirstEmitting < T > ambAdditionalSource ( Publisher < ? extends T > source ) { } }

implementation: if ( array != null ) { int n = array . length ; @ SuppressWarnings ( "unchecked" ) Publisher < ? extends T > [ ] newArray = new Publisher [ n + 1 ] ; System . arraycopy ( array , 0 , newArray , 0 , n ) ; newArray [ n ] = source ; return new FluxFirstEmitting < > ( newArray ) ; } return null ;


signature:
public class FluxConcatArray { /** * Returns a new instance which has the additional source to be merged together with
 * the current array of sources .
 * This operation doesn ' t change the current FluxMerge instance .
 * @ param source the new source to merge with the others
 * @ return the new FluxConcatArray instance */
FluxConcatArray < T > concatAdditionalSourceLast ( Publisher < ? extends T > source ) { } }

implementation: int n = array . length ; @ SuppressWarnings ( "unchecked" ) Publisher < ? extends T > [ ] newArray = new Publisher [ n + 1 ] ; System . arraycopy ( array , 0 , newArray , 0 , n ) ; newArray [ n ] = source ; return new FluxConcatArray < > ( delayError , newArray ) ;


signature:
public class FluxConcatArray { /** * Returns a new instance which has the additional source to be merged together with
 * the current array of sources .
 * This operation doesn ' t change the current FluxMerge instance .
 * @ param source the new source to merge with the others
 * @ return the new FluxConcatArray instance */
@ SuppressWarnings ( "unchecked" ) < V > FluxConcatArray < V > concatAdditionalIgnoredLast ( Publisher < ? extends V > source ) { } }

implementation:
int n = array . length ; Publisher < ? extends V > [ ] newArray = new Publisher [ n + 1 ] ; // noinspection SuspiciousSystemArraycopy
System . arraycopy ( array , 0 , newArray , 0 , n ) ; newArray [ n - 1 ] = Mono . ignoreElements ( newArray [ n - 1 ] ) ; newArray [ n ] = source ; return new FluxConcatArray < > ( delayError , newArray ) ;


signature:
public class ConnectableFlux { /** * Connects this { @ link ConnectableFlux } to the upstream source when the specified amount of
 * { @ link org . reactivestreams . Subscriber } subscribes and calls the supplied consumer with a runnable that allows disconnecting .
 * < img class = " marble " src = " doc - files / marbles / autoConnectWithMinSubscribers . svg " alt = " " >
 * @ param minSubscribers the minimum number of subscribers
 * @ param cancelSupport the consumer that will receive the { @ link Disposable } that allows disconnecting
 * @ return a { @ link Flux } that connects to the upstream source when the given amount of subscribers subscribed */
public final Flux < T > autoConnect ( int minSubscribers , Consumer < ? super Disposable > cancelSupport ) { } }

implementation: if ( minSubscribers == 0 ) { connect ( cancelSupport ) ; return this ; } if ( this instanceof Fuseable ) { return onAssembly ( new FluxAutoConnectFuseable < > ( this , minSubscribers , cancelSupport ) ) ; } return onAssembly ( new FluxAutoConnect < > ( this , minSubscribers , cancelSupport ) ) ;


signature:
public class ConnectableFlux { /** * Connect this { @ link ConnectableFlux } to its source and return a { @ link Runnable } that
 * can be used for disconnecting .
 * @ return the { @ link Disposable } that allows disconnecting the connection after . */
public final Disposable connect ( ) { } }

implementation: final Disposable [ ] out = { null } ; connect ( r -> out [ 0 ] = r ) ; return out [ 0 ] ;


signature:
public class ConnectableFlux { /** * Connects to the upstream source when the given number of { @ link org . reactivestreams . Subscriber } subscribes .
 * Disconnection can happen in two scenarios : when the upstream source completes ( or errors ) then
 * there is an immediate disconnection . However , when all subscribers have cancelled ,
 * a < strong > deferred < / strong > disconnection is scheduled . If any new subscriber comes
 * in during the { @ code gracePeriod } that follows , the disconnection is cancelled .
 * < img class = " marble " src = " doc - files / marbles / refCountWithMinSubscribersAndGracePeriod . svg " alt = " " >
 * @ param minSubscribers the number of subscribers expected to subscribe before connection
 * @ param gracePeriod the { @ link Duration } for which to wait for new subscribers before actually
 * disconnecting when all subscribers have cancelled .
 * @ return a reference counting { @ link Flux } with a grace period for disconnection */
public final Flux < T > refCount ( int minSubscribers , Duration gracePeriod ) { } }

implementation: return refCount ( minSubscribers , gracePeriod , Schedulers . parallel ( ) ) ;


signature:
public class ConnectableFlux { /** * Connects to the upstream source when the given number of { @ link org . reactivestreams . Subscriber } subscribes .
 * Disconnection can happen in two scenarios : when the upstream source completes ( or errors ) then
 * there is an immediate disconnection . However , when all subscribers have cancelled ,
 * a < strong > deferred < / strong > disconnection is scheduled . If any new subscriber comes
 * in during the { @ code gracePeriod } that follows , the disconnection is cancelled .
 * < img class = " marble " src = " doc - files / marbles / refCountWithMinSubscribersAndGracePeriod . svg " alt = " " >
 * @ param minSubscribers the number of subscribers expected to subscribe before connection
 * @ param gracePeriod the { @ link Duration } for which to wait for new subscribers before actually
 * disconnecting when all subscribers have cancelled .
 * @ param scheduler the { @ link Scheduler } on which to run timeouts
 * @ return a reference counting { @ link Flux } with a grace period for disconnection */
public final Flux < T > refCount ( int minSubscribers , Duration gracePeriod , Scheduler scheduler ) { } }

implementation: return onAssembly ( new FluxRefCountGrace < > ( this , minSubscribers , gracePeriod , scheduler ) ) ;


signature:
public class Mono { /** * Creates a deferred emitter that can be used with callback - based
 * APIs to signal at most one value , a complete or an error signal .
 * < img class = " marble " src = " doc - files / marbles / createForMono . svg " alt = " " >
 * Bridging legacy API involves mostly boilerplate code due to the lack
 * of standard types and methods . There are two kinds of API surfaces :
 * 1 ) addListener / removeListener and 2 ) callback - handler .
 * < b > 1 ) addListener / removeListener pairs < / b > < br >
 * To work with such API one has to instantiate the listener ,
 * call the sink from the listener then register it with the source :
 * < pre > < code >
 * Mono . & lt ; String & gt ; create ( sink - & gt ; {
 * HttpListener listener = event - & gt ; {
 * if ( event . getResponseCode ( ) > = 400 ) {
 * sink . error ( new RuntimeException ( " Failed " ) ) ;
 * } else {
 * String body = event . getBody ( ) ;
 * if ( body . isEmpty ( ) ) {
 * sink . success ( ) ;
 * } else {
 * sink . success ( body . toLowerCase ( ) ) ;
 * client . addListener ( listener ) ;
 * sink . onDispose ( ( ) - & gt ; client . removeListener ( listener ) ) ;
 * < / code > < / pre >
 * Note that this works only with single - value emitting listeners . Otherwise ,
 * all subsequent signals are dropped . You may have to add { @ code client . removeListener ( this ) ; }
 * to the listener ' s body .
 * < b > 2 ) callback handler < / b > < br >
 * This requires a similar instantiation pattern such as above , but usually the
 * successful completion and error are separated into different methods .
 * In addition , the legacy API may or may not support some cancellation mechanism .
 * < pre > < code >
 * Mono . & lt ; String & gt ; create ( sink - & gt ; {
 * Callback & lt ; String & gt ; callback = new Callback & lt ; String & gt ; ( ) {
 * & # 64 ; Override
 * public void onResult ( String data ) {
 * sink . success ( data . toLowerCase ( ) ) ;
 * & # 64 ; Override
 * public void onError ( Exception e ) {
 * sink . error ( e ) ;
 * / / without cancellation support :
 * client . call ( " query " , callback ) ;
 * / / with cancellation support :
 * AutoCloseable cancel = client . call ( " query " , callback ) ;
 * sink . onDispose ( ( ) - > {
 * try {
 * cancel . close ( ) ;
 * } catch ( Exception ex ) {
 * Exceptions . onErrorDropped ( ex ) ;
 * < / code > < / pre >
 * @ param callback Consume the { @ link MonoSink } provided per - subscriber by Reactor to generate signals .
 * @ param < T > The type of the value emitted
 * @ return a { @ link Mono } */
public static < T > Mono < T > create ( Consumer < MonoSink < T > > callback ) { } }

implementation: return onAssembly ( new MonoCreate < > ( callback ) ) ;


signature:
public class Mono { /** * Create a { @ link Mono } provider that will { @ link Supplier # get supply } a target { @ link Mono } to subscribe to for
 * each { @ link Subscriber } downstream .
 * < img class = " marble " src = " doc - files / marbles / deferForMono . svg " alt = " " >
 * @ param supplier a { @ link Mono } factory
 * @ param < T > the element type of the returned Mono instance
 * @ return a new { @ link Mono } factory */
public static < T > Mono < T > defer ( Supplier < ? extends Mono < ? extends T > > supplier ) { } }

implementation: return onAssembly ( new MonoDefer < > ( supplier ) ) ;


signature:
public class Mono { /** * Create a Mono which delays an onNext signal by a given { @ link Duration duration }
 * on a provided { @ link Scheduler } and completes .
 * If the demand cannot be produced in time , an onError will be signalled instead .
 * < img class = " marble " src = " doc - files / marbles / delay . svg " alt = " " >
 * @ param duration the { @ link Duration } of the delay
 * @ param timer a time - capable { @ link Scheduler } instance to run on
 * @ return a new { @ link Mono } */
public static Mono < Long > delay ( Duration duration , Scheduler timer ) { } }

implementation: return onAssembly ( new MonoDelay ( duration . toMillis ( ) , TimeUnit . MILLISECONDS , timer ) ) ;


signature:
public class Mono { /** * Create a { @ link Mono } that terminates with an error immediately after being
 * subscribed to . The { @ link Throwable } is generated by a { @ link Supplier } , invoked
 * each time there is a subscription and allowing for lazy instantiation .
 * < img class = " marble " src = " doc - files / marbles / errorWithSupplier . svg " alt = " " >
 * @ param errorSupplier the error signal { @ link Supplier } to invoke for each { @ link Subscriber }
 * @ param < T > the reified { @ link Subscriber } type
 * @ return a failing { @ link Mono } */
public static < T > Mono < T > error ( Supplier < Throwable > errorSupplier ) { } }

implementation: return onAssembly ( new MonoErrorSupplied < > ( errorSupplier ) ) ;


signature:
public class Mono { /** * Pick the first { @ link Mono } to emit any signal ( value , empty completion or error )
 * and replay that signal , effectively behaving like the fastest of these competing
 * sources .
 * < img class = " marble " src = " doc - files / marbles / firstForMono . svg " alt = " " >
 * @ param monos The deferred monos to use .
 * @ param < T > The type of the function result .
 * @ return a new { @ link Mono } behaving like the fastest of its sources . */
@ SafeVarargs public static < T > Mono < T > first ( Mono < ? extends T > ... monos ) { } }

implementation: return onAssembly ( new MonoFirst < > ( monos ) ) ;


signature:
public class Mono { /** * Pick the first available result coming from any of the given monos and populate a new { @ literal Mono } .
 * < img class = " marble " src = " doc - files / marbles / firstForMono . svg " alt = " " >
 * @ param monos The monos to use .
 * @ param < T > The type of the function result .
 * @ return a { @ link Mono } . */
public static < T > Mono < T > first ( Iterable < ? extends Mono < ? extends T > > monos ) { } }


signature:
public class Mono { /** * Expose the specified { @ link Publisher } with the { @ link Mono } API , and ensure it will emit 0 or 1 item .
 * The source emitter will be cancelled on the first ` onNext ` .
 * < img class = " marble " src = " doc - files / marbles / fromForMono . svg " alt = " " >
 * @ param source the { @ link Publisher } source
 * @ param < T > the source type
 * @ return the next item emitted as a { @ link Mono } */
public static < T > Mono < T > from ( Publisher < ? extends T > source ) { } }

implementation: if ( source instanceof Mono ) { @ SuppressWarnings ( "unchecked" ) Mono < T > casted = ( Mono < T > ) source ; return casted ; } if ( source instanceof FluxSourceMono || source instanceof FluxSourceMonoFuseable ) { @ SuppressWarnings ( "unchecked" ) FluxFromMonoOperator < T , T > wrapper = ( FluxFromMonoOperator < T , T > ) source ; @ SuppressWarnings ( "unchecked" ) Mono < T > extracted = ( Mono < T > ) wrapper . source ; return extracted ; } if ( source instanceof Flux ) { @ SuppressWarnings ( "unchecked" ) Flux < T > casted = ( Flux < T > ) source ; return casted . next ( ) ; } return onAssembly ( new MonoFromPublisher < > ( source ) ) ;


signature:
public class Mono { /** * Create a { @ link Mono } producing its value using the provided { @ link Callable } . If
 * the Callable resolves to { @ code null } , the resulting Mono completes empty .
 * < img class = " marble " src = " doc - files / marbles / fromCallable . svg " alt = " " >
 * @ param supplier { @ link Callable } that will produce the value
 * @ param < T > type of the expected value
 * @ return A { @ link Mono } . */
public static < T > Mono < T > fromCallable ( Callable < ? extends T > supplier ) { } }

implementation: return onAssembly ( new MonoCallable < > ( supplier ) ) ;


signature:
public class Mono { /** * Create a { @ link Mono } , producing its value using the provided { @ link CompletionStage } .
 * < img class = " marble " src = " doc - files / marbles / fromFuture . svg " alt = " " >
 * Note that the completion stage is not cancelled when that Mono is cancelled , but
 * that behavior can be obtained by using { @ link # doFinally ( Consumer ) } that checks
 * for a { @ link SignalType # CANCEL } and calls eg .
 * { @ link CompletionStage # toCompletableFuture ( ) . toCompletableFuture ( ) . cancel ( false ) } .
 * @ param completionStage { @ link CompletionStage } that will produce a value ( or a null to
 * complete immediately )
 * @ param < T > type of the expected value
 * @ return A { @ link Mono } . */
public static < T > Mono < T > fromCompletionStage ( CompletionStage < ? extends T > completionStage ) { } }

implementation: return onAssembly ( new MonoCompletionStage < > ( completionStage ) ) ;


signature:
public class Mono { /** * Create a { @ link Mono } that wraps a { @ link CompletionStage } on subscription ,
 * emitting the value produced by the { @ link CompletionStage } .
 * < img class = " marble " src = " doc - files / marbles / fromFutureSupplier . svg " alt = " " >
 * Note that the completion stage is not cancelled when that Mono is cancelled , but
 * that behavior can be obtained by using { @ link # doFinally ( Consumer ) } that checks
 * for a { @ link SignalType # CANCEL } and calls eg .
 * { @ link CompletionStage # toCompletableFuture ( ) . toCompletableFuture ( ) . cancel ( false ) } .
 * @ param stageSupplier The { @ link Supplier } of a { @ link CompletionStage } that will produce a value ( or a null to
 * complete immediately ) . This allows lazy triggering of CompletionStage - based APIs .
 * @ param < T > type of the expected value
 * @ return A { @ link Mono } . */
public static < T > Mono < T > fromCompletionStage ( Supplier < ? extends CompletionStage < ? extends T > > stageSupplier ) { } }

implementation: return defer ( ( ) -> onAssembly ( new MonoCompletionStage < > ( stageSupplier . get ( ) ) ) ) ;


signature:
public class Mono { /** * Convert a { @ link Publisher } to a { @ link Mono } without any cardinality check
 * ( ie this method doesn ' t check if the source is already a Mono , nor cancels the
 * source past the first element ) . Conversion supports { @ link Fuseable } sources .
 * Note this is an advanced interoperability operator that implies you know the
 * { @ link Publisher } you are converting follows the { @ link Mono } semantics and only
 * ever emits one element .
 * @ param source the Mono - compatible { @ link Publisher } to wrap
 * @ param < I > type of the value emitted by the publisher
 * @ return a wrapped { @ link Mono } */
public static < I > Mono < I > fromDirect ( Publisher < ? extends I > source ) { } }

implementation: if ( source instanceof Mono ) { @ SuppressWarnings ( "unchecked" ) Mono < I > m = ( Mono < I > ) source ; return m ; } if ( source instanceof Flux ) { @ SuppressWarnings ( "unchecked" ) Flux < I > f = ( Flux < I > ) source ; if ( source instanceof Fuseable ) { return onAssembly ( new MonoSourceFluxFuseable < > ( f ) ) ; } return onAssembly ( new MonoSourceFlux < > ( f ) ) ; } if ( source instanceof Fuseable ) { return onAssembly ( new MonoSourceFuseable < > ( source ) ) ; } return onAssembly ( new MonoSource < > ( source ) ) ;


signature:
public class Mono { /** * Create a { @ link Mono } , producing its value using the provided { @ link CompletableFuture } .
 * < img class = " marble " src = " doc - files / marbles / fromFuture . svg " alt = " " >
 * Note that the future is not cancelled when that Mono is cancelled , but that behavior
 * can be obtained by using a { @ link # doFinally ( Consumer ) } that checks
 * for a { @ link SignalType # CANCEL } and calls { @ link CompletableFuture # cancel ( boolean ) } .
 * @ param future { @ link CompletableFuture } that will produce a value ( or a null to
 * complete immediately )
 * @ param < T > type of the expected value
 * @ return A { @ link Mono } .
 * @ see # fromCompletionStage ( CompletionStage ) fromCompletionStage for a generalization */
public static < T > Mono < T > fromFuture ( CompletableFuture < ? extends T > future ) { } }

implementation: return onAssembly ( new MonoCompletionStage < > ( future ) ) ;


signature:
public class Mono { /** * Create a { @ link Mono } that wraps a { @ link CompletableFuture } on subscription ,
 * emitting the value produced by the Future .
 * < img class = " marble " src = " doc - files / marbles / fromFutureSupplier . svg " alt = " " >
 * Note that the future is not cancelled when that Mono is cancelled , but that behavior
 * can be obtained by using a { @ link # doFinally ( Consumer ) } that checks
 * for a { @ link SignalType # CANCEL } and calls { @ link CompletableFuture # cancel ( boolean ) } .
 * @ param futureSupplier The { @ link Supplier } of a { @ link CompletableFuture } that will produce a value ( or a null to
 * complete immediately ) . This allows lazy triggering of future - based APIs .
 * @ param < T > type of the expected value
 * @ return A { @ link Mono } .
 * @ see # fromCompletionStage ( Supplier ) fromCompletionStage for a generalization */
public static < T > Mono < T > fromFuture ( Supplier < ? extends CompletableFuture < ? extends T > > futureSupplier ) { } }

implementation: return defer ( ( ) -> onAssembly ( new MonoCompletionStage < > ( futureSupplier . get ( ) ) ) ) ;


signature:
public class Mono { /** * Create a { @ link Mono } , producing its value using the provided { @ link Supplier } . If
 * the Supplier resolves to { @ code null } , the resulting Mono completes empty .
 * < img class = " marble " src = " doc - files / marbles / fromSupplier . svg " alt = " " >
 * @ param supplier { @ link Supplier } that will produce the value
 * @ param < T > type of the expected value
 * @ return A { @ link Mono } . */
public static < T > Mono < T > fromSupplier ( Supplier < ? extends T > supplier ) { } }

implementation: return onAssembly ( new MonoSupplier < > ( supplier ) ) ;


signature:
public class Mono { /** * Create a new { @ link Mono } that ignores elements from the source ( dropping them ) ,
 * but completes when the source completes .
 * < img class = " marble " src = " doc - files / marbles / ignoreElementsForMono . svg " alt = " " >
 * @ reactor . discard This operator discards the element from the source .
 * @ param source the { @ link Publisher } to ignore
 * @ param < T > the source type of the ignored data
 * @ return a new completable { @ link Mono } . */
public static < T > Mono < T > ignoreElements ( Publisher < T > source ) { } }

implementation: return onAssembly ( new MonoIgnorePublisher < > ( source ) ) ;


signature:
public class Mono { /** * Create a new { @ link Mono } that emits the specified item if { @ link Optional # isPresent ( ) } otherwise only emits
 * onComplete .
 * < img class = " marble " src = " doc - files / marbles / justOrEmpty . svg " alt = " " >
 * @ param data the { @ link Optional } item to onNext or onComplete if not present
 * @ param < T > the type of the produced item
 * @ return a { @ link Mono } . */
public static < T > Mono < T > justOrEmpty ( @ Nullable Optional < ? extends T > data ) { } }

implementation: return data != null && data . isPresent ( ) ? just ( data . get ( ) ) : empty ( ) ;


signature:
public class Mono { /** * Create a new { @ link Mono } that emits the specified item if non null otherwise only emits
 * onComplete .
 * < img class = " marble " src = " doc - files / marbles / justOrEmpty . svg " alt = " " >
 * @ param data the item to onNext or onComplete if null
 * @ param < T > the type of the produced item
 * @ return a { @ link Mono } . */
public static < T > Mono < T > justOrEmpty ( @ Nullable T data ) { } }

implementation: return data != null ? just ( data ) : empty ( ) ;


signature:
public class Mono { /** * Returns a Mono that emits a Boolean value that indicates whether two Publisher sequences are the
 * same by comparing the items emitted by each Publisher pairwise .
 * < img class = " marble " src = " doc - files / marbles / sequenceEqual . svg " alt = " " >
 * @ param source1 the first Publisher to compare
 * @ param source2 the second Publisher to compare
 * @ param < T > the type of items emitted by each Publisher
 * @ return a Mono that emits a Boolean value that indicates whether the two sequences are the same */
public static < T > Mono < Boolean > sequenceEqual ( Publisher < ? extends T > source1 , Publisher < ? extends T > source2 ) { } }

implementation: return sequenceEqual ( source1 , source2 , equalsBiPredicate ( ) , Queues . SMALL_BUFFER_SIZE ) ;


signature:
public class Mono { /** * Returns a Mono that emits a Boolean value that indicates whether two Publisher sequences are the
 * same by comparing the items emitted by each Publisher pairwise based on the results of a specified
 * equality function .
 * < img class = " marble " src = " doc - files / marbles / sequenceEqual . svg " alt = " " >
 * @ param source1 the first Publisher to compare
 * @ param source2 the second Publisher to compare
 * @ param isEqual a function used to compare items emitted by each Publisher
 * @ param < T > the type of items emitted by each Publisher
 * @ return a Mono that emits a Boolean value that indicates whether the two Publisher two sequences
 * are the same according to the specified function */
public static < T > Mono < Boolean > sequenceEqual ( Publisher < ? extends T > source1 , Publisher < ? extends T > source2 , BiPredicate < ? super T , ? super T > isEqual ) { } }

implementation: return sequenceEqual ( source1 , source2 , isEqual , Queues . SMALL_BUFFER_SIZE ) ;


signature:
public class Mono { /** * Returns a Mono that emits a Boolean value that indicates whether two Publisher sequences are the
 * same by comparing the items emitted by each Publisher pairwise based on the results of a specified
 * equality function .
 * < img class = " marble " src = " doc - files / marbles / sequenceEqual . svg " alt = " " >
 * @ param source1 the first Publisher to compare
 * @ param source2 the second Publisher to compare
 * @ param isEqual a function used to compare items emitted by each Publisher
 * @ param prefetch the number of items to prefetch from the first and second source Publisher
 * @ param < T > the type of items emitted by each Publisher
 * @ return a Mono that emits a Boolean value that indicates whether the two Publisher two sequences
 * are the same according to the specified function */
public static < T > Mono < Boolean > sequenceEqual ( Publisher < ? extends T > source1 , Publisher < ? extends T > source2 , BiPredicate < ? super T , ? super T > isEqual , int prefetch ) { } }

implementation: return onAssembly ( new MonoSequenceEqual < > ( source1 , source2 , isEqual , prefetch ) ) ;


signature:
public class Mono { /** * Uses a resource , generated by a supplier for each individual Subscriber , while streaming the value from a
 * Mono derived from the same resource and makes sure the resource is released if the
 * sequence terminates or the Subscriber cancels .
 * < ul > < li > Eager resource cleanup happens just before the source termination and exceptions raised by the cleanup
 * Consumer may override the terminal event . < / li > < li > Non - eager cleanup will drop any exception . < / li > < / ul >
 * < img class = " marble " src = " doc - files / marbles / usingForMono . svg " alt = " " >
 * @ param resourceSupplier a { @ link Callable } that is called on subscribe to create the resource
 * @ param sourceSupplier a { @ link Mono } factory to create the Mono depending on the created resource
 * @ param resourceCleanup invoked on completion to clean - up the resource
 * @ param eager set to true to clean before terminating downstream subscribers
 * @ param < T > emitted type
 * @ param < D > resource type
 * @ return new { @ link Mono } */
public static < T , D > Mono < T > using ( Callable < ? extends D > resourceSupplier , Function < ? super D , ? extends Mono < ? extends T > > sourceSupplier , Consumer < ? super D > resourceCleanup , boolean eager ) { } }

implementation: return onAssembly ( new MonoUsing < > ( resourceSupplier , sourceSupplier , resourceCleanup , eager ) ) ;


signature:
public class Mono { /** * Uses a resource , generated by a supplier for each individual Subscriber , while streaming the value from a
 * Mono derived from the same resource and makes sure the resource is released if the
 * sequence terminates or the Subscriber cancels .
 * Eager resource cleanup happens just before the source termination and exceptions raised by the cleanup Consumer
 * may override the terminal event .
 * < img class = " marble " src = " doc - files / marbles / usingForMono . svg " alt = " " >
 * @ param resourceSupplier a { @ link Callable } that is called on subscribe to create the resource
 * @ param sourceSupplier a { @ link Mono } factory to create the Mono depending on the created resource
 * @ param resourceCleanup invoked on completion to clean - up the resource
 * @ param < T > emitted type
 * @ param < D > resource type
 * @ return new { @ link Mono } */
public static < T , D > Mono < T > using ( Callable < ? extends D > resourceSupplier , Function < ? super D , ? extends Mono < ? extends T > > sourceSupplier , Consumer < ? super D > resourceCleanup ) { } }

implementation: return using ( resourceSupplier , sourceSupplier , resourceCleanup , true ) ;


signature:
public class Mono { /** * Uses a resource , generated by a { @ link Publisher } for each individual { @ link Subscriber } ,
 * while streaming the values from a { @ link Publisher } derived from the same resource .
 * Whenever the resulting sequence terminates , the relevant { @ link Function } generates
 * a " cleanup " { @ link Publisher } that is invoked but doesn ' t change the content of the
 * main sequence . Instead it just defers the termination ( unless it errors , in which case
 * the error suppresses the original termination signal ) .
 * < img class = " marble " src = " doc - files / marbles / usingWhenSuccessForMono . svg " alt = " " >
 * Individual cleanups can also be associated with mono cancellation and
 * error terminations :
 * < img class = " marble " src = " doc - files / marbles / usingWhenFailureForMono . svg " alt = " " >
 * Note that if the resource supplying { @ link Publisher } emits more than one resource , the
 * subsequent resources are dropped ( { @ link Operators # onNextDropped ( Object , Context ) } ) . If
 * the publisher errors AFTER having emitted one resource , the error is also silently dropped
 * ( { @ link Operators # onErrorDropped ( Throwable , Context ) } ) .
 * An empty completion or error without at least one onNext signal triggers a short - circuit
 * of the main sequence with the same terminal signal ( no resource is established , no
 * cleanup is invoked ) .
 * @ param resourceSupplier a { @ link Publisher } that " generates " the resource ,
 * subscribed for each subscription to the main sequence
 * @ param resourceClosure a factory to derive a { @ link Publisher } from the supplied resource
 * @ param asyncComplete an asynchronous resource cleanup invoked if the resource closure terminates with onComplete
 * @ param asyncError an asynchronous resource cleanup invoked if the resource closure terminates with onError
 * @ param < T > the type of elements emitted by the resource closure , and thus the main sequence
 * @ param < D > the type of the resource object
 * @ return a new { @ link Flux } built around a " transactional " resource , with several
 * termination path triggering asynchronous cleanup sequences
 * @ see # usingWhen ( Publisher , Function , Function , Function , Function ) */
public static < T , D > Mono < T > usingWhen ( Publisher < D > resourceSupplier , Function < ? super D , ? extends Mono < ? extends T > > resourceClosure , Function < ? super D , ? extends Publisher < ? > > asyncComplete , Function < ? super D , ? extends Publisher < ? > > asyncError ) { } }

implementation: return onAssembly ( new MonoUsingWhen < > ( resourceSupplier , resourceClosure , asyncComplete , asyncError , null ) ) ;


signature:
public class Mono { /** * Aggregate given publishers into a new { @ literal Mono } that will be fulfilled
 * when all of the given { @ literal sources } have completed . An error will cause
 * pending results to be cancelled and immediate error emission to the returned { @ link Mono } .
 * < img class = " marble " src = " doc - files / marbles / when . svg " alt = " " >
 * @ param sources The sources to use .
 * @ return a { @ link Mono } . */
public static Mono < Void > when ( Publisher < ? > ... sources ) { } }

implementation: if ( sources . length == 0 ) { return empty ( ) ; } if ( sources . length == 1 ) { return empty ( sources [ 0 ] ) ; } return onAssembly ( new MonoWhen ( false , sources ) ) ;


signature:
public class Mono { /** * Aggregate given publishers into a new { @ literal Mono } that will be
 * fulfilled when all of the given { @ literal Publishers } have completed .
 * An error will cause pending results to be cancelled and immediate error emission
 * to the returned { @ link Mono } .
 * < img class = " marble " src = " doc - files / marbles / when . svg " alt = " " >
 * @ param sources The sources to use .
 * @ return a { @ link Mono } . */
public static Mono < Void > when ( final Iterable < ? extends Publisher < ? > > sources ) { } }

implementation: return onAssembly ( new MonoWhen ( false , sources ) ) ;


signature:
public class Mono { /** * Aggregate given publishers into a new { @ literal Mono } that will be
 * fulfilled when all of the given { @ literal sources } have completed . If any Publisher
 * terminates without value , the returned sequence will be terminated immediately and
 * pending results cancelled . Errors from the sources are delayed .
 * If several Publishers error , the exceptions are combined ( as suppressed exceptions on a root exception ) .
 * < img class = " marble " src = " doc - files / marbles / whenDelayError . svg " alt = " " >
 * @ param sources The sources to use .
 * @ return a { @ link Mono } . */
public static Mono < Void > whenDelayError ( final Iterable < ? extends Publisher < ? > > sources ) { } }

implementation: return onAssembly ( new MonoWhen ( true , sources ) ) ;


signature:
public class Mono { /** * Merge given publishers into a new { @ literal Mono } that will be fulfilled when
 * all of the given { @ literal sources } have completed . Errors from the sources are delayed .
 * If several Publishers error , the exceptions are combined ( as suppressed exceptions on a root exception ) .
 * < img class = " marble " src = " doc - files / marbles / whenDelayError . svg " alt = " " >
 * @ param sources The sources to use .
 * @ return a { @ link Mono } . */
public static Mono < Void > whenDelayError ( Publisher < ? > ... sources ) { } }

implementation: if ( sources . length == 0 ) { return empty ( ) ; } if ( sources . length == 1 ) { return empty ( sources [ 0 ] ) ; } return onAssembly ( new MonoWhen ( true , sources ) ) ;


signature:
public class Mono { /** * Merge given monos into a new { @ literal Mono } that will be fulfilled when all of the given { @ literal Monos }
 * have produced an item , aggregating their values into a { @ link Tuple2 } .
 * An error or < strong > empty < / strong > completion of any source will cause other sources
 * to be cancelled and the resulting Mono to immediately error or complete , respectively .
 * < img class = " marble " src = " doc - files / marbles / zipFixedSourcesForMono . svg " alt = " " >
 * @ param p1 The first upstream { @ link Publisher } to subscribe to .
 * @ param p2 The second upstream { @ link Publisher } to subscribe to .
 * @ param < T1 > type of the value from p1
 * @ param < T2 > type of the value from p2
 * @ return a { @ link Mono } . */
public static < T1 , T2 > Mono < Tuple2 < T1 , T2 > > zip ( Mono < ? extends T1 > p1 , Mono < ? extends T2 > p2 ) { } }

implementation: return zip ( p1 , p2 , Flux . tuple2Function ( ) ) ;


signature:
public class Mono { /** * Merge given monos into a new { @ literal Mono } that will be fulfilled when all of the given { @ literal Monos }
 * have produced an item , aggregating their values as defined by the combinator function .
 * An error or < strong > empty < / strong > completion of any source will cause other sources
 * to be cancelled and the resulting Mono to immediately error or complete , respectively .
 * < img class = " marble " src = " doc - files / marbles / zipTwoSourcesWithZipperForMono . svg " alt = " " >
 * @ param p1 The first upstream { @ link Publisher } to subscribe to .
 * @ param p2 The second upstream { @ link Publisher } to subscribe to .
 * @ param combinator a { @ link BiFunction } combinator function when both sources
 * complete
 * @ param < T1 > type of the value from p1
 * @ param < T2 > type of the value from p2
 * @ param < O > output value
 * @ return a { @ link Mono } . */
public static < T1 , T2 , O > Mono < O > zip ( Mono < ? extends T1 > p1 , Mono < ? extends T2 > p2 , BiFunction < ? super T1 , ? super T2 , ? extends O > combinator ) { } }

implementation: return onAssembly ( new MonoZip < T1 , O > ( false , p1 , p2 , combinator ) ) ;


signature:
public class Mono { /** * Aggregate given monos into a new { @ literal Mono } that will be fulfilled when all of the given { @ literal
 * Monos } have produced an item , aggregating their values according to the provided combinator function .
 * An error or < strong > empty < / strong > completion of any source will cause other sources
 * to be cancelled and the resulting Mono to immediately error or complete , respectively .
 * < img class = " marble " src = " doc - files / marbles / zipIterableSourcesForMono . svg " alt = " " >
 * @ param monos The monos to use .
 * @ param combinator the function to transform the combined array into an arbitrary
 * object .
 * @ param < R > the combined result
 * @ return a { @ link Mono } . */
public static < R > Mono < R > zip ( final Iterable < ? extends Mono < ? > > monos , Function < ? super Object [ ] , ? extends R > combinator ) { } }

implementation: return onAssembly ( new MonoZip < > ( false , combinator , monos ) ) ;


signature:
public class Mono { /** * Aggregate given monos into a new { @ literal Mono } that will be fulfilled when all of the given { @ literal
 * Monos } have produced an item , aggregating their values according to the provided combinator function .
 * An error or < strong > empty < / strong > completion of any source will cause other sources
 * to be cancelled and the resulting Mono to immediately error or complete , respectively .
 * < img class = " marble " src = " doc - files / marbles / zipVarSourcesWithZipperForMono . svg " alt = " " >
 * @ param monos The monos to use .
 * @ param combinator the function to transform the combined array into an arbitrary
 * object .
 * @ param < R > the combined result
 * @ return a { @ link Mono } . */
public static < R > Mono < R > zip ( Function < ? super Object [ ] , ? extends R > combinator , Mono < ? > ... monos ) { } }

implementation: if ( monos . length == 0 ) { return empty ( ) ; } if ( monos . length == 1 ) { return monos [ 0 ] . map ( d -> combinator . apply ( new Object [ ] { d } ) ) ; } return onAssembly ( new MonoZip < > ( false , combinator , monos ) ) ;


signature:
public class Mono { /** * Merge given monos into a new { @ literal Mono } that will be fulfilled when all of the given { @ literal Monos }
 * have produced an item , aggregating their values into a { @ link Tuple2 } and delaying errors .
 * If a Mono source completes without value , the other source is run to completion then the
 * resulting { @ link Mono } completes empty .
 * If both Monos error , the two exceptions are combined ( as suppressed exceptions on a root exception ) .
 * < img class = " marble " src = " doc - files / marbles / zipDelayErrorFixedSources . svg " alt = " " >
 * @ param p1 The first upstream { @ link Publisher } to subscribe to .
 * @ param p2 The second upstream { @ link Publisher } to subscribe to .
 * @ param < T1 > type of the value from p1
 * @ param < T2 > type of the value from p2
 * @ return a { @ link Mono } . */
@ SuppressWarnings ( { } }

implementation: "unchecked" , "rawtypes" } ) public static < T1 , T2 > Mono < Tuple2 < T1 , T2 > > zipDelayError ( Mono < ? extends T1 > p1 , Mono < ? extends T2 > p2 ) { return onAssembly ( new MonoZip ( true , a -> Tuples . fromArray ( ( Object [ ] ) a ) , p1 , p2 ) ) ;


signature:
public class Mono { /** * Merge given monos into a new { @ literal Mono } that will be fulfilled when all of the given { @ literal Monos }
 * have produced an item , aggregating their values into a { @ link Tuple5 } and delaying errors .
 * If a Mono source completes without value , all other sources are run to completion then
 * the resulting { @ link Mono } completes empty .
 * If several Monos error , their exceptions are combined ( as suppressed exceptions on a root exception ) .
 * < img class = " marble " src = " doc - files / marbles / zipDelayErrorFixedSources . svg " alt = " " >
 * @ param p1 The first upstream { @ link Publisher } to subscribe to .
 * @ param p2 The second upstream { @ link Publisher } to subscribe to .
 * @ param p3 The third upstream { @ link Publisher } to subscribe to .
 * @ param p4 The fourth upstream { @ link Publisher } to subscribe to .
 * @ param p5 The fifth upstream { @ link Publisher } to subscribe to .
 * @ param < T1 > type of the value from p1
 * @ param < T2 > type of the value from p2
 * @ param < T3 > type of the value from p3
 * @ param < T4 > type of the value from p4
 * @ param < T5 > type of the value from p5
 * @ return a { @ link Mono } . */
@ SuppressWarnings ( { } }

implementation: "unchecked" , "rawtypes" } ) public static < T1 , T2 , T3 , T4 , T5 > Mono < Tuple5 < T1 , T2 , T3 , T4 , T5 > > zipDelayError ( Mono < ? extends T1 > p1 , Mono < ? extends T2 > p2 , Mono < ? extends T3 > p3 , Mono < ? extends T4 > p4 , Mono < ? extends T5 > p5 ) { return onAssembly ( new MonoZip ( true , a -> Tuples . fromArray ( ( Object [ ] ) a ) , p1 , p2 , p3 , p4 , p5 ) ) ;


signature:
public class Mono { /** * Aggregate given monos into a new { @ literal Mono } that will be fulfilled when all of the given { @ literal
 * Monos } have produced an item . Errors from the sources are delayed .
 * If a Mono source completes without value , all other sources are run to completion then
 * the resulting { @ link Mono } completes empty .
 * If several Monos error , their exceptions are combined ( as suppressed exceptions on a root exception ) .
 * < img class = " marble " src = " doc - files / marbles / zipDelayErrorIterableSources . svg " alt = " " >
 * @ param monos The monos to use .
 * @ param combinator the function to transform the combined array into an arbitrary
 * object .
 * @ param < R > the combined result
 * @ return a { @ link Mono } . */
public static < R > Mono < R > zipDelayError ( final Iterable < ? extends Mono < ? > > monos , Function < ? super Object [ ] , ? extends R > combinator ) { } }

implementation: return onAssembly ( new MonoZip < > ( true , combinator , monos ) ) ;


signature:
public class Mono { /** * Merge given monos into a new { @ literal Mono } that will be fulfilled when all of the
 * given { @ literal Monos } have produced an item , aggregating their values according to
 * the provided combinator function and delaying errors .
 * If a Mono source completes without value , all other sources are run to completion then
 * the resulting { @ link Mono } completes empty .
 * If several Monos error , their exceptions are combined ( as suppressed exceptions on a root exception ) .
 * < img class = " marble " src = " doc - files / marbles / zipDelayErrorVarSourcesWithZipper . svg " alt = " " >
 * @ param monos The monos to use .
 * @ param combinator the function to transform the combined array into an arbitrary
 * object .
 * @ param < R > the combined result
 * @ return a combined { @ link Mono } . */
public static < R > Mono < R > zipDelayError ( Function < ? super Object [ ] , ? extends R > combinator , Mono < ? > ... monos ) { } }

implementation: if ( monos . length == 0 ) { return empty ( ) ; } if ( monos . length == 1 ) { return monos [ 0 ] . map ( d -> combinator . apply ( new Object [ ] { d } ) ) ; } return onAssembly ( new MonoZip < > ( true , combinator , monos ) ) ;


signature:
public class Mono { /** * Transform this { @ link Mono } into a target type .
 * < blockquote > < pre >
 * { @ code mono . as ( Flux : : from ) . subscribe ( ) }
 * < / pre > < / blockquote >
 * @ param transformer the { @ link Function } to immediately map this { @ link Mono }
 * into a target type
 * @ param < P > the returned instance type
 * @ return the { @ link Mono } transformed to an instance of P
 * @ see # compose for a bounded conversion to { @ link Publisher } */
public final < P > P as ( Function < ? super Mono < T > , P > transformer ) { } }

implementation: return transformer . apply ( this ) ;


signature:
public class Mono { /** * Join the termination signals from this mono and another source into the returned
 * void mono
 * < img class = " marble " src = " doc - files / marbles / and . svg " alt = " " >
 * @ param other the { @ link Publisher } to wait for
 * complete
 * @ return a new combined Mono
 * @ see # when */
public final Mono < Void > and ( Publisher < ? > other ) { } }

implementation: if ( this instanceof MonoWhen ) { @ SuppressWarnings ( "unchecked" ) MonoWhen o = ( MonoWhen ) this ; Mono < Void > result = o . whenAdditionalSource ( other ) ; if ( result != null ) { return result ; } } return when ( this , other ) ;


signature:
public class Mono { /** * Subscribe to this { @ link Mono } and < strong > block indefinitely < / strong > until a next signal is
 * received . Returns that value , or null if the Mono completes empty . In case the Mono
 * errors , the original exception is thrown ( wrapped in a { @ link RuntimeException } if
 * it was a checked exception ) .
 * < img class = " marble " src = " doc - files / marbles / block . svg " alt = " " >
 * Note that each block ( ) will trigger a new subscription : in other words , the result
 * might miss signal from hot publishers .
 * @ return T the result */
@ Nullable public T block ( ) { } }

implementation: BlockingMonoSubscriber < T > subscriber = new BlockingMonoSubscriber < > ( ) ; Operators . onLastAssembly ( this ) . subscribe ( Operators . toCoreSubscriber ( subscriber ) ) ; return subscriber . blockingGet ( ) ;


signature:
public class Mono { /** * Subscribe to this { @ link Mono } and < strong > block < / strong > until a next signal is
 * received or a timeout expires . Returns that value , or null if the Mono completes
 * empty . In case the Mono errors , the original exception is thrown ( wrapped in a
 * { @ link RuntimeException } if it was a checked exception ) .
 * If the provided timeout expires , a { @ link RuntimeException } is thrown .
 * < img class = " marble " src = " doc - files / marbles / blockWithTimeout . svg " alt = " " >
 * Note that each block ( ) will trigger a new subscription : in other words , the result
 * might miss signal from hot publishers .
 * @ param timeout maximum time period to wait for before raising a { @ link RuntimeException }
 * @ return T the result */
@ Nullable public T block ( Duration timeout ) { } }

implementation: BlockingMonoSubscriber < T > subscriber = new BlockingMonoSubscriber < > ( ) ; Operators . onLastAssembly ( this ) . subscribe ( Operators . toCoreSubscriber ( subscriber ) ) ; return subscriber . blockingGet ( timeout . toMillis ( ) , TimeUnit . MILLISECONDS ) ;


signature:
public class Mono { /** * Subscribe to this { @ link Mono } and < strong > block indefinitely < / strong > until a next signal is
 * received or the Mono completes empty . Returns an { @ link Optional } , which can be used
 * to replace the empty case with an Exception via { @ link Optional # orElseThrow ( Supplier ) } .
 * In case the Mono itself errors , the original exception is thrown ( wrapped in a
 * { @ link RuntimeException } if it was a checked exception ) .
 * < img class = " marble " src = " doc - files / marbles / blockOptional . svg " alt = " " >
 * Note that each blockOptional ( ) will trigger a new subscription : in other words , the result
 * might miss signal from hot publishers .
 * @ return T the result */
public Optional < T > blockOptional ( ) { } }

implementation: BlockingOptionalMonoSubscriber < T > subscriber = new BlockingOptionalMonoSubscriber < > ( ) ; Operators . onLastAssembly ( this ) . subscribe ( Operators . toCoreSubscriber ( subscriber ) ) ; return subscriber . blockingGet ( ) ;


signature:
public class Mono { /** * Subscribe to this { @ link Mono } and < strong > block < / strong > until a next signal is
 * received , the Mono completes empty or a timeout expires . Returns an { @ link Optional }
 * for the first two cases , which can be used to replace the empty case with an
 * Exception via { @ link Optional # orElseThrow ( Supplier ) } .
 * In case the Mono itself errors , the original exception is thrown ( wrapped in a
 * { @ link RuntimeException } if it was a checked exception ) .
 * If the provided timeout expires , a { @ link RuntimeException } is thrown .
 * < img class = " marble " src = " doc - files / marbles / blockOptionalWithTimeout . svg " alt = " " >
 * Note that each block ( ) will trigger a new subscription : in other words , the result
 * might miss signal from hot publishers .
 * @ param timeout maximum time period to wait for before raising a { @ link RuntimeException }
 * @ return T the result */
public Optional < T > blockOptional ( Duration timeout ) { } }

implementation: BlockingOptionalMonoSubscriber < T > subscriber = new BlockingOptionalMonoSubscriber < > ( ) ; Operators . onLastAssembly ( this ) . subscribe ( Operators . toCoreSubscriber ( subscriber ) ) ; return subscriber . blockingGet ( timeout . toMillis ( ) , TimeUnit . MILLISECONDS ) ;


signature:
public class Mono { /** * Cast the current { @ link Mono } produced type into a target produced type .
 * < img class = " marble " src = " doc - files / marbles / castForMono . svg " alt = " " >
 * @ param < E > the { @ link Mono } output type
 * @ param clazz the target type to cast to
 * @ return a casted { @ link Mono } */
public final < E > Mono < E > cast ( Class < E > clazz ) { } }

implementation: Objects . requireNonNull ( clazz , "clazz" ) ; return map ( clazz :: cast ) ;


signature:
public class Mono { /** * Turn this { @ link Mono } into a hot source and cache last emitted signals for further
 * { @ link Subscriber } , with an expiry timeout .
 * Completion and Error will also be replayed until { @ code ttl } triggers in which case
 * the next { @ link Subscriber } will start over a new subscription .
 * < img class = " marble " src = " doc - files / marbles / cacheWithTtlForMono . svg " alt = " " >
 * @ param ttl Time - to - live for each cached item and post termination .
 * @ param timer the { @ link Scheduler } on which to measure the duration .
 * @ return a replaying { @ link Mono } */
public final Mono < T > cache ( Duration ttl , Scheduler timer ) { } }

implementation: return onAssembly ( new MonoCacheTime < > ( this , ttl , timer ) ) ;


signature:
public class Mono { /** * Turn this { @ link Mono } into a hot source and cache last emitted signal for further
 * { @ link Subscriber } , with an expiry timeout ( TTL ) that depends on said signal .
 * Empty completion and Error will also be replayed according to their respective TTL .
 * If the relevant TTL generator throws any { @ link Exception } , that exception will be
 * propagated to the { @ link Subscriber } that encountered the cache miss , but the cache
 * will be immediately cleared , so further Subscribers might re - populate the cache in
 * case the error was transient . In case the source was emitting an error , that error
 * is { @ link Hooks # onErrorDropped ( Consumer ) dropped } and added as a suppressed exception .
 * In case the source was emitting a value , that value is { @ link Hooks # onNextDropped ( Consumer ) dropped } .
 * @ param ttlForValue the TTL - generating { @ link Function } invoked when source is valued
 * @ param ttlForError the TTL - generating { @ link Function } invoked when source is erroring
 * @ param ttlForEmpty the TTL - generating { @ link Supplier } invoked when source is empty
 * @ return a replaying { @ link Mono } */
public final Mono < T > cache ( Function < ? super T , Duration > ttlForValue , Function < Throwable , Duration > ttlForError , Supplier < Duration > ttlForEmpty ) { } }

implementation: return onAssembly ( new MonoCacheTime < > ( this , ttlForValue , ttlForError , ttlForEmpty , Schedulers . parallel ( ) ) ) ;


signature:
public class Mono { /** * Activate assembly marker for this particular { @ link Mono } by giving it a description that
 * will be reflected in the assembly traceback in case of an error upstream of the
 * checkpoint . Note that unlike { @ link # checkpoint ( ) } , this doesn ' t create a
 * filled stack trace , avoiding the main cost of the operator .
 * However , as a trade - off the description must be unique enough for the user to find
 * out where this Mono was assembled . If you only want a generic description , and
 * still rely on the stack trace to find the assembly site , use the
 * { @ link # checkpoint ( String , boolean ) } variant .
 * It should be placed towards the end of the reactive chain , as errors
 * triggered downstream of it cannot be observed and augmented with assembly trace .
 * @ param description a unique enough description to include in the light assembly traceback .
 * @ return the assembly marked { @ link Mono } */
public final Mono < T > checkpoint ( String description ) { } }

implementation: return checkpoint ( Objects . requireNonNull ( description ) , false ) ;


signature:
public class Mono { /** * Activate assembly tracing or the lighter assembly marking depending on the
 * { @ code forceStackTrace } option .
 * By setting the { @ code forceStackTrace } parameter to { @ literal true } , activate assembly
 * tracing for this particular { @ link Mono } and give it a description that
 * will be reflected in the assembly traceback in case of an error upstream of the
 * checkpoint . Note that unlike { @ link # checkpoint ( String ) } , this will incur
 * the cost of an exception stack trace creation . The description could for
 * example be a meaningful name for the assembled mono or a wider correlation ID ,
 * since the stack trace will always provide enough information to locate where this
 * Flux was assembled .
 * By setting { @ code forceStackTrace } to { @ literal false } , behaves like
 * { @ link # checkpoint ( String ) } and is subject to the same caveat in choosing the
 * description .
 * It should be placed towards the end of the reactive chain , as errors
 * triggered downstream of it cannot be observed and augmented with assembly marker .
 * @ param description a description ( must be unique enough if forceStackTrace is set
 * to false ) .
 * @ param forceStackTrace false to make a light checkpoint without a stacktrace , true
 * to use a stack trace .
 * @ return the assembly marked { @ link Mono } . */
public final Mono < T > checkpoint ( @ Nullable String description , boolean forceStackTrace ) { } }

implementation: final AssemblySnapshot stacktrace ; if ( ! forceStackTrace ) { stacktrace = new AssemblyLightSnapshot ( description ) ; } else { stacktrace = new AssemblySnapshot ( description , Traces . callSiteSupplierFactory . get ( ) ) ; } return new MonoOnAssembly < > ( this , stacktrace ) ;


signature:
public class Mono { /** * Defer the given transformation to this { @ link Mono } in order to generate a
 * target { @ link Mono } type . A transformation will occur for each
 * { @ link Subscriber } . For instance :
 * < blockquote > < pre >
 * mono . compose ( original - > original . log ( ) ) ;
 * < / pre > < / blockquote >
 * < img class = " marble " src = " doc - files / marbles / composeForMono . svg " alt = " " >
 * @ param transformer the { @ link Function } to lazily map this { @ link Mono } into a target { @ link Mono }
 * instance upon subscription .
 * @ param < V > the item type in the returned { @ link Publisher }
 * @ return a new { @ link Mono }
 * @ see # as as ( ) for a loose conversion to an arbitrary type
 * @ see # transform ( Function ) */
public final < V > Mono < V > compose ( Function < ? super Mono < T > , ? extends Publisher < V > > transformer ) { } }

implementation: return defer ( ( ) -> from ( transformer . apply ( this ) ) ) ;


signature:
public class Mono { /** * Provide a default single value if this mono is completed without any data
 * < img class = " marble " src = " doc - files / marbles / defaultIfEmpty . svg " alt = " " >
 * @ param defaultV the alternate value if this sequence is empty
 * @ return a new { @ link Mono }
 * @ see Flux # defaultIfEmpty ( Object ) */
public final Mono < T > defaultIfEmpty ( T defaultV ) { } }

implementation:
if ( this instanceof Fuseable . ScalarCallable ) { try { T v = block ( ) ; if ( v == null ) { return Mono . just ( defaultV ) ; } } catch ( Throwable e ) { // leave MonoError returns as this
} return this ; } return onAssembly ( new MonoDefaultIfEmpty < > ( this , defaultV ) ) ;


signature:
public class Mono { /** * Delay this { @ link Mono } element ( { @ link Subscriber # onNext } signal ) by a given
 * { @ link Duration } , on a particular { @ link Scheduler } . Empty monos or error signals are not delayed .
 * < img class = " marble " src = " doc - files / marbles / delayElement . svg " alt = " " >
 * Note that the scheduler on which the mono chain continues execution will be the
 * scheduler provided if the mono is valued , or the current scheduler if the mono
 * completes empty or errors .
 * @ param delay { @ link Duration } by which to delay the { @ link Subscriber # onNext } signal
 * @ param timer a time - capable { @ link Scheduler } instance to delay the value signal on
 * @ return a delayed { @ link Mono } */
public final Mono < T > delayElement ( Duration delay , Scheduler timer ) { } }

implementation: return onAssembly ( new MonoDelayElement < > ( this , delay . toMillis ( ) , TimeUnit . MILLISECONDS , timer ) ) ;


signature:
public class Mono { /** * Subscribe to this { @ link Mono } and another { @ link Publisher } that is generated from
 * this Mono ' s element and which will be used as a trigger for relaying said element .
 * That is to say , the resulting { @ link Mono } delays until this Mono ' s element is
 * emitted , generates a trigger Publisher and then delays again until the trigger
 * Publisher terminates .
 * Note that contiguous calls to all delayUntil are fused together .
 * The triggers are generated and subscribed to in sequence , once the previous trigger
 * completes . Error is propagated immediately
 * downstream . In both cases , an error in the source is immediately propagated .
 * < img class = " marble " src = " doc - files / marbles / delayUntilForMono . svg " alt = " " >
 * @ param triggerProvider a { @ link Function } that maps this Mono ' s value into a
 * { @ link Publisher } whose termination will trigger relaying the value .
 * @ return this Mono , but delayed until the derived publisher terminates . */
public final Mono < T > delayUntil ( Function < ? super T , ? extends Publisher < ? > > triggerProvider ) { } }

implementation: Objects . requireNonNull ( triggerProvider , "triggerProvider required" ) ; if ( this instanceof MonoDelayUntil ) { return ( ( MonoDelayUntil < T > ) this ) . copyWithNewTriggerGenerator ( false , triggerProvider ) ; } return onAssembly ( new MonoDelayUntil < > ( this , triggerProvider ) ) ;


signature:
public class Mono { /** * Delay the { @ link Mono # subscribe ( Subscriber ) subscription } to this { @ link Mono } source until the given
 * { @ link Duration } elapses .
 * < img class = " marble " src = " doc - files / marbles / delaySubscriptionForMono . svg " alt = " " >
 * @ param delay { @ link Duration } before subscribing this { @ link Mono }
 * @ param timer a time - capable { @ link Scheduler } instance to run on
 * @ return a delayed { @ link Mono } */
public final Mono < T > delaySubscription ( Duration delay , Scheduler timer ) { } }

implementation: return delaySubscription ( Mono . delay ( delay , timer ) ) ;


signature:
public class Mono { /** * Delay the subscription to this { @ link Mono } until another { @ link Publisher }
 * signals a value or completes .
 * < img class = " marble " src = " doc - files / marbles / delaySubscriptionWithPublisherForMono . svg " alt = " " >
 * @ param subscriptionDelay a
 * { @ link Publisher } to signal by next or complete this { @ link Mono # subscribe ( Subscriber ) }
 * @ param < U > the other source type
 * @ return a delayed { @ link Mono } */
public final < U > Mono < T > delaySubscription ( Publisher < U > subscriptionDelay ) { } }

implementation: return onAssembly ( new MonoDelaySubscription < > ( this , subscriptionDelay ) ) ;