signature:
public class Mono { /** * An operator working only if this { @ link Mono } emits onNext , onError or onComplete { @ link Signal }
 * instances , transforming these { @ link # materialize ( ) materialized } signals into
 * real signals on the { @ link Subscriber } .
 * The error { @ link Signal } will trigger onError and complete { @ link Signal } will trigger
 * onComplete .
 * < img class = " marble " src = " doc - files / marbles / dematerializeForMono . svg " alt = " " >
 * @ param < X > the dematerialized type
 * @ return a dematerialized { @ link Mono }
 * @ see # materialize ( ) */
public final < X > Mono < X > dematerialize ( ) { } }

implementation: @ SuppressWarnings ( "unchecked" ) Mono < Signal < X > > thiz = ( Mono < Signal < X > > ) this ; return onAssembly ( new MonoDematerialize < > ( thiz ) ) ;


signature:
public class Mono { /** * Add behavior triggered after the { @ link Mono } terminates , either by completing downstream successfully or with an error .
 * The arguments will be null depending on success , success with data and error :
 * < ul >
 * < li > null , null : completed without data < / li >
 * < li > T , null : completed with data < / li >
 * < li > null , Throwable : failed with / without data < / li >
 * < / ul >
 * < img class = " marble " src = " doc - files / marbles / doAfterSuccessOrError . svg " alt = " " >
 * @ param afterSuccessOrError the callback to call after { @ link Subscriber # onNext } , { @ link Subscriber # onComplete } without preceding { @ link Subscriber # onNext } or { @ link Subscriber # onError }
 * @ return a new { @ link Mono } */
public final Mono < T > doAfterSuccessOrError ( BiConsumer < ? super T , Throwable > afterSuccessOrError ) { } }

implementation: return onAssembly ( new MonoPeekTerminal < > ( this , null , null , afterSuccessOrError ) ) ;


signature:
public class Mono { /** * Add behavior ( side - effect ) triggered after the { @ link Mono } terminates , either by
 * completing downstream successfully or with an error .
 * < img class = " marble " src = " doc - files / marbles / doAfterTerminateForMono . svg " alt = " " >
 * @ param afterTerminate the callback to call after { @ link Subscriber # onComplete } or { @ link Subscriber # onError }
 * @ return an observed { @ link Flux } */
public final Mono < T > doAfterTerminate ( Runnable afterTerminate ) { } }

implementation: Objects . requireNonNull ( afterTerminate , "afterTerminate" ) ; return doAfterSuccessOrError ( ( s , e ) -> afterTerminate . run ( ) ) ;


signature:
public class Mono { /** * Add behavior triggering < strong > after < / strong > the { @ link Mono } terminates for any reason ,
 * including cancellation . The terminating event ( { @ link SignalType # ON _ COMPLETE } ,
 * { @ link SignalType # ON _ ERROR } and { @ link SignalType # CANCEL } ) is passed to the consumer ,
 * which is executed after the signal has been passed downstream .
 * Note that the fact that the signal is propagated downstream before the callback is
 * executed means that several doFinally in a row will be executed in
 * < strong > reverse order < / strong > . If you want to assert the execution of the callback
 * please keep in mind that the Mono will complete before it is executed , so its
 * effect might not be visible immediately after eg . a { @ link # block ( ) } .
 * < img class = " marble " src = " doc - files / marbles / doFinallyForMono . svg " alt = " " >
 * @ param onFinally the callback to execute after a terminal signal ( complete , error
 * or cancel )
 * @ return an observed { @ link Mono } */
public final Mono < T > doFinally ( Consumer < SignalType > onFinally ) { } }

implementation: Objects . requireNonNull ( onFinally , "onFinally" ) ; if ( this instanceof Fuseable ) { return onAssembly ( new MonoDoFinallyFuseable < > ( this , onFinally ) ) ; } return onAssembly ( new MonoDoFinally < > ( this , onFinally ) ) ;


signature:
public class Mono { /** * Modify the behavior of the < i > whole chain < / i > of operators upstream of this one to
 * conditionally clean up elements that get < i > discarded < / i > by these operators .
 * The { @ code discardHook } must be idempotent and safe to use on any instance of the desired
 * type .
 * Calls to this method are additive , and the order of invocation of the { @ code discardHook }
 * is the same as the order of declaration ( calling { @ code . filter ( . . . ) . doOnDiscard ( first ) . doOnDiscard ( second ) }
 * will let the filter invoke { @ code first } then { @ code second } handlers ) .
 * Two main categories of discarding operators exist :
 * < ul >
 * < li > filtering operators , dropping some source elements as part of their designed behavior < / li >
 * < li > operators that prefetch a few elements and keep them around pending a request , but get cancelled / in error < / li >
 * < / ul >
 * These operators are identified in the javadoc by the presence of an { @ code onDiscard Support } section .
 * @ param type the { @ link Class } of elements in the upstream chain of operators that
 * this cleanup hook should take into account .
 * @ param discardHook a { @ link Consumer } of elements in the upstream chain of operators
 * that performs the cleanup .
 * @ return a { @ link Mono } that cleans up matching elements that get discarded upstream of it . */
public final < R > Mono < T > doOnDiscard ( final Class < R > type , final Consumer < ? super R > discardHook ) { } }

implementation: return subscriberContext ( Operators . discardLocalAdapter ( type , discardHook ) ) ;


signature:
public class Mono { /** * Add behavior triggered when the { @ link Mono } completes successfully .
 * < ul >
 * < li > null : completed without data < / li >
 * < li > T : completed with data < / li >
 * < / ul >
 * < img class = " marble " src = " doc - files / marbles / doOnSuccess . svg " alt = " " >
 * @ param onSuccess the callback to call on , argument is null if the { @ link Mono }
 * completes without data
 * { @ link Subscriber # onNext } or { @ link Subscriber # onComplete } without preceding { @ link Subscriber # onNext }
 * @ return a new { @ link Mono } */
public final Mono < T > doOnSuccess ( Consumer < ? super T > onSuccess ) { } }

implementation: Objects . requireNonNull ( onSuccess , "onSuccess" ) ; return onAssembly ( new MonoPeekTerminal < > ( this , onSuccess , null , null ) ) ;


signature:
public class Mono { /** * Add behavior triggered when the { @ link Mono } emits an item , fails with an error
 * or completes successfully . All these events are represented as a { @ link Signal }
 * that is passed to the side - effect callback . Note that this is an advanced operator ,
 * typically used for monitoring of a Mono .
 * These { @ link Signal } have a { @ link Context } associated to them .
 * < img class = " marble " src = " doc - files / marbles / doOnEachForMono . svg " alt = " " >
 * @ param signalConsumer the mandatory callback to call on
 * { @ link Subscriber # onNext ( Object ) } , { @ link Subscriber # onError ( Throwable ) } and
 * { @ link Subscriber # onComplete ( ) }
 * @ return an observed { @ link Mono }
 * @ see # doOnNext ( Consumer )
 * @ see # doOnError ( Consumer )
 * @ see # materialize ( )
 * @ see Signal */
public final Mono < T > doOnEach ( Consumer < ? super Signal < T > > signalConsumer ) { } }

implementation: Objects . requireNonNull ( signalConsumer , "signalConsumer" ) ; if ( this instanceof Fuseable ) { return onAssembly ( new MonoDoOnEachFuseable < > ( this , signalConsumer ) ) ; } return onAssembly ( new MonoDoOnEach < > ( this , signalConsumer ) ) ;


signature:
public class Mono { /** * Add behavior triggered when the { @ link Mono } completes with an error matching the given exception type .
 * < img class = " marble " src = " doc - files / marbles / doOnErrorWithClassPredicateForMono . svg " alt = " " >
 * @ param exceptionType the type of exceptions to handle
 * @ param onError the error handler for relevant errors
 * @ param < E > type of the error to handle
 * @ return an observed { @ link Mono } */
public final < E extends Throwable > Mono < T > doOnError ( Class < E > exceptionType , final Consumer < ? super E > onError ) { } }

implementation: Objects . requireNonNull ( exceptionType , "type" ) ; @ SuppressWarnings ( "unchecked" ) Consumer < Throwable > handler = ( Consumer < Throwable > ) onError ; return doOnError ( exceptionType :: isInstance , handler ) ;


signature:
public class Mono { /** * Add behavior triggering a { @ link LongConsumer } when the { @ link Mono } receives any request .
 * Note that non fatal error raised in the callback will not be propagated and
 * will simply trigger { @ link Operators # onOperatorError ( Throwable , Context ) } .
 * < img class = " marble " src = " doc - files / marbles / doOnRequestForMono . svg " alt = " " >
 * @ param consumer the consumer to invoke on each request
 * @ return an observed { @ link Mono } */
public final Mono < T > doOnRequest ( final LongConsumer consumer ) { } }

implementation: Objects . requireNonNull ( consumer , "consumer" ) ; return doOnSignal ( this , null , null , null , null , consumer , null ) ;


signature:
public class Mono { /** * Add behavior triggered when the { @ link Mono } terminates , either by completing successfully or with an error .
 * < ul >
 * < li > null , null : completing without data < / li >
 * < li > T , null : completing with data < / li >
 * < li > null , Throwable : failing with / without data < / li >
 * < / ul >
 * < img class = " marble " src = " doc - files / marbles / doOnTerminateForMono . svg " alt = " " >
 * @ param onSuccessOrError the callback to call { @ link Subscriber # onNext } , { @ link Subscriber # onComplete } without preceding { @ link Subscriber # onNext } or { @ link Subscriber # onError }
 * @ return a new { @ link Mono } */
public final Mono < T > doOnSuccessOrError ( BiConsumer < ? super T , Throwable > onSuccessOrError ) { } }

implementation: Objects . requireNonNull ( onSuccessOrError , "onSuccessOrError" ) ; return onAssembly ( new MonoPeekTerminal < > ( this , null , onSuccessOrError , null ) ) ;


signature:
public class Mono { /** * Map this { @ link Mono } sequence into { @ link reactor . util . function . Tuple2 Tuple2 & lt ; Long , T & gt ; }
 * of timemillis and source data . The timemillis corresponds to the elapsed time between the subscribe and the first
 * next signal , as measured by the provided { @ link Scheduler } .
 * < img class = " marble " src = " doc - files / marbles / elapsedForMono . svg " alt = " " >
 * @ param scheduler a { @ link Scheduler } instance to read time from
 * @ return a new { @ link Mono } that emits a tuple of time elapsed in milliseconds and matching data */
public final Mono < Tuple2 < Long , T > > elapsed ( Scheduler scheduler ) { } }

implementation: Objects . requireNonNull ( scheduler , "scheduler" ) ; return onAssembly ( new MonoElapsed < > ( this , scheduler ) ) ;


signature:
public class Mono { /** * Recursively expand elements into a graph and emit all the resulting element ,
 * in a depth - first traversal order .
 * That is : emit the value from this { @ link Mono } , expand it and emit the first value
 * at this first level of recursion , and so on . . . When no more recursion is possible ,
 * backtrack to the previous level and re - apply the strategy .
 * For example , given the hierarchical structure
 * < pre >
 * - AA
 * - aa1
 * - AB
 * - ab1
 * - a1
 * < / pre >
 * Expands { @ code Mono . just ( A ) } into
 * < pre >
 * AA
 * aa1
 * AB
 * ab1
 * a1
 * < / pre >
 * @ param expander the { @ link Function } applied at each level of recursion to expand
 * values into a { @ link Publisher } , producing a graph .
 * @ param capacityHint a capacity hint to prepare the inner queues to accommodate n
 * elements per level of recursion .
 * @ return this Mono expanded depth - first to a { @ link Flux } */
public final Flux < T > expandDeep ( Function < ? super T , ? extends Publisher < ? extends T > > expander , int capacityHint ) { } }

implementation: return Flux . onAssembly ( new MonoExpand < > ( this , expander , false , capacityHint ) ) ;


signature:
public class Mono { /** * Recursively expand elements into a graph and emit all the resulting element using
 * a breadth - first traversal strategy .
 * That is : emit the value from this { @ link Mono } first , then it each at a first level of
 * recursion and emit all of the resulting values , then expand all of these at a
 * second level and so on . . .
 * For example , given the hierarchical structure
 * < pre >
 * - AA
 * - aa1
 * - AB
 * - ab1
 * - a1
 * < / pre >
 * Expands { @ code Mono . just ( A ) } into
 * < pre >
 * AA
 * AB
 * a1
 * aa1
 * ab1
 * < / pre >
 * @ param expander the { @ link Function } applied at each level of recursion to expand
 * values into a { @ link Publisher } , producing a graph .
 * @ param capacityHint a capacity hint to prepare the inner queues to accommodate n
 * elements per level of recursion .
 * @ return this Mono expanded breadth - first to a { @ link Flux } */
public final Flux < T > expand ( Function < ? super T , ? extends Publisher < ? extends T > > expander , int capacityHint ) { } }

implementation: return Flux . onAssembly ( new MonoExpand < > ( this , expander , true , capacityHint ) ) ;


signature:
public class Mono { /** * If this { @ link Mono } is valued , test the result and replay it if predicate returns true .
 * Otherwise complete without value .
 * < img class = " marble " src = " doc - files / marbles / filterForMono . svg " alt = " " >
 * @ reactor . discard This operator discards the element if it does not match the filter . It
 * also discards upon cancellation or error triggered by a data signal .
 * @ param tester the predicate to evaluate
 * @ return a filtered { @ link Mono } */
public final Mono < T > filter ( final Predicate < ? super T > tester ) { } }

implementation: if ( this instanceof Fuseable ) { return onAssembly ( new MonoFilterFuseable < > ( this , tester ) ) ; } return onAssembly ( new MonoFilter < > ( this , tester ) ) ;


signature:
public class Mono { /** * If this { @ link Mono } is valued , test the value asynchronously using a generated
 * { @ code Publisher < Boolean > } test . The value from the Mono is replayed if the
 * first item emitted by the test is { @ literal true } . It is dropped if the test is
 * either empty or its first emitted value is { @ literal false } .
 * Note that only the first value of the test publisher is considered , and unless it
 * is a { @ link Mono } , test will be cancelled after receiving that first value .
 * < img class = " marble " src = " doc - files / marbles / filterWhenForMono . svg " alt = " " >
 * @ reactor . discard This operator discards the element if it does not match the filter . It
 * also discards upon cancellation or error triggered by a data signal .
 * @ param asyncPredicate the function generating a { @ link Publisher } of { @ link Boolean }
 * to filter the Mono with
 * @ return a filtered { @ link Mono } */
public final Mono < T > filterWhen ( Function < ? super T , ? extends Publisher < Boolean > > asyncPredicate ) { } }

implementation: return onAssembly ( new MonoFilterWhen < > ( this , asyncPredicate ) ) ;


signature:
public class Mono { /** * Transform the item emitted by this { @ link Mono } asynchronously , returning the
 * value emitted by another { @ link Mono } ( possibly changing the value type ) .
 * < img class = " marble " src = " doc - files / marbles / flatMapForMono . svg " alt = " " >
 * @ param transformer the function to dynamically bind a new { @ link Mono }
 * @ param < R > the result type bound
 * @ return a new { @ link Mono } with an asynchronously mapped value . */
public final < R > Mono < R > flatMap ( Function < ? super T , ? extends Mono < ? extends R > > transformer ) { } }

implementation: return onAssembly ( new MonoFlatMap < > ( this , transformer ) ) ;


signature:
public class Mono { /** * Transform the item emitted by this { @ link Mono } into a Publisher , then forward
 * its emissions into the returned { @ link Flux } .
 * < img class = " marble " src = " doc - files / marbles / flatMapMany . svg " alt = " " >
 * @ param mapper the
 * { @ link Function } to produce a sequence of R from the the eventual passed { @ link Subscriber # onNext }
 * @ param < R > the merged sequence type
 * @ return a new { @ link Flux } as the sequence is not guaranteed to be single at most */
public final < R > Flux < R > flatMapMany ( Function < ? super T , ? extends Publisher < ? extends R > > mapper ) { } }

implementation: return Flux . onAssembly ( new MonoFlatMapMany < > ( this , mapper ) ) ;


signature:
public class Mono { /** * Transform the signals emitted by this { @ link Mono } into signal - specific Publishers ,
 * then forward the applicable Publisher ' s emissions into the returned { @ link Flux } .
 * < img class = " marble " src = " doc - files / marbles / flatMapManyWithMappersOnTerminalEvents . svg " alt = " " >
 * @ param mapperOnNext the { @ link Function } to call on next data and returning a sequence to merge
 * @ param mapperOnError the { @ link Function } to call on error signal and returning a sequence to merge
 * @ param mapperOnComplete the { @ link Function } to call on complete signal and returning a sequence to merge
 * @ param < R > the type of the produced inner sequence
 * @ return a new { @ link Flux } as the sequence is not guaranteed to be single at most
 * @ see Flux # flatMap ( Function , Function , Supplier ) */
public final < R > Flux < R > flatMapMany ( Function < ? super T , ? extends Publisher < ? extends R > > mapperOnNext , Function < ? super Throwable , ? extends Publisher < ? extends R > > mapperOnError , Supplier < ? extends Publisher < ? extends R > > mapperOnComplete ) { } }

implementation: return flux ( ) . flatMap ( mapperOnNext , mapperOnError , mapperOnComplete ) ;


signature:
public class Mono { /** * Transform the item emitted by this { @ link Mono } into { @ link Iterable } , then forward
 * its elements into the returned { @ link Flux } . The prefetch argument allows to
 * give an arbitrary prefetch size to the inner { @ link Iterable } .
 * < img class = " marble " src = " doc - files / marbles / flatMapIterableForMono . svg " alt = " " >
 * @ param mapper the { @ link Function } to transform input item into a sequence { @ link Iterable }
 * @ param < R > the merged output sequence type
 * @ return a merged { @ link Flux } */
public final < R > Flux < R > flatMapIterable ( Function < ? super T , ? extends Iterable < ? extends R > > mapper ) { } }

implementation: return Flux . onAssembly ( new MonoFlattenIterable < > ( this , mapper , Integer . MAX_VALUE , Queues . one ( ) ) ) ;


signature:
public class Mono { /** * Convert this { @ link Mono } to a { @ link Flux }
 * @ return a { @ link Flux } variant of this { @ link Mono } */
public final Flux < T > flux ( ) { } }

implementation: if ( this instanceof Callable ) { if ( this instanceof Fuseable . ScalarCallable ) { T v ; try { v = block ( ) ; } catch ( Throwable t ) { return Flux . error ( t ) ; } if ( v == null ) { return Flux . empty ( ) ; } return Flux . just ( v ) ; } @ SuppressWarnings ( "unchecked" ) Callable < T > thiz = ( Callable < T > ) this ; return Flux . onAssembly ( new FluxCallable < > ( thiz ) ) ; } return Flux . wrap ( this ) ;


signature:
public class Mono { /** * Handle the items emitted by this { @ link Mono } by calling a biconsumer with the
 * output sink for each onNext . At most one { @ link SynchronousSink # next ( Object ) }
 * call must be performed and / or 0 or 1 { @ link SynchronousSink # error ( Throwable ) } or
 * { @ link SynchronousSink # complete ( ) } .
 * @ param handler the handling { @ link BiConsumer }
 * @ param < R > the transformed type
 * @ return a transformed { @ link Mono } */
public final < R > Mono < R > handle ( BiConsumer < ? super T , SynchronousSink < R > > handler ) { } }

implementation: if ( this instanceof Fuseable ) { return onAssembly ( new MonoHandleFuseable < > ( this , handler ) ) ; } return onAssembly ( new MonoHandle < > ( this , handler ) ) ;


signature:
public class Mono { /** * Observe all Reactive Streams signals and use { @ link Logger } support to handle trace implementation . Default will
 * use { @ link Level # INFO } and java . util . logging . If SLF4J is available , it will be used instead .
 * < img class = " marble " src = " doc - files / marbles / logForMono . svg " alt = " " >
 * @ param category to be mapped into logger configuration ( e . g . org . springframework
 * . reactor ) . If category ends with " . " like " reactor . " , a generated operator
 * suffix will complete , e . g . " reactor . Flux . Map " .
 * @ return a new { @ link Mono } */
public final Mono < T > log ( @ Nullable String category ) { } }

implementation: return log ( category , Level . INFO ) ;


signature:
public class Mono { /** * Observe Reactive Streams signals matching the passed filter { @ code options } and
 * use { @ link Logger } support to
 * handle trace
 * implementation . Default will
 * use the passed { @ link Level } and java . util . logging . If SLF4J is available , it will be used instead .
 * Options allow fine grained filtering of the traced signal , for instance to only capture onNext and onError :
 * < pre >
 * mono . log ( " category " , Level . INFO , SignalType . ON _ NEXT , SignalType . ON _ ERROR )
 * < / pre >
 * < img class = " marble " src = " doc - files / marbles / logForMono . svg " alt = " " >
 * @ param category to be mapped into logger configuration ( e . g . org . springframework
 * . reactor ) . If category ends with " . " like " reactor . " , a generated operator
 * suffix will complete , e . g . " reactor . Mono . Map " .
 * @ param level the { @ link Level } to enforce for this tracing Mono ( only FINEST , FINE ,
 * INFO , WARNING and SEVERE are taken into account )
 * @ param showOperatorLine capture the current stack to display operator
 * class / line number .
 * @ param options a vararg { @ link SignalType } option to filter log messages
 * @ return a new unaltered { @ link Mono } */
public final Mono < T > log ( @ Nullable String category , Level level , boolean showOperatorLine , SignalType ... options ) { } }

implementation: SignalLogger < T > log = new SignalLogger < > ( this , category , level , showOperatorLine , options ) ; if ( this instanceof Fuseable ) { return onAssembly ( new MonoLogFuseable < > ( this , log ) ) ; } return onAssembly ( new MonoLog < > ( this , log ) ) ;


signature:
public class Mono { /** * Observe Reactive Streams signals matching the passed filter { @ code options } and
 * trace them using a specific user - provided { @ link Logger } , at { @ link Level # INFO } level .
 * < img class = " marble " src = " doc - files / marbles / logForMono . svg " alt = " " >
 * @ param logger the { @ link Logger } to use , instead of resolving one through a category .
 * @ return a new { @ link Mono } that logs signals */
public final Mono < T > log ( Logger logger ) { } }

implementation: return log ( logger , Level . INFO , false ) ;


signature:
public class Mono { /** * Observe Reactive Streams signals matching the passed filter { @ code options } and
 * trace them using a specific user - provided { @ link Logger } , at the given { @ link Level } .
 * Options allow fine grained filtering of the traced signal , for instance to only
 * capture onNext and onError :
 * < pre >
 * flux . log ( myCustomLogger , Level . INFO , SignalType . ON _ NEXT , SignalType . ON _ ERROR )
 * < / pre >
 * < img class = " marble " src = " doc - files / marbles / logForMono . svg " alt = " " >
 * @ param logger the { @ link Logger } to use , instead of resolving one through a category .
 * @ param level the { @ link Level } to enforce for this tracing Flux ( only FINEST , FINE ,
 * INFO , WARNING and SEVERE are taken into account )
 * @ param showOperatorLine capture the current stack to display operator class / line number .
 * @ param options a vararg { @ link SignalType } option to filter log messages
 * @ return a new { @ link Mono } that logs signals */
public final Mono < T > log ( Logger logger , Level level , boolean showOperatorLine , SignalType ... options ) { } }

implementation: SignalLogger < T > log = new SignalLogger < > ( this , "IGNORED" , level , showOperatorLine , s -> logger , options ) ; if ( this instanceof Fuseable ) { return onAssembly ( new MonoLogFuseable < > ( this , log ) ) ; } return onAssembly ( new MonoLog < > ( this , log ) ) ;


signature:
public class Mono { /** * Transform the item emitted by this { @ link Mono } by applying a synchronous function to it .
 * < img class = " marble " src = " doc - files / marbles / mapForMono . svg " alt = " " >
 * @ param mapper the synchronous transforming { @ link Function }
 * @ param < R > the transformed type
 * @ return a new { @ link Mono } */
public final < R > Mono < R > map ( Function < ? super T , ? extends R > mapper ) { } }

implementation: if ( this instanceof Fuseable ) { return onAssembly ( new MonoMapFuseable < > ( this , mapper ) ) ; } return onAssembly ( new MonoMap < > ( this , mapper ) ) ;


signature:
public class Mono { /** * Activate metrics for this sequence , provided there is an instrumentation facade
 * on the classpath ( otherwise this method is a pure no - op ) .
 * Metrics are gathered on { @ link Subscriber } events , and it is recommended to also
 * { @ link # name ( String ) name } ( and optionally { @ link # tag ( String , String ) tag } ) the
 * sequence .
 * @ return an instrumented { @ link Mono } */
public final Mono < T > metrics ( ) { } }

implementation: if ( ! Metrics . isInstrumentationAvailable ( ) ) { return this ; } if ( this instanceof Fuseable ) { return onAssembly ( new MonoMetricsFuseable < > ( this ) ) ; } return onAssembly ( new MonoMetrics < > ( this ) ) ;


signature:
public class Mono { /** * Emit the first available result from this mono or the other mono .
 * < img class = " marble " src = " doc - files / marbles / orForMono . svg " alt = " " >
 * @ param other the racing other { @ link Mono } to compete with for the result
 * @ return a new { @ link Mono }
 * @ see # first */
public final Mono < T > or ( Mono < ? extends T > other ) { } }

implementation: if ( this instanceof MonoFirst ) { MonoFirst < T > a = ( MonoFirst < T > ) this ; Mono < T > result = a . orAdditionalSource ( other ) ; if ( result != null ) { return result ; } } return first ( this , other ) ;


signature:
public class Mono { /** * Evaluate the emitted value against the given { @ link Class } type . If the
 * value matches the type , it is passed into the new { @ link Mono } . Otherwise the
 * value is ignored .
 * < img class = " marble " src = " doc - files / marbles / ofTypeForMono . svg " alt = " " >
 * @ param clazz the { @ link Class } type to test values against
 * @ return a new { @ link Mono } filtered on the requested type */
public final < U > Mono < U > ofType ( final Class < U > clazz ) { } }

implementation: Objects . requireNonNull ( clazz , "clazz" ) ; return filter ( o -> clazz . isAssignableFrom ( o . getClass ( ) ) ) . cast ( clazz ) ;


signature:
public class Mono { /** * Transform any error emitted by this { @ link Mono } by synchronously applying a function to it .
 * < img class = " marble " src = " doc - files / marbles / onErrorMapForMono . svg " alt = " " >
 * @ param mapper the error transforming { @ link Function }
 * @ return a { @ link Mono } that transforms source errors to other errors */
public final Mono < T > onErrorMap ( Function < ? super Throwable , ? extends Throwable > mapper ) { } }

implementation: return onErrorResume ( e -> Mono . error ( mapper . apply ( e ) ) ) ;


signature:
public class Mono { /** * Subscribe to a fallback publisher when any error occurs , using a function to
 * choose the fallback depending on the error .
 * < img class = " marble " src = " doc - files / marbles / onErrorResumeForMono . svg " alt = " " >
 * @ param fallback the function to choose the fallback to an alternative { @ link Mono }
 * @ return a { @ link Mono } falling back upon source onError
 * @ see Flux # onErrorResume */
public final Mono < T > onErrorResume ( Function < ? super Throwable , ? extends Mono < ? extends T > > fallback ) { } }

implementation: return onAssembly ( new MonoOnErrorResume < > ( this , fallback ) ) ;


signature:
public class Mono { /** * Simply emit a captured fallback value when an error of the specified type is
 * observed on this { @ link Mono } .
 * < img class = " marble " src = " doc - files / marbles / onErrorReturnForMono . svg " alt = " " >
 * @ param type the error type to match
 * @ param fallbackValue the value to emit if an error occurs that matches the type
 * @ param < E > the error type
 * @ return a new falling back { @ link Mono } */
public final < E extends Throwable > Mono < T > onErrorReturn ( Class < E > type , T fallbackValue ) { } }

implementation: return onErrorResume ( type , throwable -> just ( fallbackValue ) ) ;


signature:
public class Mono { /** * Simply emit a captured fallback value when an error matching the given predicate is
 * observed on this { @ link Mono } .
 * < img class = " marble " src = " doc - files / marbles / onErrorReturnForMono . svg " alt = " " >
 * @ param predicate the error predicate to match
 * @ param fallbackValue the value to emit if an error occurs that matches the predicate
 * @ return a new { @ link Mono } */
public final Mono < T > onErrorReturn ( Predicate < ? super Throwable > predicate , T fallbackValue ) { } }

implementation: return onErrorResume ( predicate , throwable -> just ( fallbackValue ) ) ;


signature:
public class Mono { /** * Share a { @ link Mono } for the duration of a function that may transform it and
 * consume it as many times as necessary without causing multiple subscriptions
 * to the upstream .
 * @ param transform the transformation function
 * @ param < R > the output value type
 * @ return a new { @ link Mono } */
public final < R > Mono < R > publish ( Function < ? super Mono < T > , ? extends Mono < ? extends R > > transform ) { } }

implementation: return onAssembly ( new MonoPublishMulticast < > ( this , transform ) ) ;


signature:
public class Mono { /** * Run onNext , onComplete and onError on a supplied { @ link Scheduler }
 * { @ link Worker Worker } .
 * This operator influences the threading context where the rest of the operators in
 * the chain below it will execute , up to a new occurrence of { @ code publishOn } .
 * < img class = " marble " src = " doc - files / marbles / publishOnForMono . svg " alt = " " >
 * Typically used for fast publisher , slow consumer ( s ) scenarios .
 * < blockquote > < pre >
 * { @ code mono . publishOn ( Schedulers . single ( ) ) . subscribe ( ) }
 * < / pre > < / blockquote >
 * @ param scheduler a { @ link Scheduler } providing the { @ link Worker } where to publish
 * @ return an asynchronously producing { @ link Mono } */
public final Mono < T > publishOn ( Scheduler scheduler ) { } }

implementation:
if ( this instanceof Callable ) { if ( this instanceof Fuseable . ScalarCallable ) { try { T value = block ( ) ; return onAssembly ( new MonoSubscribeOnValue < > ( value , scheduler ) ) ; } catch ( Throwable t ) { // leave MonoSubscribeOnCallable defer error
} } @ SuppressWarnings ( "unchecked" ) Callable < T > c = ( Callable < T > ) this ; return onAssembly ( new MonoSubscribeOnCallable < > ( c , scheduler ) ) ; } return onAssembly ( new MonoPublishOn < > ( this , scheduler ) ) ;


signature:
public class Mono { /** * Repeatedly subscribe to the source { @ literal numRepeat } times . This results in
 * { @ code numRepeat + 1 } total subscriptions to the original source . As a consequence ,
 * using 0 plays the original sequence once .
 * < img class = " marble " src = " doc - files / marbles / repeatWithAttemptsForMono . svg " alt = " " >
 * @ param numRepeat the number of times to re - subscribe on onComplete ( positive , or 0 for original sequence only )
 * @ return a { @ link Flux } that repeats on onComplete , up to the specified number of repetitions */
public final Flux < T > repeat ( long numRepeat ) { } }

implementation: if ( numRepeat == 0 ) { return this . flux ( ) ; } return Flux . onAssembly ( new MonoRepeat < > ( this , numRepeat ) ) ;


signature:
public class Mono { /** * Repeatedly subscribe to this { @ link Mono } as long as the current subscription to this
 * { @ link Mono } completes empty and the companion { @ link Publisher } produces an onNext signal .
 * Any terminal signal will terminate the resulting { @ link Mono } with the same signal immediately .
 * < img class = " marble " src = " doc - files / marbles / repeatWhenEmpty . svg " alt = " " >
 * @ param repeatFactory the { @ link Function } that returns the associated { @ link Publisher }
 * companion , given a { @ link Flux } that signals each onComplete as a 0 - based incrementing { @ link Long } .
 * @ return a { @ link Mono } that resubscribes to this { @ link Mono } if the previous subscription was empty ,
 * as long as the companion { @ link Publisher } produces an onNext signal */
public final Mono < T > repeatWhenEmpty ( Function < Flux < Long > , ? extends Publisher < ? > > repeatFactory ) { } }

implementation: return repeatWhenEmpty ( Integer . MAX_VALUE , repeatFactory ) ;


signature:
public class Mono { /** * Repeatedly subscribe to this { @ link Mono } as long as the current subscription to this
 * { @ link Mono } completes empty and the companion { @ link Publisher } produces an onNext signal .
 * Any terminal signal will terminate the resulting { @ link Mono } with the same signal immediately .
 * Emits an { @ link IllegalStateException } if { @ code maxRepeat } is exceeded ( provided
 * it is different from { @ code Integer . MAX _ VALUE } ) .
 * < img class = " marble " src = " doc - files / marbles / repeatWhenEmpty . svg " alt = " " >
 * @ param maxRepeat the maximum number of repeats ( infinite if { @ code Integer . MAX _ VALUE } )
 * @ param repeatFactory the { @ link Function } that returns the associated { @ link Publisher }
 * companion , given a { @ link Flux } that signals each onComplete as a 0 - based incrementing { @ link Long } .
 * @ return a { @ link Mono } that resubscribes to this { @ link Mono } if the previous subscription was empty ,
 * as long as the companion { @ link Publisher } produces an onNext signal and the maximum number of repeats isn ' t exceeded . */
public final Mono < T > repeatWhenEmpty ( int maxRepeat , Function < Flux < Long > , ? extends Publisher < ? > > repeatFactory ) { } }

implementation: return Mono . defer ( ( ) -> { Flux < Long > iterations ; if ( maxRepeat == Integer . MAX_VALUE ) { iterations = Flux . fromStream ( LongStream . range ( 0 , Long . MAX_VALUE ) . boxed ( ) ) ; } else { iterations = Flux . range ( 0 , maxRepeat ) . map ( Integer :: longValue ) . concatWith ( Flux . error ( new IllegalStateException ( "Exceeded maximum number of repeats" ) , true ) ) ; } return this . repeatWhen ( o -> repeatFactory . apply ( o . zipWith ( iterations , 1 , ( c , i ) -> i ) ) ) . next ( ) ; } ) ;


signature:
public class Mono { /** * Re - subscribes to this { @ link Mono } sequence up to the specified number of retries if it signals any
 * error that match the given { @ link Predicate } , otherwise push the error downstream .
 * < img class = " marble " src = " doc - files / marbles / retryWithAttemptsAndPredicateForMono . svg " alt = " " >
 * @ param numRetries the number of times to tolerate an error
 * @ param retryMatcher the predicate to evaluate if retry should occur based on a given error signal
 * @ return a { @ link Mono } that retries on onError up to the specified number of retry
 * attempts , only if the predicate matches . */
public final Mono < T > retry ( long numRetries , Predicate < ? super Throwable > retryMatcher ) { } }

implementation: return defer ( ( ) -> retry ( Flux . countingPredicate ( retryMatcher , numRetries ) ) ) ;


signature:
public class Mono { /** * Retries this { @ link Mono } when a companion sequence signals
 * an item in response to this { @ link Mono } error signal
 * < p > If the companion sequence signals when the { @ link Mono } is active , the retry
 * attempt is suppressed and any terminal signal will terminate the { @ link Mono } source with the same signal
 * immediately .
 * < img class = " marble " src = " doc - files / marbles / retryWhenForMono . svg " alt = " " >
 * Note that if the companion { @ link Publisher } created by the { @ code whenFactory }
 * emits { @ link Context } as trigger objects , the content of these Context will be added
 * to the operator ' s own { @ link Context } .
 * @ param whenFactory the { @ link Function } that returns the associated { @ link Publisher }
 * companion , given a { @ link Flux } that signals each onError as a { @ link Throwable } .
 * @ return a { @ link Mono } that retries on onError when the companion { @ link Publisher } produces an
 * onNext signal */
public final Mono < T > retryWhen ( Function < Flux < Throwable > , ? extends Publisher < ? > > whenFactory ) { } }

implementation: return onAssembly ( new MonoRetryWhen < > ( this , whenFactory ) ) ;


signature:
public class Mono { /** * In case of error , retry this { @ link Mono } up to { @ code numRetries } times using a
 * randomized exponential backoff strategy ( jitter ) . The jitter factor is { @ code 50 % }
 * but the effective backoff delay cannot be less than { @ code firstBackoff } .
 * The randomized exponential backoff is good at preventing two typical issues with
 * other simpler backoff strategies , namely :
 * < ul >
 * < li >
 * having an exponentially growing backoff delay with a small initial delay gives
 * the best tradeoff between not overwhelming the server and serving the client as
 * fast as possible
 * < / li >
 * < li >
 * having a jitter , or randomized backoff delay , is beneficial in avoiding " retry - storms "
 * where eg . numerous clients would hit the server at the same time , causing it to
 * display transient failures which would cause all clients to retry at the same
 * backoff times , ultimately sparing no load on the server .
 * < / li >
 * < / ul >
 * < img class = " marble " src = " doc - files / marbles / retryBackoffForMono . svg " alt = " " >
 * @ param numRetries the maximum number of attempts before an { @ link IllegalStateException }
 * is raised ( having the original retry - triggering exception as cause ) .
 * @ param firstBackoff the first backoff delay to apply then grow exponentially . Also
 * minimum delay even taking jitter into account .
 * @ return a { @ link Mono } that retries on onError with exponentially growing randomized delays between retries . */
public final Mono < T > retryBackoff ( long numRetries , Duration firstBackoff ) { } }

implementation: return retryBackoff ( numRetries , firstBackoff , Duration . ofMillis ( Long . MAX_VALUE ) , 0.5d ) ;


signature:
public class Mono { /** * In case of error , retry this { @ link Mono } up to { @ code numRetries } times using a
 * randomized exponential backoff strategy . The jitter factor is { @ code 50 % }
 * but the effective backoff delay cannot be less than { @ code firstBackoff } nor more
 * than { @ code maxBackoff } .
 * The randomized exponential backoff is good at preventing two typical issues with
 * other simpler backoff strategies , namely :
 * < ul >
 * < li >
 * having an exponentially growing backoff delay with a small initial delay gives
 * the best tradeoff between not overwhelming the server and serving the client as
 * fast as possible
 * < / li >
 * < li >
 * having a jitter , or randomized backoff delay , is beneficial in avoiding " retry - storms "
 * where eg . numerous clients would hit the server at the same time , causing it to
 * display transient failures which would cause all clients to retry at the same
 * backoff times , ultimately sparing no load on the server .
 * < / li >
 * < / ul >
 * < img class = " marble " src = " doc - files / marbles / retryBackoffForMono . svg " alt = " " >
 * @ param numRetries the maximum number of attempts before an { @ link IllegalStateException }
 * is raised ( having the original retry - triggering exception as cause ) .
 * @ param firstBackoff the first backoff delay to apply then grow exponentially . Also
 * minimum delay even taking jitter into account .
 * @ param maxBackoff the maximum delay to apply despite exponential growth and jitter .
 * @ return a { @ link Mono } that retries on onError with exponentially growing randomized delays between retries . */
public final Mono < T > retryBackoff ( long numRetries , Duration firstBackoff , Duration maxBackoff ) { } }

implementation: return retryBackoff ( numRetries , firstBackoff , maxBackoff , 0.5d ) ;


signature:
public class Mono { /** * Expect exactly one item from this { @ link Mono } source or signal
 * { @ link java . util . NoSuchElementException } for an empty source .
 * < img class = " marble " src = " doc - files / marbles / singleForMono . svg " alt = " " >
 * Note Mono doesn ' t need { @ link Flux # single ( Object ) } , since it is equivalent to
 * { @ link # defaultIfEmpty ( Object ) } in a { @ link Mono } .
 * @ return a { @ link Mono } with the single item or an error signal */
public final Mono < T > single ( ) { } }

implementation: if ( this instanceof Callable ) { if ( this instanceof Fuseable . ScalarCallable ) { @ SuppressWarnings ( "unchecked" ) Fuseable . ScalarCallable < T > scalarCallable = ( Fuseable . ScalarCallable < T > ) this ; T v ; try { v = scalarCallable . call ( ) ; } catch ( Exception e ) { return Mono . error ( e ) ; } if ( v == null ) { return Mono . error ( new NoSuchElementException ( "Source was a (constant) empty" ) ) ; } return Mono . just ( v ) ; } @ SuppressWarnings ( "unchecked" ) Callable < T > thiz = ( Callable < T > ) this ; return Mono . onAssembly ( new MonoCallable < > ( thiz ) ) ; } return Mono . onAssembly ( new MonoSingleMono < > ( this ) ) ;


signature:
public class Mono { /** * Subscribe to this { @ link Mono } and request unbounded demand .
 * This version doesn ' t specify any consumption behavior for the events from the
 * chain , especially no error handling , so other variants should usually be preferred .
 * < img class = " marble " src = " doc - files / marbles / subscribeIgoringAllSignalsForMono . svg " alt = " " >
 * @ return a new { @ link Disposable } that can be used to cancel the underlying { @ link Subscription } */
public final Disposable subscribe ( ) { } }

implementation: if ( this instanceof MonoProcessor ) { MonoProcessor < T > s = ( MonoProcessor < T > ) this ; s . connect ( ) ; return s ; } else { return subscribeWith ( new LambdaMonoSubscriber < > ( null , null , null , null ) ) ; }


signature:
public class Mono { /** * Subscribe { @ link Consumer } to this { @ link Mono } that will respectively consume all the
 * elements in the sequence , handle errors , react to completion , and request upon subscription .
 * It will let the provided { @ link Subscription subscriptionConsumer }
 * request the adequate amount of data , or request unbounded demand
 * { @ code Long . MAX _ VALUE } if no such consumer is provided .
 * For a passive version that observe and forward incoming data see { @ link # doOnSuccess ( Consumer ) } and
 * { @ link # doOnError ( java . util . function . Consumer ) } .
 * Keep in mind that since the sequence can be asynchronous , this will immediately
 * return control to the calling thread . This can give the impression the consumer is
 * not invoked when executing in a main thread or a unit test for instance .
 * < img class = " marble " src = " doc - files / marbles / subscribeForMono . svg " alt = " " >
 * @ param consumer the consumer to invoke on each value
 * @ param errorConsumer the consumer to invoke on error signal
 * @ param completeConsumer the consumer to invoke on complete signal
 * @ param subscriptionConsumer the consumer to invoke on subscribe signal , to be used
 * for the initial { @ link Subscription # request ( long ) request } , or null for max request
 * @ return a new { @ link Disposable } that can be used to cancel the underlying { @ link Subscription } */
public final Disposable subscribe ( @ Nullable Consumer < ? super T > consumer , @ Nullable Consumer < ? super Throwable > errorConsumer , @ Nullable Runnable completeConsumer , @ Nullable Consumer < ? super Subscription > subscriptionConsumer ) { } }

implementation: return subscribeWith ( new LambdaMonoSubscriber < > ( consumer , errorConsumer , completeConsumer , subscriptionConsumer ) ) ;


signature:
public class Mono { /** * Enrich a potentially empty downstream { @ link Context } by adding all values
 * from the given { @ link Context } , producing a new { @ link Context } that is propagated
 * upstream .
 * The { @ link Context } propagation happens once per subscription ( not on each onNext ) :
 * it is done during the { @ code subscribe ( Subscriber ) } phase , which runs from
 * the last operator of a chain towards the first .
 * So this operator enriches a { @ link Context } coming from under it in the chain
 * ( downstream , by default an empty one ) and makes the new enriched { @ link Context }
 * visible to operators above it in the chain .
 * @ param mergeContext the { @ link Context } to merge with a previous { @ link Context }
 * state , returning a new one .
 * @ return a contextualized { @ link Mono }
 * @ see Context */
public final Mono < T > subscriberContext ( Context mergeContext ) { } }

implementation: return subscriberContext ( c -> c . putAll ( mergeContext ) ) ;


signature:
public class Mono { /** * Tag this mono with a key / value pair . These can be retrieved as a { @ link Set } of
 * all tags throughout the publisher chain by using { @ link Scannable # tags ( ) } ( as
 * traversed
 * by { @ link Scannable # parents ( ) } ) .
 * @ param key a tag key
 * @ param value a tag value
 * @ return the same sequence , but bearing tags */
public final Mono < T > tag ( String key , String value ) { } }

implementation: return MonoName . createOrAppend ( this , key , value ) ;


signature:
public class Mono { /** * Give this Mono a chance to resolve within a specified time frame but complete if it
 * doesn ' t . This works a bit like { @ link # timeout ( Duration ) } except that the resulting
 * { @ link Mono } completes rather than errors when the timer expires .
 * < img class = " marble " src = " doc - files / marbles / takeWithTimespanForMono . svg " alt = " " >
 * The timeframe is evaluated using the provided { @ link Scheduler } .
 * @ param duration the maximum duration to wait for the source Mono to resolve .
 * @ param timer the { @ link Scheduler } on which to measure the duration .
 * @ return a new { @ link Mono } that will propagate the signals from the source unless
 * no signal is received for { @ code duration } , in which case it completes . */
public final Mono < T > take ( Duration duration , Scheduler timer ) { } }

implementation: return takeUntilOther ( Mono . delay ( duration , timer ) ) ;


signature:
public class Mono { /** * Let this { @ link Mono } complete then play another Mono .
 * In other words ignore element from this { @ link Mono } and transform its completion signal into the
 * emission and completion signal of a provided { @ code Mono < V > } . Error signal is
 * replayed in the resulting { @ code Mono < V > } .
 * < img class = " marble " src = " doc - files / marbles / thenWithMonoForMono . svg " alt = " " >
 * @ reactor . discard This operator discards the element from the source .
 * @ param other a { @ link Mono } to emit from after termination
 * @ param < V > the element type of the supplied Mono
 * @ return a new { @ link Mono } that emits from the supplied { @ link Mono } */
public final < V > Mono < V > then ( Mono < V > other ) { } }

implementation: if ( this instanceof MonoIgnoreThen ) { MonoIgnoreThen < T > a = ( MonoIgnoreThen < T > ) this ; return a . shift ( other ) ; } return onAssembly ( new MonoIgnoreThen < > ( new Publisher [ ] { this } , other ) ) ;


signature:
public class Mono { /** * Let this { @ link Mono } complete then emit the provided value .
 * < img class = " marble " src = " doc - files / marbles / thenReturn . svg " alt = " " >
 * @ reactor . discard This operator discards the element from the source .
 * @ param value a value to emit after termination
 * @ param < V > the element type of the supplied value
 * @ return a new { @ link Mono } that emits the supplied value */
public final < V > Mono < V > thenReturn ( V value ) { } }

implementation: return then ( Mono . just ( value ) ) ;


signature:
public class Mono { /** * Let this { @ link Mono } complete then play another { @ link Publisher } .
 * In other words ignore element from this mono and transform the completion signal into a
 * { @ code Flux < V > } that will emit elements from the provided { @ link Publisher } .
 * < img class = " marble " src = " doc - files / marbles / thenManyForMono . svg " alt = " " >
 * @ reactor . discard This operator discards the element from the source .
 * @ param other a { @ link Publisher } to emit from after termination
 * @ param < V > the element type of the supplied Publisher
 * @ return a new { @ link Flux } that emits from the supplied { @ link Publisher } after
 * this Mono completes . */
public final < V > Flux < V > thenMany ( Publisher < V > other ) { } }

implementation: @ SuppressWarnings ( "unchecked" ) Flux < V > concat = ( Flux < V > ) Flux . concat ( ignoreElement ( ) , other ) ; return Flux . onAssembly ( concat ) ;


signature:
public class Mono { /** * Switch to a fallback { @ link Mono } in case no item arrives within the given { @ link Duration } .
 * If the fallback { @ link Mono } is null , signal a { @ link TimeoutException } instead .
 * < img class = " marble " src = " doc - files / marbles / timeoutFallbackForMono . svg " alt = " " >
 * @ param timeout the timeout before the onNext signal from this { @ link Mono }
 * @ param fallback the fallback { @ link Mono } to subscribe to when a timeout occurs
 * @ return a { @ link Mono } that will fallback to a different { @ link Mono } in case of timeout */
public final Mono < T > timeout ( Duration timeout , Mono < ? extends T > fallback ) { } }

implementation: return timeout ( timeout , fallback , Schedulers . parallel ( ) ) ;


signature:
public class Mono { /** * Signal a { @ link TimeoutException } error in case an item doesn ' t arrive before the given period ,
 * as measured on the provided { @ link Scheduler } .
 * < img class = " marble " src = " doc - files / marbles / timeoutForMono . svg " alt = " " >
 * @ param timeout the timeout before the onNext signal from this { @ link Mono }
 * @ param timer a time - capable { @ link Scheduler } instance to run the delay on
 * @ return an expirable { @ link Mono } */
public final Mono < T > timeout ( Duration timeout , Scheduler timer ) { } }

implementation: return timeout ( timeout , null , timer ) ;


signature:
public class Mono { /** * Switch to a fallback { @ link Mono } in case an item doesn ' t arrive before the given period ,
 * as measured on the provided { @ link Scheduler } .
 * < p > If the given { @ link Mono } is null , signal a { @ link TimeoutException } .
 * < img class = " marble " src = " doc - files / marbles / timeoutFallbackForMono . svg " alt = " " >
 * @ param timeout the timeout before the onNext signal from this { @ link Mono }
 * @ param fallback the fallback { @ link Mono } to subscribe when a timeout occurs
 * @ param timer a time - capable { @ link Scheduler } instance to run on
 * @ return an expirable { @ link Mono } with a fallback { @ link Mono } */
public final Mono < T > timeout ( Duration timeout , @ Nullable Mono < ? extends T > fallback , Scheduler timer ) { } }

implementation: final Mono < Long > _timer = Mono . delay ( timeout , timer ) . onErrorReturn ( 0L ) ; if ( fallback == null ) { return onAssembly ( new MonoTimeout < > ( this , _timer , timeout . toMillis ( ) + "ms" ) ) ; } return onAssembly ( new MonoTimeout < > ( this , _timer , fallback ) ) ;


signature:
public class Mono { /** * Signal a { @ link TimeoutException } in case the item from this { @ link Mono } has
 * not been emitted before the given { @ link Publisher } emits .
 * < img class = " marble " src = " doc - files / marbles / timeoutPublisher . svg " alt = " " >
 * @ param firstTimeout the timeout { @ link Publisher } that must not emit before the first signal from this { @ link Mono }
 * @ param < U > the element type of the timeout Publisher
 * @ return an expirable { @ link Mono } if the item does not come before a { @ link Publisher } signals */
public final < U > Mono < T > timeout ( Publisher < U > firstTimeout ) { } }

implementation: return onAssembly ( new MonoTimeout < > ( this , firstTimeout , "first signal from a Publisher" ) ) ;


signature:
public class Mono { /** * Wrap this { @ link Mono } into a { @ link MonoProcessor } ( turning it hot and allowing to block ,
 * cancel , as well as many other operations ) . Note that the { @ link MonoProcessor }
 * is subscribed to its parent source if any .
 * @ return a { @ link MonoProcessor } to use to either retrieve value or cancel the underlying { @ link Subscription } */
public final MonoProcessor < T > toProcessor ( ) { } }

implementation: MonoProcessor < T > result ; if ( this instanceof MonoProcessor ) { result = ( MonoProcessor < T > ) this ; } else { result = new MonoProcessor < > ( this ) ; } result . connect ( ) ; return result ;


signature:
public class Mono { /** * Transform this { @ link Mono } in order to generate a target { @ link Mono } . Unlike { @ link # compose ( Function ) } , the
 * provided function is executed as part of assembly .
 * < pre >
 * Function < Mono , Mono > applySchedulers = mono - > mono . subscribeOn ( Schedulers . io ( ) )
 * . publishOn ( Schedulers . parallel ( ) ) ;
 * mono . transform ( applySchedulers ) . map ( v - > v * v ) . subscribe ( ) ;
 * < / pre >
 * < img class = " marble " src = " doc - files / marbles / transformForMono . svg " alt = " " >
 * @ param transformer the { @ link Function } to immediately map this { @ link Mono } into a target { @ link Mono }
 * instance .
 * @ param < V > the item type in the returned { @ link Mono }
 * @ return a new { @ link Mono }
 * @ see # compose ( Function ) compose ( Function ) for deferred composition of { @ link Mono } for each { @ link Subscriber }
 * @ see # as ( Function ) as ( Function ) for a loose conversion to an arbitrary type */
public final < V > Mono < V > transform ( Function < ? super Mono < T > , ? extends Publisher < V > > transformer ) { } }

implementation: return onAssembly ( from ( transformer . apply ( this ) ) ) ;


signature:
public class Mono { /** * Wait for the result from this mono , use it to create a second mono via the
 * provided { @ code rightGenerator } function and combine both results into a { @ link Tuple2 } .
 * < img class = " marble " src = " doc - files / marbles / zipWhenForMono . svg " alt = " " >
 * @ param rightGenerator the { @ link Function } to generate a { @ code Mono } to combine with
 * @ param < T2 > the element type of the other Mono instance
 * @ return a new combined Mono */
public final < T2 > Mono < Tuple2 < T , T2 > > zipWhen ( Function < T , Mono < ? extends T2 > > rightGenerator ) { } }

implementation: return zipWhen ( rightGenerator , Tuples :: of ) ;


signature:
public class Mono { /** * Wait for the result from this mono , use it to create a second mono via the
 * provided { @ code rightGenerator } function and combine both results into an arbitrary
 * { @ code O } object , as defined by the provided { @ code combinator } function .
 * < img class = " marble " src = " doc - files / marbles / zipWhenWithZipperForMono . svg " alt = " " >
 * @ param rightGenerator the { @ link Function } to generate a { @ code Mono } to combine with
 * @ param combinator a { @ link BiFunction } combinator function when both sources complete
 * @ param < T2 > the element type of the other Mono instance
 * @ param < O > the element type of the combination
 * @ return a new combined Mono */
public final < T2 , O > Mono < O > zipWhen ( Function < T , Mono < ? extends T2 > > rightGenerator , BiFunction < T , T2 , O > combinator ) { } }

implementation: Objects . requireNonNull ( rightGenerator , "rightGenerator function is mandatory to get the right-hand side Mono" ) ; Objects . requireNonNull ( combinator , "combinator function is mandatory to combine results from both Monos" ) ; return flatMap ( t -> rightGenerator . apply ( t ) . map ( t2 -> combinator . apply ( t , t2 ) ) ) ;


signature:
public class Mono { /** * Combine the result from this mono and another into a { @ link Tuple2 } .
 * An error or < strong > empty < / strong > completion of any source will cause the other source
 * to be cancelled and the resulting Mono to immediately error or complete , respectively .
 * < img class = " marble " src = " doc - files / marbles / zipWithOtherForMono . svg " alt = " " >
 * @ param other the { @ link Mono } to combine with
 * @ param < T2 > the element type of the other Mono instance
 * @ return a new combined Mono */
public final < T2 > Mono < Tuple2 < T , T2 > > zipWith ( Mono < ? extends T2 > other ) { } }

implementation: return zipWith ( other , Flux . tuple2Function ( ) ) ;


signature:
public class Mono { /** * Combine the result from this mono and another into an arbitrary { @ code O } object ,
 * as defined by the provided { @ code combinator } function .
 * An error or < strong > empty < / strong > completion of any source will cause the other source
 * to be cancelled and the resulting Mono to immediately error or complete , respectively .
 * < img class = " marble " src = " doc - files / marbles / zipWithOtherUsingZipperForMono . svg " alt = " " >
 * @ param other the { @ link Mono } to combine with
 * @ param combinator a { @ link BiFunction } combinator function when both sources
 * complete
 * @ param < T2 > the element type of the other Mono instance
 * @ param < O > the element type of the combination
 * @ return a new combined Mono */
public final < T2 , O > Mono < O > zipWith ( Mono < ? extends T2 > other , BiFunction < ? super T , ? super T2 , ? extends O > combinator ) { } }

implementation: if ( this instanceof MonoZip ) { @ SuppressWarnings ( "unchecked" ) MonoZip < T , O > o = ( MonoZip < T , O > ) this ; Mono < O > result = o . zipAdditionalSource ( other , combinator ) ; if ( result != null ) { return result ; } } return zip ( this , other , combinator ) ;


signature:
public class Mono { /** * To be used by custom operators : invokes assembly { @ link Hooks } pointcut given a
 * { @ link Mono } , potentially returning a new { @ link Mono } . This is for example useful
 * to activate cross - cutting concerns at assembly time , eg . a generalized
 * { @ link # checkpoint ( ) } .
 * @ param < T > the value type
 * @ param source the source to apply assembly hooks onto
 * @ return the source , potentially wrapped with assembly time cross - cutting behavior
 * @ deprecated use { @ link Operators # onLastAssembly ( CorePublisher ) } */
@ SuppressWarnings ( "unchecked" ) @ Deprecated protected static < T > Mono < T > onLastAssembly ( Mono < T > source ) { } }

implementation: Function < Publisher , Publisher > hook = Hooks . onLastOperatorHook ; if ( hook == null ) { return source ; } return ( Mono < T > ) Objects . requireNonNull ( hook . apply ( source ) , "LastOperator hook returned null" ) ;


signature:
public class JdkFlowAdapter { /** * Return a java { @ code Flow . Publisher } from a { @ link Flux }
 * @ param publisher the source Publisher to convert
 * @ param < T > the type of the publisher
 * @ return a java { @ code Flow . Publisher } from the given { @ link Publisher } */
public static < T > Flow . Publisher < T > publisherToFlowPublisher ( final Publisher < T > publisher ) { } }

implementation: return new PublisherAsFlowPublisher < > ( publisher ) ;


signature:
public class JdkFlowAdapter { /** * Return a { @ link Flux } from a java { @ code Flow . Publisher }
 * @ param publisher the source Publisher to convert
 * @ param < T > the type of the publisher
 * @ return a { @ link Flux } from a java { @ code Flow . Publisher } */
public static < T > Flux < T > flowPublisherToFlux ( Flow . Publisher < T > publisher ) { } }

implementation: return new FlowPublisherAsFlux < > ( publisher ) ;


signature:
public class Hooks { /** * Add a { @ link Publisher } operator interceptor for each operator created
 * ( { @ link Flux } or { @ link Mono } ) . The passed function is applied to the original
 * operator { @ link Publisher } and can return a different { @ link Publisher } ,
 * on the condition that it generically maintains the same data type as the original .
 * Use of the { @ link Flux } / { @ link Mono } APIs is discouraged as it will recursively
 * call this hook , leading to { @ link StackOverflowError } .
 * Note that sub - hooks are cumulative , but invoking this method twice with the same instance
 * ( or any instance that has the same ` toString ` ) will result in only a single instance
 * being applied . See { @ link # onEachOperator ( String , Function ) } for a variant that
 * allows you to name the sub - hooks ( and thus replace them or remove them individually
 * later on ) . Can be fully reset via { @ link # resetOnEachOperator ( ) } .
 * This pointcut function cannot make use of { @ link Flux } , { @ link Mono } or
 * { @ link ParallelFlux } APIs as it would lead to a recursive call to the hook : the
 * operator calls would effectively invoke onEachOperator from onEachOperator .
 * @ param onEachOperator the sub - hook : a function to intercept each operation call
 * ( e . g . { @ code map ( fn ) } and { @ code map ( fn2 ) } in { @ code flux . map ( fn ) . map ( fn2 ) . subscribe ( ) } )
 * @ see # onEachOperator ( String , Function )
 * @ see # resetOnEachOperator ( String )
 * @ see # resetOnEachOperator ( )
 * @ see # onLastOperator ( Function ) */
public static void onEachOperator ( Function < ? super Publisher < Object > , ? extends Publisher < Object > > onEachOperator ) { } }

implementation: onEachOperator ( onEachOperator . toString ( ) , onEachOperator ) ;


signature:
public class Hooks { /** * Add or replace a named { @ link Publisher } operator interceptor for each operator created
 * ( { @ link Flux } or { @ link Mono } ) . The passed function is applied to the original
 * operator { @ link Publisher } and can return a different { @ link Publisher } ,
 * on the condition that it generically maintains the same data type as the original .
 * Use of the { @ link Flux } / { @ link Mono } APIs is discouraged as it will recursively
 * call this hook , leading to { @ link StackOverflowError } .
 * Note that sub - hooks are cumulative . Invoking this method twice with the same key will
 * replace the old sub - hook with that name , but keep the execution order ( eg . A - h1 , B - h2,
 * A - h3 will keep A - B execution order , leading to hooks h3 then h2 being executed ) .
 * Removing a particular key using { @ link # resetOnEachOperator ( String ) } then adding it
 * back will result in the execution order changing ( the later sub - hook being executed
 * last ) . Can be fully reset via { @ link # resetOnEachOperator ( ) } .
 * This pointcut function cannot make use of { @ link Flux } , { @ link Mono } or
 * { @ link ParallelFlux } APIs as it would lead to a recursive call to the hook : the
 * operator calls would effectively invoke onEachOperator from onEachOperator .
 * @ param key the key for the sub - hook to add / replace
 * @ param onEachOperator the sub - hook : a function to intercept each operation call
 * ( e . g . { @ code map ( fn ) } and { @ code map ( fn2 ) } in { @ code flux . map ( fn ) . map ( fn2 ) . subscribe ( ) } )
 * @ see # onEachOperator ( Function )
 * @ see # resetOnEachOperator ( String )
 * @ see # resetOnEachOperator ( )
 * @ see # onLastOperator ( String , Function ) */
public static void onEachOperator ( String key , Function < ? super Publisher < Object > , ? extends Publisher < Object > > onEachOperator ) { } }

implementation: Objects . requireNonNull ( key , "key" ) ; Objects . requireNonNull ( onEachOperator , "onEachOperator" ) ; log . debug ( "Hooking onEachOperator: {}" , key ) ; synchronized ( log ) { onEachOperatorHooks . put ( key , onEachOperator ) ; onEachOperatorHook = createOrUpdateOpHook ( onEachOperatorHooks . values ( ) ) ; }


signature:
public class Hooks { /** * Remove the sub - hook with key { @ code key } from the onEachOperator hook . No - op if
 * no such key has been registered , and equivalent to calling { @ link # resetOnEachOperator ( ) }
 * if it was the last sub - hook .
 * @ param key the key of the sub - hook to remove */
public static void resetOnEachOperator ( String key ) { } }

implementation: Objects . requireNonNull ( key , "key" ) ; log . debug ( "Reset onEachOperator: {}" , key ) ; synchronized ( log ) { onEachOperatorHooks . remove ( key ) ; onEachOperatorHook = createOrUpdateOpHook ( onEachOperatorHooks . values ( ) ) ; }


signature:
public class Hooks { /** * Override global error dropped strategy which by default bubble back the error .
 * The hook is cumulative , so calling this method several times will set up the hook
 * for as many consumer invocations ( even if called with the same consumer instance ) .
 * @ param c the { @ link Consumer } to apply to dropped errors */
public static void onErrorDropped ( Consumer < ? super Throwable > c ) { } }

implementation: Objects . requireNonNull ( c , "onErrorDroppedHook" ) ; log . debug ( "Hooking new default : onErrorDropped" ) ; synchronized ( log ) { if ( onErrorDroppedHook != null ) { @ SuppressWarnings ( "unchecked" ) Consumer < Throwable > _c = ( ( Consumer < Throwable > ) onErrorDroppedHook ) . andThen ( c ) ; onErrorDroppedHook = _c ; } else { onErrorDroppedHook = c ; } }


signature:
public class Hooks { /** * Add a { @ link Publisher } operator interceptor for the last operator created
 * in every flow ( { @ link Flux } or { @ link Mono } ) . The passed function is applied
 * to the original operator { @ link Publisher } and can return a different { @ link Publisher } ,
 * on the condition that it generically maintains the same data type as the original .
 * Note that sub - hooks are cumulative , but invoking this method twice with the same
 * instance ( or any instance that has the same ` toString ` ) will result in only a single
 * instance being applied . See { @ link # onLastOperator ( String , Function ) } for a variant
 * that allows you to name the sub - hooks ( and thus replace them or remove them individually
 * later on ) . Can be fully reset via { @ link # resetOnLastOperator ( ) } .
 * This pointcut function cannot make use of { @ link Flux } , { @ link Mono } or
 * { @ link ParallelFlux } APIs as it would lead to a recursive call to the hook : the
 * operator calls would effectively invoke onEachOperator from onEachOperator .
 * @ param onLastOperator the sub - hook : a function to intercept last operation call
 * ( e . g . { @ code map ( fn2 ) } in { @ code flux . map ( fn ) . map ( fn2 ) . subscribe ( ) } )
 * @ see # onLastOperator ( String , Function )
 * @ see # resetOnLastOperator ( String )
 * @ see # resetOnLastOperator ( )
 * @ see # onEachOperator ( Function ) */
public static void onLastOperator ( Function < ? super Publisher < Object > , ? extends Publisher < Object > > onLastOperator ) { } }

implementation: onLastOperator ( onLastOperator . toString ( ) , onLastOperator ) ;


signature:
public class Hooks { /** * Add or replace a named { @ link Publisher } operator interceptor for the last operator created
 * in every flow ( { @ link Flux } or { @ link Mono } ) . The passed function is applied
 * to the original operator { @ link Publisher } and can return a different { @ link Publisher } ,
 * on the condition that it generically maintains the same data type as the original .
 * Use of the { @ link Flux } / { @ link Mono } APIs is discouraged as it will recursively
 * call this hook , leading to { @ link StackOverflowError } .
 * Note that sub - hooks are cumulative . Invoking this method twice with the same key will
 * replace the old sub - hook with that name , but keep the execution order ( eg . A - h1 , B - h2,
 * A - h3 will keep A - B execution order , leading to hooks h3 then h2 being executed ) .
 * Removing a particular key using { @ link # resetOnLastOperator ( String ) } then adding it
 * back will result in the execution order changing ( the later sub - hook being executed
 * last ) . Can be fully reset via { @ link # resetOnLastOperator ( ) } .
 * This pointcut function cannot make use of { @ link Flux } , { @ link Mono } or
 * { @ link ParallelFlux } APIs as it would lead to a recursive call to the hook : the
 * operator calls would effectively invoke onEachOperator from onEachOperator .
 * @ param key the key for the sub - hook to add / replace
 * @ param onLastOperator the sub - hook : a function to intercept last operation call
 * ( e . g . { @ code map ( fn2 ) } in { @ code flux . map ( fn ) . map ( fn2 ) . subscribe ( ) } )
 * @ see # onLastOperator ( Function )
 * @ see # resetOnLastOperator ( String )
 * @ see # resetOnLastOperator ( )
 * @ see # onEachOperator ( String , Function ) */
public static void onLastOperator ( String key , Function < ? super Publisher < Object > , ? extends Publisher < Object > > onLastOperator ) { } }

implementation: Objects . requireNonNull ( key , "key" ) ; Objects . requireNonNull ( onLastOperator , "onLastOperator" ) ; log . debug ( "Hooking onLastOperator: {}" , key ) ; synchronized ( log ) { onLastOperatorHooks . put ( key , onLastOperator ) ; onLastOperatorHook = createOrUpdateOpHook ( onLastOperatorHooks . values ( ) ) ; }


signature:
public class Hooks { /** * Remove the sub - hook with key { @ code key } from the onLastOperator hook . No - op if
 * no such key has been registered , and equivalent to calling { @ link # resetOnLastOperator ( ) }
 * if it was the last sub - hook .
 * @ param key the key of the sub - hook to remove */
public static void resetOnLastOperator ( String key ) { } }

implementation: Objects . requireNonNull ( key , "key" ) ; log . debug ( "Reset onLastOperator: {}" , key ) ; synchronized ( log ) { onLastOperatorHooks . remove ( key ) ; onLastOperatorHook = createOrUpdateOpHook ( onLastOperatorHooks . values ( ) ) ; }


signature:
public class Hooks { /** * Override global data dropped strategy which by default logs at DEBUG level .
 * The hook is cumulative , so calling this method several times will set up the hook
 * for as many consumer invocations ( even if called with the same consumer instance ) .
 * @ param c the { @ link Consumer } to apply to data ( onNext ) that is dropped
 * @ see # onNextDroppedFail ( ) */
public static void onNextDropped ( Consumer < Object > c ) { } }

implementation: Objects . requireNonNull ( c , "onNextDroppedHook" ) ; log . debug ( "Hooking new default : onNextDropped" ) ; synchronized ( log ) { if ( onNextDroppedHook != null ) { onNextDroppedHook = onNextDroppedHook . andThen ( c ) ; } else { onNextDroppedHook = c ; } }


signature:
public class Hooks { /** * Set the custom global error mode hook for operators that support resuming
 * during an error in their { @ link org . reactivestreams . Subscriber # onNext ( Object ) } .
 * The hook is a { @ link BiFunction } of { @ link Throwable } and potentially null { @ link Object } .
 * If it is also a { @ link java . util . function . BiPredicate } , its
 * { @ link java . util . function . BiPredicate # test ( Object , Object ) test } method should be
 * used to determine if an error should be processed ( matching predicate ) or completely
 * skipped ( non - matching predicate ) . Typical usage , as in { @ link Operators } , is to
 * check if the predicate matches and fallback to { @ link Operators # onOperatorError ( Throwable , Context ) }
 * if it doesn ' t .
 * @ param onNextError the new { @ link BiFunction } to use . */
public static void onNextError ( BiFunction < ? super Throwable , Object , ? extends Throwable > onNextError ) { } }

implementation: Objects . requireNonNull ( onNextError , "onNextError" ) ; log . debug ( "Hooking new default : onNextError" ) ; if ( onNextError instanceof OnNextFailureStrategy ) { synchronized ( log ) { onNextErrorHook = ( OnNextFailureStrategy ) onNextError ; } } else { synchronized ( log ) { onNextErrorHook = new OnNextFailureStrategy . LambdaOnNextErrorStrategy ( onNextError ) ; } }


signature:
public class Hooks { /** * Add a custom error mapping , overriding the default one . Custom mapping can be an
 * accumulation of several sub - hooks each subsequently added via this method .
 * Note that sub - hooks are cumulative , but invoking this method twice with the same
 * instance ( or any instance that has the same ` toString ` ) will result in only a single
 * instance being applied . See { @ link # onOperatorError ( String , BiFunction ) } for a variant
 * that allows you to name the sub - hooks ( and thus replace them or remove them individually
 * later on ) . Can be fully reset via { @ link # resetOnOperatorError ( ) } .
 * For reference , the default mapping is to unwrap the exception and , if the second
 * parameter is another exception , to add it to the first as suppressed .
 * @ param onOperatorError an operator error { @ link BiFunction } mapper , returning an arbitrary exception
 * given the failure and optionally some original context ( data or error ) .
 * @ see # onOperatorError ( String , BiFunction )
 * @ see # resetOnOperatorError ( String )
 * @ see # resetOnOperatorError ( ) */
public static void onOperatorError ( BiFunction < ? super Throwable , Object , ? extends Throwable > onOperatorError ) { } }

implementation: onOperatorError ( onOperatorError . toString ( ) , onOperatorError ) ;


signature:
public class Hooks { /** * Add or replace a named custom error mapping , overriding the default one . Custom
 * mapping can be an accumulation of several sub - hooks each subsequently added via this
 * method .
 * Note that invoking this method twice with the same key will replace the old sub - hook
 * with that name , but keep the execution order ( eg . A - h1 , B - h2 , A - h3 will keep A - B
 * execution order , leading to hooks h3 then h2 being executed ) . Removing a particular
 * key using { @ link # resetOnOperatorError ( String ) } then adding it back will result in
 * the execution order changing ( the later sub - hook being executed last ) .
 * Can be fully reset via { @ link # resetOnOperatorError ( ) } .
 * For reference , the default mapping is to unwrap the exception and , if the second
 * parameter is another exception , to add it to the first as a suppressed .
 * @ param key the key for the sub - hook to add / replace
 * @ param onOperatorError an operator error { @ link BiFunction } mapper , returning an arbitrary exception
 * given the failure and optionally some original context ( data or error ) .
 * @ see # onOperatorError ( String , BiFunction )
 * @ see # resetOnOperatorError ( String )
 * @ see # resetOnOperatorError ( ) */
public static void onOperatorError ( String key , BiFunction < ? super Throwable , Object , ? extends Throwable > onOperatorError ) { } }

implementation: Objects . requireNonNull ( key , "key" ) ; Objects . requireNonNull ( onOperatorError , "onOperatorError" ) ; log . debug ( "Hooking onOperatorError: {}" , key ) ; synchronized ( log ) { onOperatorErrorHooks . put ( key , onOperatorError ) ; onOperatorErrorHook = createOrUpdateOpErrorHook ( onOperatorErrorHooks . values ( ) ) ; }


signature:
public class Hooks { /** * Remove the sub - hook with key { @ code key } from the onOperatorError hook . No - op if
 * no such key has been registered , and equivalent to calling { @ link # resetOnOperatorError ( ) }
 * if it was the last sub - hook .
 * @ param key the key of the sub - hook to remove */
public static void resetOnOperatorError ( String key ) { } }

implementation: Objects . requireNonNull ( key , "key" ) ; log . debug ( "Reset onOperatorError: {}" , key ) ; synchronized ( log ) { onOperatorErrorHooks . remove ( key ) ; onOperatorErrorHook = createOrUpdateOpErrorHook ( onOperatorErrorHooks . values ( ) ) ; }


signature:
public class Hooks { /** * Immutable views on hook trackers , for testing purpose */
static final Map < String , Function < ? super Publisher < Object > , ? extends Publisher < Object > > > getOnEachOperatorHooks ( ) { } }

implementation: return Collections . unmodifiableMap ( onEachOperatorHooks ) ;


signature:
public class FluxZip { /** * Handle values either from the iterable mode or array mode , taking into account the
 * possibility that some sources were already resolved ( being { @ link Callable } ) :
 * - if all sources have been resolved ( sc = = n ) , simply apply the mapper
 * - if some sources have been resolved ( n > sc > 0 ) , use a coordinator with the sparse
 * array , which will subscribe to unresolved sources only
 * - if no source is resolved , none was callable : use a coordinator without the sparse
 * array , resulting on an inner subscription for each source
 * @ param s the subscriber
 * @ param srcs the array of sources , some of which can be callable
 * @ param scalars a sparse array of values resolved for the callable sources , null if not resolved
 * @ param n the number of sources
 * @ param sc the number of already resolved sources in the scalars array */
void handleBoth ( CoreSubscriber < ? super R > s , Publisher < ? extends T > [ ] srcs , @ Nullable Object [ ] scalars , int n , int sc ) { } }

implementation: if ( sc != 0 && scalars != null ) { if ( n != sc ) { ZipSingleCoordinator < T , R > coordinator = new ZipSingleCoordinator < > ( s , scalars , n , zipper ) ; s . onSubscribe ( coordinator ) ; coordinator . subscribe ( n , sc , srcs ) ; } else { Operators . MonoSubscriber < R , R > sds = new Operators . MonoSubscriber < > ( s ) ; s . onSubscribe ( sds ) ; R r ; try { r = Objects . requireNonNull ( zipper . apply ( scalars ) , "The zipper returned a null value" ) ; } catch ( Throwable e ) { s . onError ( Operators . onOperatorError ( e , s . currentContext ( ) ) ) ; return ; } sds . complete ( r ) ; } } else { ZipCoordinator < T , R > coordinator = new ZipCoordinator < > ( s , zipper , n , queueSupplier , prefetch ) ; s . onSubscribe ( coordinator ) ; coordinator . subscribe ( srcs , n ) ; }


signature:
public class MonoDelayUntil { /** * Add a trigger generator to wait for .
 * @ param delayError the delayError parameter for the trigger being added , ignored if already true
 * @ param triggerGenerator the new trigger to add to the copy of the operator
 * @ return a new { @ link MonoDelayUntil } instance with same source but additional trigger generator */
@ SuppressWarnings ( "unchecked" ) MonoDelayUntil < T > copyWithNewTriggerGenerator ( boolean delayError , Function < ? super T , ? extends Publisher < ? > > triggerGenerator ) { } }

implementation: Objects . requireNonNull ( triggerGenerator , "triggerGenerator" ) ; Function < ? super T , ? extends Publisher < ? > > [ ] oldTriggers = this . otherGenerators ; Function < ? super T , ? extends Publisher < ? > > [ ] newTriggers = new Function [ oldTriggers . length + 1 ] ; System . arraycopy ( oldTriggers , 0 , newTriggers , 0 , oldTriggers . length ) ; newTriggers [ oldTriggers . length ] = triggerGenerator ; return new MonoDelayUntil < > ( this . source , newTriggers ) ;


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

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


signature:
public class Disposables { /** * Atomically set the field to a { @ link Disposable } and dispose the old content .
 * @ param updater the target field updater
 * @ param holder the target instance holding the field
 * @ param newValue the new Disposable to set
 * @ return true if successful , false if the field contains the { @ link # DISPOSED } instance . */
static < T > boolean set ( AtomicReferenceFieldUpdater < T , Disposable > updater , T holder , @ Nullable Disposable newValue ) { } }

implementation: for ( ; ; ) { Disposable current = updater . get ( holder ) ; if ( current == DISPOSED ) { if ( newValue != null ) { newValue . dispose ( ) ; } return false ; } if ( updater . compareAndSet ( holder , current , newValue ) ) { if ( current != null ) { current . dispose ( ) ; } return true ; } }


signature:
public class Disposables { /** * Atomically dispose the { @ link Disposable } in the field if not already disposed .
 * @ param updater the target field updater
 * @ param holder the target instance holding the field
 * @ return true if the { @ link Disposable } held by the field was properly disposed */
static < T > boolean dispose ( AtomicReferenceFieldUpdater < T , Disposable > updater , T holder ) { } }

implementation: Disposable current = updater . get ( holder ) ; Disposable d = DISPOSED ; if ( current != d ) { current = updater . getAndSet ( holder , d ) ; if ( current != d ) { if ( current != null ) { current . dispose ( ) ; } return true ; } } return false ;


signature:
public class FluxPeek { /** * see https : / / github . com / reactor / reactor - core / issues / 270 */
static < T > void afterCompleteWithFailure ( SignalPeek < T > parent , Throwable callbackFailure , Context context ) { } }

implementation: Exceptions . throwIfFatal ( callbackFailure ) ; Throwable _e = Operators . onOperatorError ( callbackFailure , context ) ; Operators . onErrorDropped ( _e , context ) ;


signature:
public class FluxPeek { /** * see https : / / github . com / reactor / reactor - core / issues / 270 */
static < T > void afterErrorWithFailure ( SignalPeek < T > parent , Throwable callbackFailure , Throwable originalError , Context context ) { } }

implementation: Exceptions . throwIfFatal ( callbackFailure ) ; Throwable _e = Operators . onOperatorError ( null , callbackFailure , originalError , context ) ; Operators . onErrorDropped ( _e , context ) ;


signature:
public class Tuples { /** * Create a { @ link Tuple2 } with the given array if it is small
 * enough to fit inside a { @ link Tuple2 } to { @ link Tuple8 } .
 * @ param list the content of the Tuple ( size 1 to 8)
 * @ return The new { @ link Tuple2 } .
 * @ throws IllegalArgumentException if the array is not of length 1-8 */
public static Tuple2 fromArray ( Object [ ] list ) { } }

implementation:
// noinspection ConstantConditions
if ( list == null || list . length < 2 ) { throw new IllegalArgumentException ( "null or too small array, need between 2 and 8 values" ) ; } switch ( list . length ) { case 2 : return of ( list [ 0 ] , list [ 1 ] ) ; case 3 : return of ( list [ 0 ] , list [ 1 ] , list [ 2 ] ) ; case 4 : return of ( list [ 0 ] , list [ 1 ] , list [ 2 ] , list [ 3 ] ) ; case 5 : return of ( list [ 0 ] , list [ 1 ] , list [ 2 ] , list [ 3 ] , list [ 4 ] ) ; case 6 : return of ( list [ 0 ] , list [ 1 ] , list [ 2 ] , list [ 3 ] , list [ 4 ] , list [ 5 ] ) ; case 7 : return of ( list [ 0 ] , list [ 1 ] , list [ 2 ] , list [ 3 ] , list [ 4 ] , list [ 5 ] , list [ 6 ] ) ; case 8 : return of ( list [ 0 ] , list [ 1 ] , list [ 2 ] , list [ 3 ] , list [ 4 ] , list [ 5 ] , list [ 6 ] , list [ 7 ] ) ; } throw new IllegalArgumentException ( "too many arguments (" + list . length + "), need between 2 and 8 values" ) ;


signature:
public class Tuples { /** * Create a { @ link Tuple2 } with the given objects .
 * @ param t1 The first value in the tuple . Not null .
 * @ param t2 The second value in the tuple . Not null .
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ return The new { @ link Tuple2 } . */
public static < T1 , T2 > Tuple2 < T1 , T2 > of ( T1 t1 , T2 t2 ) { } }

implementation: return new Tuple2 < > ( t1 , t2 ) ;


signature:
public class Tuples { /** * Create a { @ link Tuple3 } with the given objects .
 * @ param t1 The first value in the tuple . Not null .
 * @ param t2 The second value in the tuple . Not null .
 * @ param t3 The third value in the tuple . Not null .
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ param < T3 > The type of the third value .
 * @ return The new { @ link Tuple3 } . */
public static < T1 , T2 , T3 > Tuple3 < T1 , T2 , T3 > of ( T1 t1 , T2 t2 , T3 t3 ) { } }

implementation: return new Tuple3 < > ( t1 , t2 , t3 ) ;


signature:
public class Tuples { /** * Create a { @ link Tuple6 } with the given objects .
 * @ param t1 The first value in the tuple . Not null .
 * @ param t2 The second value in the tuple . Not null .
 * @ param t3 The third value in the tuple . Not null .
 * @ param t4 The fourth value in the tuple . Not null .
 * @ param t5 The fifth value in the tuple . Not null .
 * @ param t6 The sixth value in the tuple . Not null .
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ param < T3 > The type of the third value .
 * @ param < T4 > The type of the fourth value .
 * @ param < T5 > The type of the fifth value .
 * @ param < T6 > The type of the sixth value .
 * @ return The new { @ link Tuple6 } . */
public static < T1 , T2 , T3 , T4 , T5 , T6 > Tuple6 < T1 , T2 , T3 , T4 , T5 , T6 > of ( T1 t1 , T2 t2 , T3 t3 , T4 t4 , T5 t5 , T6 t6 ) { } }

implementation: return new Tuple6 < > ( t1 , t2 , t3 , t4 , t5 , t6 ) ;


signature:
public class Tuples { /** * A converting function from Object array to { @ link Tuples } to R .
 * @ param < R > The type of the return value .
 * @ param delegate the function to delegate to
 * @ return The unchecked conversion function to R . */
public static < R > Function < Object [ ] , R > fnAny ( final Function < Tuple2 , R > delegate ) { } }

implementation: return objects -> delegate . apply ( Tuples . fnAny ( ) . apply ( objects ) ) ;


signature:
public class Tuples { /** * A converting function from Object array to { @ link Tuple2}
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ return The unchecked conversion function to { @ link Tuple2 } . */
@ SuppressWarnings ( "unchecked" ) public static < T1 , T2 > Function < Object [ ] , Tuple2 < T1 , T2 > > fn2 ( ) { } }

implementation: return empty ;


signature:
public class Tuples { /** * A converting function from Object array to { @ link Tuple3}
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ param < T3 > The type of the third value .
 * @ return The unchecked conversion function to { @ link Tuple3 } . */
@ SuppressWarnings ( "unchecked" ) public static < T1 , T2 , T3 > Function < Object [ ] , Tuple3 < T1 , T2 , T3 > > fn3 ( ) { } }


signature:
public class Tuples { /** * A converting function from Object array to { @ link Tuple3 } to R .
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ param < T3 > The type of the third value .
 * @ param < R > The type of the return value .
 * @ param delegate the function to delegate to
 * @ return The unchecked conversion function to R . */
public static < T1 , T2 , T3 , R > Function < Object [ ] , R > fn3 ( final Function < Tuple3 < T1 , T2 , T3 > , R > delegate ) { } }

implementation: return objects -> delegate . apply ( Tuples . < T1 , T2 , T3 > fn3 ( ) . apply ( objects ) ) ;


signature:
public class Tuples { /** * A converting function from Object array to { @ link Tuple4}
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ param < T3 > The type of the third value .
 * @ param < T4 > The type of the fourth value .
 * @ return The unchecked conversion function to { @ link Tuple4 } . */
@ SuppressWarnings ( "unchecked" ) public static < T1 , T2 , T3 , T4 > Function < Object [ ] , Tuple4 < T1 , T2 , T3 , T4 > > fn4 ( ) { } }


signature:
public class Tuples { /** * A converting function from Object array to { @ link Tuple4 } to R .
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ param < T3 > The type of the third value .
 * @ param < T4 > The type of the fourth value .
 * @ param < R > The type of the return value .
 * @ param delegate the function to delegate to
 * @ return The unchecked conversion function to R . */
public static < T1 , T2 , T3 , T4 , R > Function < Object [ ] , R > fn4 ( final Function < Tuple4 < T1 , T2 , T3 , T4 > , R > delegate ) { } }

implementation: return objects -> delegate . apply ( Tuples . < T1 , T2 , T3 , T4 > fn4 ( ) . apply ( objects ) ) ;


signature:
public class Tuples { /** * A converting function from Object array to { @ link Tuple5}
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ param < T3 > The type of the third value .
 * @ param < T4 > The type of the fourth value .
 * @ param < T5 > The type of the fifth value .
 * @ return The unchecked conversion function to { @ link Tuple5 } . */
@ SuppressWarnings ( "unchecked" ) public static < T1 , T2 , T3 , T4 , T5 > Function < Object [ ] , Tuple5 < T1 , T2 , T3 , T4 , T5 > > fn5 ( ) { } }


signature:
public class Tuples { /** * A converting function from Object array to { @ link Tuple4 } to R .
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ param < T3 > The type of the third value .
 * @ param < T4 > The type of the fourth value .
 * @ param < T5 > The type of the fifth value .
 * @ param < R > The type of the return value .
 * @ param delegate the function to delegate to
 * @ return The unchecked conversion function to R . */
public static < T1 , T2 , T3 , T4 , T5 , R > Function < Object [ ] , R > fn5 ( final Function < Tuple5 < T1 , T2 , T3 , T4 , T5 > , R > delegate ) { } }

implementation: return objects -> delegate . apply ( Tuples . < T1 , T2 , T3 , T4 , T5 > fn5 ( ) . apply ( objects ) ) ;


signature:
public class Tuples { /** * A converting function from Object array to { @ link Tuple6}
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ param < T3 > The type of the third value .
 * @ param < T4 > The type of the fourth value .
 * @ param < T5 > The type of the fifth value .
 * @ param < T6 > The type of the sixth value .
 * @ return The unchecked conversion function to { @ link Tuple6 } . */
@ SuppressWarnings ( "unchecked" ) public static < T1 , T2 , T3 , T4 , T5 , T6 > Function < Object [ ] , Tuple6 < T1 , T2 , T3 , T4 , T5 , T6 > > fn6 ( ) { } }


signature:
public class Tuples { /** * A converting function from Object array to { @ link Tuple6 } to R .
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ param < T3 > The type of the third value .
 * @ param < T4 > The type of the fourth value .
 * @ param < T5 > The type of the fifth value .
 * @ param < T6 > The type of the sixth value .
 * @ param < R > The type of the return value .
 * @ param delegate the function to delegate to
 * @ return The unchecked conversion function to R . */
public static < T1 , T2 , T3 , T4 , T5 , T6 , R > Function < Object [ ] , R > fn6 ( final Function < Tuple6 < T1 , T2 , T3 , T4 , T5 , T6 > , R > delegate ) { } }

implementation: return objects -> delegate . apply ( Tuples . < T1 , T2 , T3 , T4 , T5 , T6 > fn6 ( ) . apply ( objects ) ) ;


signature:
public class Tuples { /** * A converting function from Object array to { @ link Tuple7}
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ param < T3 > The type of the third value .
 * @ param < T4 > The type of the fourth value .
 * @ param < T5 > The type of the fifth value .
 * @ param < T6 > The type of the sixth value .
 * @ param < T7 > The type of the seventh value .
 * @ return The unchecked conversion function to { @ link Tuple7 } . */
@ SuppressWarnings ( "unchecked" ) public static < T1 , T2 , T3 , T4 , T5 , T6 , T7 > Function < Object [ ] , Tuple7 < T1 , T2 , T3 , T4 , T5 , T6 , T7 > > fn7 ( ) { } }


signature:
public class Tuples { /** * A converting function from Object array to { @ link Tuple7 } to R .
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ param < T3 > The type of the third value .
 * @ param < T4 > The type of the fourth value .
 * @ param < T5 > The type of the fifth value .
 * @ param < T6 > The type of the sixth value .
 * @ param < T7 > The type of the seventh value .
 * @ param < R > The type of the return value .
 * @ param delegate the function to delegate to
 * @ return The unchecked conversion function to R . */
public static < T1 , T2 , T3 , T4 , T5 , T6 , T7 , R > Function < Object [ ] , R > fn7 ( final Function < Tuple7 < T1 , T2 , T3 , T4 , T5 , T6 , T7 > , R > delegate ) { } }

implementation: return objects -> delegate . apply ( Tuples . < T1 , T2 , T3 , T4 , T5 , T6 , T7 > fn7 ( ) . apply ( objects ) ) ;


signature:
public class Tuples { /** * A converting function from Object array to { @ link Tuple8}
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ param < T3 > The type of the third value .
 * @ param < T4 > The type of the fourth value .
 * @ param < T5 > The type of the fifth value .
 * @ param < T6 > The type of the sixth value .
 * @ param < T7 > The type of the seventh value .
 * @ param < T8 > The type of the eighth value .
 * @ return The unchecked conversion function to { @ link Tuple8 } . */
@ SuppressWarnings ( "unchecked" ) public static < T1 , T2 , T3 , T4 , T5 , T6 , T7 , T8 > Function < Object [ ] , Tuple8 < T1 , T2 , T3 , T4 , T5 , T6 , T7 , T8 > > fn8 ( ) { } }


signature:
public class Tuples { /** * A converting function from Object array to { @ link Tuple8}
 * @ param < T1 > The type of the first value .
 * @ param < T2 > The type of the second value .
 * @ param < T3 > The type of the third value .
 * @ param < T4 > The type of the fourth value .
 * @ param < T5 > The type of the fifth value .
 * @ param < T6 > The type of the sixth value .
 * @ param < T7 > The type of the seventh value .
 * @ param < T8 > The type of the eighth value .
 * @ param < R > The type of the return value .
 * @ param delegate the function to delegate to
 * @ return The unchecked conversion function to { @ link Tuple8 } . */
public static < T1 , T2 , T3 , T4 , T5 , T6 , T7 , T8 , R > Function < Object [ ] , R > fn8 ( final Function < Tuple8 < T1 , T2 , T3 , T4 , T5 , T6 , T7 , T8 > , R > delegate ) { } }

implementation: return objects -> delegate . apply ( Tuples . < T1 , T2 , T3 , T4 , T5 , T6 , T7 , T8 > fn8 ( ) . apply ( objects ) ) ;


signature:
public class Tuples { /** * Prepare a string representation of the values suitable for a Tuple of any
 * size by accepting an array of elements . This builds a { @ link StringBuilder }
 * containing the String representation of each object , comma separated . It manages
 * nulls as well by putting an empty string and the comma .
 * @ param values the values of the tuple to represent
 * @ return a { @ link StringBuilder } initialized with the string representation of the
 * values in the Tuple . */
static StringBuilder tupleStringRepresentation ( Object ... values ) { } }

implementation: StringBuilder sb = new StringBuilder ( ) ; for ( int i = 0 ; i < values . length ; i ++ ) { Object t = values [ i ] ; if ( i != 0 ) { sb . append ( ',' ) ; } if ( t != null ) { sb . append ( t ) ; } } return sb ;


signature:
public class Schedulers { /** * Create a { @ link Scheduler } which uses a backing { @ link Executor } to schedule
 * Runnables for async operators .
 * Trampolining here means tasks submitted in a burst are queued by the Worker itself ,
 * which acts as a sole task from the perspective of the { @ link ExecutorService } ,
 * so no reordering ( but also no threading ) .
 * @ param executor an { @ link Executor }
 * @ param trampoline set to false if this { @ link Scheduler } is used by " operators "
 * that already conflate { @ link Runnable } executions ( publishOn , subscribeOn . . . )
 * @ return a new { @ link Scheduler } */
public static Scheduler fromExecutor ( Executor executor , boolean trampoline ) { } }

implementation: if ( ! trampoline && executor instanceof ExecutorService ) { return fromExecutorService ( ( ExecutorService ) executor ) ; } return new ExecutorScheduler ( executor , trampoline ) ;


signature:
public class Schedulers { /** * { @ link Scheduler } that dynamically creates ExecutorService - based Workers and caches
 * the thread pools , reusing them once the Workers have been shut down .
 * The maximum number of created thread pools is unbounded .
 * This scheduler is not restartable .
 * @ param name Thread prefix
 * @ param ttlSeconds Time - to - live for an idle { @ link reactor . core . scheduler . Scheduler . Worker }
 * @ param daemon false if the { @ link Scheduler } requires an explicit { @ link
 * Scheduler # dispose ( ) } to exit the VM .
 * @ return a new { @ link Scheduler } that hosts a fixed pool of single - threaded
 * ExecutorService - based workers and is suited for parallel work */
public static Scheduler newElastic ( String name , int ttlSeconds , boolean daemon ) { } }

implementation: return newElastic ( ttlSeconds , new ReactorThreadFactory ( name , ElasticScheduler . COUNTER , daemon , false , Schedulers :: defaultUncaughtException ) ) ;