signature:
public class Flux { /** * Add behavior ( side - effects ) triggered when the { @ link Flux } 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 Flux . These { @ link Signal } have a { @ link Context }
 * associated to them .
 * < img class = " marble " src = " doc - files / marbles / doOnEachForFlux . 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 Flux }
 * @ see # doOnNext ( Consumer )
 * @ see # doOnError ( Consumer )
 * @ see # doOnComplete ( Runnable )
 * @ see # materialize ( )
 * @ see Signal */
public final Flux < T > doOnEach ( Consumer < ? super Signal < T > > signalConsumer ) { } }

implementation: if ( this instanceof Fuseable ) { return onAssembly ( new FluxDoOnEachFuseable < > ( this , signalConsumer ) ) ; } return onAssembly ( new FluxDoOnEach < > ( this , signalConsumer ) ) ;


signature:
public class Flux { /** * Add behavior ( side - effect ) triggered when the { @ link Flux } completes with an error matching the given exception .
 * < img class = " marble " src = " doc - files / marbles / doOnErrorWithPredicateForFlux . svg " alt = " " >
 * @ param predicate the matcher for exceptions to handle
 * @ param onError the error handler for each error
 * @ return an observed { @ link Flux } */
public final Flux < T > doOnError ( Predicate < ? super Throwable > predicate , final Consumer < ? super Throwable > onError ) { } }

implementation: Objects . requireNonNull ( predicate , "predicate" ) ; return doOnError ( t -> { if ( predicate . test ( t ) ) { onError . accept ( t ) ; } } ) ;


signature:
public class Flux { /** * Add behavior ( side - effect ) triggered when the { @ link Flux } emits an item .
 * < img class = " marble " src = " doc - files / marbles / doOnNextForFlux . svg " alt = " " >
 * @ param onNext the callback to call on { @ link Subscriber # onNext }
 * @ reactor . errorMode This operator supports { @ link # onErrorContinue ( BiConsumer ) resuming on errors }
 * ( including when fusion is enabled ) . Exceptions thrown by the consumer are passed to
 * the { @ link # onErrorContinue ( BiConsumer ) } error consumer ( the value consumer
 * is not invoked , as the source element will be part of the sequence ) . The onNext
 * signal is then propagated as normal .
 * @ return an observed { @ link Flux } */
public final Flux < T > doOnNext ( Consumer < ? super T > onNext ) { } }

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


signature:
public class Flux { /** * Add behavior ( side - effect ) triggered when the { @ link Flux } is subscribed .
 * This method is < strong > not < / strong > intended for capturing the subscription and calling its methods ,
 * but for side effects like monitoring . For instance , the correct way to cancel a subscription is
 * to call { @ link Disposable # dispose ( ) } on the Disposable returned by { @ link Flux # subscribe ( ) } .
 * < img class = " marble " src = " doc - files / marbles / doOnSubscribe . svg " alt = " " >
 * @ param onSubscribe the callback to call on { @ link Subscriber # onSubscribe }
 * @ return an observed { @ link Flux } */
public final Flux < T > doOnSubscribe ( Consumer < ? super Subscription > onSubscribe ) { } }

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


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

implementation: Objects . requireNonNull ( onTerminate , "onTerminate" ) ; return doOnSignal ( this , null , null , e -> onTerminate . run ( ) , onTerminate , null , null , null ) ;


signature:
public class Flux { /** * Add behavior ( side - effect ) triggered < strong > after < / strong > the { @ link Flux } 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 Flux will complete before it is executed , so its
 * effect might not be visible immediately after eg . a { @ link # blockLast ( ) } .
 * < img class = " marble " src = " doc - files / marbles / doFinallyForFlux . svg " alt = " " >
 * @ param onFinally the callback to execute after a terminal signal ( complete , error
 * or cancel )
 * @ return an observed { @ link Flux } */
public final Flux < T > doFinally ( Consumer < SignalType > onFinally ) { } }

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


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

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


signature:
public class Flux { /** * Emit only the element at the given index position or fall back to a
 * default value if the sequence is shorter .
 * < img class = " marble " src = " doc - files / marbles / elementAtWithDefault . svg " alt = " " >
 * @ reactor . discard This operator discards elements that appear before the requested index .
 * @ param index zero - based index of the only item to emit
 * @ param defaultValue a default value to emit if the sequence is shorter
 * @ return a { @ link Mono } of the item at the specified zero - based index or a default value */
public final Mono < T > elementAt ( int index , T defaultValue ) { } }

implementation: return Mono . onAssembly ( new MonoElementAt < > ( this , index , defaultValue ) ) ;


signature:
public class Flux { /** * Recursively expand elements into a graph and emit all the resulting element ,
 * in a depth - first traversal order .
 * That is : emit one value from this { @ link Flux } , 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
 * - BB
 * - bb1
 * < / pre >
 * Expands { @ code Flux . just ( A , B ) } into
 * < pre >
 * AA
 * aa1
 * BB
 * bb1
 * < / pre >
 * @ param expander the { @ link Function } applied at each level of recursion to expand
 * values into a { @ link Publisher } , producing a graph .
 * @ return a { @ link Flux } expanded depth - first */
public final Flux < T > expandDeep ( Function < ? super T , ? extends Publisher < ? extends T > > expander ) { } }

implementation: return expandDeep ( expander , Queues . SMALL_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Recursively expand elements into a graph and emit all the resulting element using
 * a breadth - first traversal strategy .
 * That is : emit the values from this { @ link Flux } first , then expand 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
 * - BB
 * - bb1
 * < / pre >
 * Expands { @ code Flux . just ( A , B ) } into
 * < pre >
 * AA
 * BB
 * aa1
 * bb1
 * < / pre >
 * @ param expander the { @ link Function } applied at each level of recursion to expand
 * values into a { @ link Publisher } , producing a graph .
 * @ return an breadth - first expanded { @ link Flux } */
public final Flux < T > expand ( Function < ? super T , ? extends Publisher < ? extends T > > expander ) { } }

implementation: return expand ( expander , Queues . SMALL_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Evaluate each source value against the given { @ link Predicate } . If the predicate test succeeds , the value is
 * emitted . If the predicate test fails , the value is ignored and a request of 1 is made upstream .
 * < img class = " marble " src = " doc - files / marbles / filterForFlux . svg " alt = " " >
 * @ param p the { @ link Predicate } to test values against
 * @ reactor . errorMode This operator supports { @ link # onErrorContinue ( BiConsumer ) resuming on errors }
 * ( including when fusion is enabled ) . Exceptions thrown by the predicate are
 * considered as if the predicate returned false : they cause the source value to be
 * dropped and a new element ( { @ code request ( 1 ) } ) being requested from upstream .
 * @ reactor . discard This operator discards elements that do not match the filter . It
 * also discards elements internally queued for backpressure upon cancellation or error triggered by a data signal .
 * @ return a new { @ link Flux } containing only values that pass the predicate test */
public final Flux < T > filter ( Predicate < ? super T > p ) { } }

implementation: if ( this instanceof Fuseable ) { return onAssembly ( new FluxFilterFuseable < > ( this , p ) ) ; } return onAssembly ( new FluxFilter < > ( this , p ) ) ;


signature:
public class Flux { /** * Test each value emitted by this { @ link Flux } asynchronously using a generated
 * { @ code Publisher < Boolean > } test . A value is replayed if the first item emitted
 * by its corresponding test is { @ literal true } . It is dropped if its 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 . Test
 * publishers are generated and subscribed to in sequence .
 * < img class = " marble " src = " doc - files / marbles / filterWhenForFlux . svg " alt = " " >
 * @ reactor . discard This operator discards elements that do not match the filter . It
 * also discards elements internally queued for backpressure upon cancellation or error triggered by a data signal .
 * @ param asyncPredicate the function generating a { @ link Publisher } of { @ link Boolean }
 * for each value , to filter the Flux with
 * @ return a filtered { @ link Flux } */
public final Flux < T > filterWhen ( Function < ? super T , ? extends Publisher < Boolean > > asyncPredicate ) { } }

implementation: return filterWhen ( asyncPredicate , Queues . SMALL_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Test each value emitted by this { @ link Flux } asynchronously using a generated
 * { @ code Publisher < Boolean > } test . A value is replayed if the first item emitted
 * by its corresponding test is { @ literal true } . It is dropped if its 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 . Test
 * publishers are generated and subscribed to in sequence .
 * < img class = " marble " src = " doc - files / marbles / filterWhenForFlux . svg " alt = " " >
 * @ reactor . discard This operator discards elements that do not match the filter . It
 * also discards elements internally queued for backpressure upon cancellation or error triggered by a data signal .
 * @ param asyncPredicate the function generating a { @ link Publisher } of { @ link Boolean }
 * for each value , to filter the Flux with
 * @ param bufferSize the maximum expected number of values to hold pending a result of
 * their respective asynchronous predicates , rounded to the next power of two . This is
 * capped depending on the size of the heap and the JVM limits , so be careful with
 * large values ( although eg . { @ literal 65536 } should still be fine ) . Also serves as
 * the initial request size for the source .
 * @ return a filtered { @ link Flux } */
public final Flux < T > filterWhen ( Function < ? super T , ? extends Publisher < Boolean > > asyncPredicate , int bufferSize ) { } }

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


signature:
public class Flux { /** * Transform the elements emitted by this { @ link Flux } asynchronously into Publishers ,
 * then flatten these inner publishers into a single { @ link Flux } through merging ,
 * which allow them to interleave .
 * There are three dimensions to this operator that can be compared with
 * { @ link # flatMapSequential ( Function ) flatMapSequential } and { @ link # concatMap ( Function ) concatMap } :
 * < ul >
 * < li > < b > Generation of inners and subscription < / b > : this operator is eagerly
 * subscribing to its inners . < / li >
 * < li > < b > Ordering of the flattened values < / b > : this operator does not necessarily preserve
 * original ordering , as inner element are flattened as they arrive . < / li >
 * < li > < b > Interleaving < / b > : this operator lets values from different inners interleave
 * ( similar to merging the inner sequences ) . < / li >
 * < / ul >
 * < img class = " marble " src = " doc - files / marbles / flatMapForFlux . svg " alt = " " >
 * @ reactor . discard This operator discards elements internally queued for backpressure upon cancellation or error triggered by a data signal .
 * @ param mapper the { @ link Function } to transform input sequence into N sequences { @ link Publisher }
 * @ param < R > the merged output sequence type
 * @ reactor . errorMode This operator supports { @ link # onErrorContinue ( BiConsumer ) resuming on errors }
 * in the mapper { @ link Function } . Exceptions thrown by the mapper then behave as if
 * it had mapped the value to an empty publisher . If the mapper does map to a scalar
 * publisher ( an optimization in which the value can be resolved immediately without
 * subscribing to the publisher , e . g . a { @ link Mono # fromCallable ( Callable ) } ) but said
 * publisher throws , this can be resumed from in the same manner .
 * @ return a new { @ link Flux } */
public final < R > Flux < R > flatMap ( Function < ? super T , ? extends Publisher < ? extends R > > mapper ) { } }

implementation: return flatMap ( mapper , Queues . SMALL_BUFFER_SIZE , Queues . XS_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Transform the elements emitted by this { @ link Flux } asynchronously into Publishers ,
 * then flatten these inner publishers into a single { @ link Flux } through merging ,
 * which allow them to interleave .
 * There are three dimensions to this operator that can be compared with
 * { @ link # flatMapSequential ( Function ) flatMapSequential } and { @ link # concatMap ( Function ) concatMap } :
 * < ul >
 * < li > < b > Generation of inners and subscription < / b > : this operator is eagerly
 * subscribing to its inners . < / li >
 * < li > < b > Ordering of the flattened values < / b > : this operator does not necessarily preserve
 * original ordering , as inner element are flattened as they arrive . < / li >
 * < li > < b > Interleaving < / b > : this operator lets values from different inners interleave
 * ( similar to merging the inner sequences ) . < / li >
 * < / ul >
 * The concurrency argument allows to control how many { @ link Publisher } can be
 * subscribed to and merged in parallel . In turn , that argument shows the size of
 * the first { @ link Subscription # request } to the upstream .
 * < img class = " marble " src = " doc - files / marbles / flatMapWithConcurrency . svg " alt = " " >
 * @ reactor . discard This operator discards elements internally queued for backpressure upon cancellation or error triggered by a data signal .
 * @ param mapper the { @ link Function } to transform input sequence into N sequences { @ link Publisher }
 * @ param concurrency the maximum number of in - flight inner sequences
 * @ param < V > the merged output sequence type
 * @ reactor . errorMode This operator supports { @ link # onErrorContinue ( BiConsumer ) resuming on errors }
 * in the mapper { @ link Function } . Exceptions thrown by the mapper then behave as if
 * it had mapped the value to an empty publisher . If the mapper does map to a scalar
 * publisher ( an optimization in which the value can be resolved immediately without
 * subscribing to the publisher , e . g . a { @ link Mono # fromCallable ( Callable ) } ) but said
 * publisher throws , this can be resumed from in the same manner .
 * @ return a new { @ link Flux } */
public final < V > Flux < V > flatMap ( Function < ? super T , ? extends Publisher < ? extends V > > mapper , int concurrency ) { } }

implementation: return flatMap ( mapper , concurrency , Queues . XS_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Transform the elements emitted by this { @ link Flux } asynchronously into Publishers ,
 * then flatten these inner publishers into a single { @ link Flux } through merging ,
 * which allow them to interleave .
 * There are three dimensions to this operator that can be compared with
 * { @ link # flatMapSequential ( Function ) flatMapSequential } and { @ link # concatMap ( Function ) concatMap } :
 * < ul >
 * < li > < b > Generation of inners and subscription < / b > : this operator is eagerly
 * subscribing to its inners . < / li >
 * < li > < b > Ordering of the flattened values < / b > : this operator does not necessarily preserve
 * original ordering , as inner element are flattened as they arrive . < / li >
 * < li > < b > Interleaving < / b > : this operator lets values from different inners interleave
 * ( similar to merging the inner sequences ) . < / li >
 * < / ul >
 * The concurrency argument allows to control how many { @ link Publisher } can be
 * subscribed to and merged in parallel . In turn , that argument shows the size of
 * the first { @ link Subscription # request } to the upstream .
 * The prefetch argument allows to give an arbitrary prefetch size to the merged
 * { @ link Publisher } ( in other words prefetch size means the size of the first
 * { @ link Subscription # request } to the merged { @ link Publisher } ) .
 * < img class = " marble " src = " doc - files / marbles / flatMapWithConcurrencyAndPrefetch . svg " alt = " " >
 * @ reactor . discard This operator discards elements internally queued for backpressure upon cancellation or error triggered by a data signal .
 * @ param mapper the { @ link Function } to transform input sequence into N sequences { @ link Publisher }
 * @ param concurrency the maximum number of in - flight inner sequences
 * @ param prefetch the maximum in - flight elements from each inner { @ link Publisher } sequence
 * @ param < V > the merged output sequence type
 * @ reactor . errorMode This operator supports { @ link # onErrorContinue ( BiConsumer ) resuming on errors }
 * in the mapper { @ link Function } . Exceptions thrown by the mapper then behave as if
 * it had mapped the value to an empty publisher . If the mapper does map to a scalar
 * publisher ( an optimization in which the value can be resolved immediately without
 * subscribing to the publisher , e . g . a { @ link Mono # fromCallable ( Callable ) } ) but said
 * publisher throws , this can be resumed from in the same manner .
 * @ return a merged { @ link Flux } */
public final < V > Flux < V > flatMap ( Function < ? super T , ? extends Publisher < ? extends V > > mapper , int concurrency , int prefetch ) { } }

implementation: return flatMap ( mapper , false , concurrency , prefetch ) ;


signature:
public class Flux { /** * Transform the elements emitted by this { @ link Flux } asynchronously into Publishers ,
 * then flatten these inner publishers into a single { @ link Flux } through merging ,
 * which allow them to interleave .
 * There are three dimensions to this operator that can be compared with
 * { @ link # flatMapSequential ( Function ) flatMapSequential } and { @ link # concatMap ( Function ) concatMap } :
 * < ul >
 * < li > < b > Generation of inners and subscription < / b > : this operator is eagerly
 * subscribing to its inners . < / li >
 * < li > < b > Ordering of the flattened values < / b > : this operator does not necessarily preserve
 * original ordering , as inner element are flattened as they arrive . < / li >
 * < li > < b > Interleaving < / b > : this operator lets values from different inners interleave
 * ( similar to merging the inner sequences ) . < / li >
 * < / ul >
 * The concurrency argument allows to control how many { @ link Publisher } can be
 * subscribed to and merged in parallel . The prefetch argument allows to give an
 * arbitrary prefetch size to the merged { @ link Publisher } . This variant will delay
 * any error until after the rest of the flatMap backlog has been processed .
 * < img class = " marble " src = " doc - files / marbles / flatMapWithConcurrencyAndPrefetch . svg " alt = " " >
 * @ reactor . discard This operator discards elements internally queued for backpressure upon cancellation or error triggered by a data signal .
 * @ param mapper the { @ link Function } to transform input sequence into N sequences { @ link Publisher }
 * @ param concurrency the maximum number of in - flight inner sequences
 * @ param prefetch the maximum in - flight elements from each inner { @ link Publisher } sequence
 * @ param < V > the merged output sequence type
 * @ reactor . errorMode This operator supports { @ link # onErrorContinue ( BiConsumer ) resuming on errors }
 * in the mapper { @ link Function } . Exceptions thrown by the mapper then behave as if
 * it had mapped the value to an empty publisher . If the mapper does map to a scalar
 * publisher ( an optimization in which the value can be resolved immediately without
 * subscribing to the publisher , e . g . a { @ link Mono # fromCallable ( Callable ) } ) but said
 * publisher throws , this can be resumed from in the same manner .
 * @ return a merged { @ link Flux } */
public final < V > Flux < V > flatMapDelayError ( Function < ? super T , ? extends Publisher < ? extends V > > mapper , int concurrency , int prefetch ) { } }

implementation: return flatMap ( mapper , true , concurrency , prefetch ) ;


signature:
public class Flux { /** * Transform the signals emitted by this { @ link Flux } asynchronously into Publishers ,
 * then flatten these inner publishers into a single { @ link Flux } through merging ,
 * which allow them to interleave . Note that at least one of the signal mappers must
 * be provided , and all provided mappers must produce a publisher .
 * There are three dimensions to this operator that can be compared with
 * { @ link # flatMapSequential ( Function ) flatMapSequential } and { @ link # concatMap ( Function ) concatMap } :
 * < ul >
 * < li > < b > Generation of inners and subscription < / b > : this operator is eagerly
 * subscribing to its inners . < / li >
 * < li > < b > Ordering of the flattened values < / b > : this operator does not necessarily preserve
 * original ordering , as inner element are flattened as they arrive . < / li >
 * < li > < b > Interleaving < / b > : this operator lets values from different inners interleave
 * ( similar to merging the inner sequences ) . < / li >
 * < / ul >
 * OnError will be transformed into completion signal after its mapping callback has been applied .
 * < img class = " marble " src = " doc - files / marbles / flatMapWithMappersOnTerminalEventsForFlux . svg " alt = " " >
 * @ param mapperOnNext the { @ link Function } to call on next data and returning a sequence to merge .
 * Use { @ literal null } to ignore ( provided at least one other mapper is specified ) .
 * @ param mapperOnError the { @ link Function } to call on error signal and returning a sequence to merge .
 * Use { @ literal null } to ignore ( provided at least one other mapper is specified ) .
 * @ param mapperOnComplete the { @ link Function } to call on complete signal and returning a sequence to merge .
 * Use { @ literal null } to ignore ( provided at least one other mapper is specified ) .
 * @ param < R > the output { @ link Publisher } type target
 * @ return a new { @ link Flux } */
public final < R > Flux < R > flatMap ( @ Nullable Function < ? super T , ? extends Publisher < ? extends R > > mapperOnNext , @ Nullable Function < ? super Throwable , ? extends Publisher < ? extends R > > mapperOnError , @ Nullable Supplier < ? extends Publisher < ? extends R > > mapperOnComplete ) { } }

implementation: return onAssembly ( new FluxFlatMap < > ( new FluxMapSignal < > ( this , mapperOnNext , mapperOnError , mapperOnComplete ) , identityFunction ( ) , false , Queues . XS_BUFFER_SIZE , Queues . xs ( ) , Queues . XS_BUFFER_SIZE , Queues . xs ( ) ) ) ;


signature:
public class Flux { /** * Transform the items emitted by this { @ link Flux } into { @ link Iterable } , then flatten the elements from those by
 * merging them into a single { @ link Flux } .
 * < img class = " marble " src = " doc - files / marbles / flatMapIterableForFlux . svg " alt = " " >
 * Note that unlike { @ link # flatMap ( Function ) } and { @ link # concatMap ( Function ) } , with Iterable there is
 * no notion of eager vs lazy inner subscription . The content of the Iterables are all played sequentially .
 * Thus { @ code flatMapIterable } and { @ code concatMapIterable } are equivalent offered as a discoverability
 * improvement for users that explore the API with the concat vs flatMap expectation .
 * @ reactor . discard This operator discards elements internally queued for backpressure upon cancellation or error triggered by a data signal .
 * @ param mapper the { @ link Function } to transform input sequence into N { @ link Iterable }
 * @ param < R > the merged output sequence type
 * @ return a concatenation of the values from the Iterables obtained from each element in this { @ link Flux } */
public final < R > Flux < R > flatMapIterable ( Function < ? super T , ? extends Iterable < ? extends R > > mapper ) { } }

implementation: return flatMapIterable ( mapper , Queues . SMALL_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Transform the elements emitted by this { @ link Flux } asynchronously into Publishers ,
 * then flatten these inner publishers into a single { @ link Flux } , but merge them in
 * the order of their source element .
 * There are three dimensions to this operator that can be compared with
 * { @ link # flatMap ( Function ) flatMap } and { @ link # concatMap ( Function ) concatMap } :
 * < ul >
 * < li > < b > Generation of inners and subscription < / b > : this operator is eagerly
 * subscribing to its inners ( like flatMap ) . < / li >
 * < li > < b > Ordering of the flattened values < / b > : this operator queues elements from
 * late inners until all elements from earlier inners have been emitted , thus emitting
 * inner sequences as a whole , in an order that matches their source ' s order . < / li >
 * < li > < b > Interleaving < / b > : this operator does not let values from different inners
 * interleave ( similar looking result to concatMap , but due to queueing of values
 * that would have been interleaved otherwise ) . < / li >
 * < / ul >
 * That is to say , whenever a source element is emitted it is transformed to an inner
 * { @ link Publisher } . However , if such an early inner takes more time to complete than
 * subsequent faster inners , the data from these faster inners will be queued until
 * the earlier inner completes , so as to maintain source ordering .
 * < img class = " marble " src = " doc - files / marbles / flatMapSequential . svg " alt = " " >
 * @ param mapper the { @ link Function } to transform input sequence into N sequences { @ link Publisher }
 * @ param < R > the merged output sequence type
 * @ return a merged { @ link Flux } , subscribing early but keeping the original ordering */
public final < R > Flux < R > flatMapSequential ( Function < ? super T , ? extends Publisher < ? extends R > > mapper ) { } }

implementation: return flatMapSequential ( mapper , Queues . SMALL_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Transform the elements emitted by this { @ link Flux } asynchronously into Publishers ,
 * then flatten these inner publishers into a single { @ link Flux } , but merge them in
 * the order of their source element .
 * There are three dimensions to this operator that can be compared with
 * { @ link # flatMap ( Function ) flatMap } and { @ link # concatMap ( Function ) concatMap } :
 * < ul >
 * < li > < b > Generation of inners and subscription < / b > : this operator is eagerly
 * subscribing to its inners ( like flatMap ) . < / li >
 * < li > < b > Ordering of the flattened values < / b > : this operator queues elements from
 * late inners until all elements from earlier inners have been emitted , thus emitting
 * inner sequences as a whole , in an order that matches their source ' s order . < / li >
 * < li > < b > Interleaving < / b > : this operator does not let values from different inners
 * interleave ( similar looking result to concatMap , but due to queueing of values
 * that would have been interleaved otherwise ) . < / li >
 * < / ul >
 * That is to say , whenever a source element is emitted it is transformed to an inner
 * { @ link Publisher } . However , if such an early inner takes more time to complete than
 * subsequent faster inners , the data from these faster inners will be queued until
 * the earlier inner completes , so as to maintain source ordering .
 * The concurrency argument allows to control how many merged { @ link Publisher } can happen in parallel .
 * < img class = " marble " src = " doc - files / marbles / flatMapSequentialWithConcurrency . svg " alt = " " >
 * @ param mapper the { @ link Function } to transform input sequence into N sequences { @ link Publisher }
 * @ param maxConcurrency the maximum number of in - flight inner sequences
 * @ param < R > the merged output sequence type
 * @ return a merged { @ link Flux } , subscribing early but keeping the original ordering */
public final < R > Flux < R > flatMapSequential ( Function < ? super T , ? extends Publisher < ? extends R > > mapper , int maxConcurrency ) { } }

implementation: return flatMapSequential ( mapper , maxConcurrency , Queues . XS_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Transform the elements emitted by this { @ link Flux } asynchronously into Publishers ,
 * then flatten these inner publishers into a single { @ link Flux } , but merge them in
 * the order of their source element .
 * There are three dimensions to this operator that can be compared with
 * { @ link # flatMap ( Function ) flatMap } and { @ link # concatMap ( Function ) concatMap } :
 * < ul >
 * < li > < b > Generation of inners and subscription < / b > : this operator is eagerly
 * subscribing to its inners ( like flatMap ) . < / li >
 * < li > < b > Ordering of the flattened values < / b > : this operator queues elements from
 * late inners until all elements from earlier inners have been emitted , thus emitting
 * inner sequences as a whole , in an order that matches their source ' s order . < / li >
 * < li > < b > Interleaving < / b > : this operator does not let values from different inners
 * interleave ( similar looking result to concatMap , but due to queueing of values
 * that would have been interleaved otherwise ) . < / li >
 * < / ul >
 * That is to say , whenever a source element is emitted it is transformed to an inner
 * { @ link Publisher } . However , if such an early inner takes more time to complete than
 * subsequent faster inners , the data from these faster inners will be queued until
 * the earlier inner completes , so as to maintain source ordering .
 * The concurrency argument allows to control how many merged { @ link Publisher }
 * can happen in parallel . The prefetch argument allows to give an arbitrary prefetch
 * size to the merged { @ link Publisher } .
 * < img class = " marble " src = " doc - files / marbles / flatMapSequentialWithConcurrencyAndPrefetch . svg " alt = " " >
 * @ param mapper the { @ link Function } to transform input sequence into N sequences { @ link Publisher }
 * @ param maxConcurrency the maximum number of in - flight inner sequences
 * @ param prefetch the maximum in - flight elements from each inner { @ link Publisher } sequence
 * @ param < R > the merged output sequence type
 * @ return a merged { @ link Flux } , subscribing early but keeping the original ordering */
public final < R > Flux < R > flatMapSequential ( Function < ? super T , ? extends Publisher < ? extends R > > mapper , int maxConcurrency , int prefetch ) { } }

implementation: return flatMapSequential ( mapper , false , maxConcurrency , prefetch ) ;


signature:
public class Flux { /** * Transform the elements emitted by this { @ link Flux } asynchronously into Publishers ,
 * then flatten these inner publishers into a single { @ link Flux } , but merge them in
 * the order of their source element .
 * There are three dimensions to this operator that can be compared with
 * { @ link # flatMap ( Function ) flatMap } and { @ link # concatMap ( Function ) concatMap } :
 * < ul >
 * < li > < b > Generation of inners and subscription < / b > : this operator is eagerly
 * subscribing to its inners ( like flatMap ) . < / li >
 * < li > < b > Ordering of the flattened values < / b > : this operator queues elements from
 * late inners until all elements from earlier inners have been emitted , thus emitting
 * inner sequences as a whole , in an order that matches their source ' s order . < / li >
 * < li > < b > Interleaving < / b > : this operator does not let values from different inners
 * interleave ( similar looking result to concatMap , but due to queueing of values
 * that would have been interleaved otherwise ) . < / li >
 * < / ul >
 * That is to say , whenever a source element is emitted it is transformed to an inner
 * { @ link Publisher } . However , if such an early inner takes more time to complete than
 * subsequent faster inners , the data from these faster inners will be queued until
 * the earlier inner completes , so as to maintain source ordering .
 * The concurrency argument allows to control how many merged { @ link Publisher }
 * can happen in parallel . The prefetch argument allows to give an arbitrary prefetch
 * size to the merged { @ link Publisher } . This variant will delay any error until after the
 * rest of the flatMap backlog has been processed .
 * < img class = " marble " src = " doc - files / marbles / flatMapSequentialWithConcurrencyAndPrefetch . svg " alt = " " >
 * @ param mapper the { @ link Function } to transform input sequence into N sequences { @ link Publisher }
 * @ param maxConcurrency the maximum number of in - flight inner sequences
 * @ param prefetch the maximum in - flight elements from each inner { @ link Publisher } sequence
 * @ param < R > the merged output sequence type
 * @ return a merged { @ link Flux } , subscribing early but keeping the original ordering */
public final < R > Flux < R > flatMapSequentialDelayError ( Function < ? super T , ? extends Publisher < ? extends R > > mapper , int maxConcurrency , int prefetch ) { } }

implementation: return flatMapSequential ( mapper , true , maxConcurrency , prefetch ) ;


signature:
public class Flux { /** * Divide this sequence into dynamically created { @ link Flux } ( or groups ) for each
 * unique key , as produced by the provided keyMapper { @ link Function } . Note that
 * groupBy works best with a low cardinality of groups , so chose your keyMapper
 * function accordingly .
 * < img class = " marble " src = " doc - files / marbles / groupByWithKeyMapper . svg " alt = " " >
 * The groups need to be drained and consumed downstream for groupBy to work correctly .
 * Notably when the criteria produces a large amount of groups , it can lead to hanging
 * if the groups are not suitably consumed downstream ( eg . due to a { @ code flatMap }
 * with a { @ code maxConcurrency } parameter that is set too low ) .
 * @ param keyMapper the key mapping { @ link Function } that evaluates an incoming data and returns a key .
 * @ param < K > the key type extracted from each value of this sequence
 * @ return a { @ link Flux } of { @ link GroupedFlux } grouped sequences */
public final < K > Flux < GroupedFlux < K , T > > groupBy ( Function < ? super T , ? extends K > keyMapper ) { } }

implementation: return groupBy ( keyMapper , identityFunction ( ) ) ;


signature:
public class Flux { /** * Divide this sequence into dynamically created { @ link Flux } ( or groups ) for each
 * unique key , as produced by the provided keyMapper { @ link Function } . Source elements
 * are also mapped to a different value using the { @ code valueMapper } . Note that
 * groupBy works best with a low cardinality of groups , so chose your keyMapper
 * function accordingly .
 * < img class = " marble " src = " doc - files / marbles / groupByWithKeyMapperAndValueMapper . svg " alt = " " >
 * The groups need to be drained and consumed downstream for groupBy to work correctly .
 * Notably when the criteria produces a large amount of groups , it can lead to hanging
 * if the groups are not suitably consumed downstream ( eg . due to a { @ code flatMap }
 * with a { @ code maxConcurrency } parameter that is set too low ) .
 * @ param keyMapper the key mapping function that evaluates an incoming data and returns a key .
 * @ param valueMapper the value mapping function that evaluates which data to extract for re - routing .
 * @ param < K > the key type extracted from each value of this sequence
 * @ param < V > the value type extracted from each value of this sequence
 * @ return a { @ link Flux } of { @ link GroupedFlux } grouped sequences */
public final < K , V > Flux < GroupedFlux < K , V > > groupBy ( Function < ? super T , ? extends K > keyMapper , Function < ? super T , ? extends V > valueMapper ) { } }

implementation: return groupBy ( keyMapper , valueMapper , Queues . SMALL_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Divide this sequence into dynamically created { @ link Flux } ( or groups ) for each
 * unique key , as produced by the provided keyMapper { @ link Function } . Source elements
 * are also mapped to a different value using the { @ code valueMapper } . Note that
 * groupBy works best with a low cardinality of groups , so chose your keyMapper
 * function accordingly .
 * < img class = " marble " src = " doc - files / marbles / groupByWithKeyMapperAndValueMapper . svg " alt = " " >
 * The groups need to be drained and consumed downstream for groupBy to work correctly .
 * Notably when the criteria produces a large amount of groups , it can lead to hanging
 * if the groups are not suitably consumed downstream ( eg . due to a { @ code flatMap }
 * with a { @ code maxConcurrency } parameter that is set too low ) .
 * @ param keyMapper the key mapping function that evaluates an incoming data and returns a key .
 * @ param valueMapper the value mapping function that evaluates which data to extract for re - routing .
 * @ param prefetch the number of values to prefetch from the source
 * @ param < K > the key type extracted from each value of this sequence
 * @ param < V > the value type extracted from each value of this sequence
 * @ return a { @ link Flux } of { @ link GroupedFlux } grouped sequences */
public final < K , V > Flux < GroupedFlux < K , V > > groupBy ( Function < ? super T , ? extends K > keyMapper , Function < ? super T , ? extends V > valueMapper , int prefetch ) { } }

implementation: return onAssembly ( new FluxGroupBy < > ( this , keyMapper , valueMapper , Queues . unbounded ( prefetch ) , Queues . unbounded ( prefetch ) , prefetch ) ) ;


signature:
public class Flux { /** * Map values from two Publishers into time windows and emit combination of values
 * in case their windows overlap . The emitted elements are obtained by passing the
 * value from this { @ link Flux } and a { @ link Flux } emitting the value from the other
 * { @ link Publisher } to a { @ link BiFunction } .
 * There are no guarantees in what order the items get combined when multiple items from
 * one or both source Publishers overlap .
 * Unlike { @ link Flux # join } , items from the second { @ link Publisher } will be provided
 * as a { @ link Flux } to the { @ code resultSelector } .
 * < img class = " marble " src = " doc - files / marbles / groupJoin . svg " alt = " " >
 * @ param other the other { @ link Publisher } to correlate items with
 * @ param leftEnd a function that returns a Publisher whose emissions indicate the
 * time window for the source value to be considered
 * @ param rightEnd a function that returns a Publisher whose emissions indicate the
 * time window for the { @ code right } Publisher value to be considered
 * @ param resultSelector a function that takes an item emitted by this { @ link Flux } and
 * a { @ link Flux } representation of the overlapping item from the other { @ link Publisher }
 * and returns the value to be emitted by the resulting { @ link Flux }
 * @ param < TRight > the type of the elements from the right { @ link Publisher }
 * @ param < TLeftEnd > the type for this { @ link Flux } window signals
 * @ param < TRightEnd > the type for the right { @ link Publisher } window signals
 * @ param < R > the combined result type
 * @ return a joining { @ link Flux }
 * @ see # join ( Publisher , Function , Function , BiFunction ) */
public final < TRight , TLeftEnd , TRightEnd , R > Flux < R > groupJoin ( Publisher < ? extends TRight > other , Function < ? super T , ? extends Publisher < TLeftEnd > > leftEnd , Function < ? super TRight , ? extends Publisher < TRightEnd > > rightEnd , BiFunction < ? super T , ? super Flux < TRight > , ? extends R > resultSelector ) { } }

implementation: return onAssembly ( new FluxGroupJoin < T , TRight , TLeftEnd , TRightEnd , R > ( this , other , leftEnd , rightEnd , resultSelector , Queues . unbounded ( Queues . XS_BUFFER_SIZE ) , Queues . unbounded ( Queues . XS_BUFFER_SIZE ) ) ) ;


signature:
public class Flux { /** * Handle the items emitted by this { @ link Flux } 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
 * @ reactor . errorMode This operator supports { @ link # onErrorContinue ( BiConsumer ) resuming on errors } ( including when
 * fusion is enabled ) when the { @ link BiConsumer } throws an exception or if an error is signaled explicitly via
 * { @ link SynchronousSink # error ( Throwable ) } .
 * @ return a transformed { @ link Flux } */
public final < R > Flux < R > handle ( BiConsumer < ? super T , SynchronousSink < R > > handler ) { } }

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


signature:
public class Flux { /** * Emit a single boolean true if any of the elements of this { @ link Flux } sequence is
 * equal to the provided value .
 * The implementation uses short - circuit logic and completes with true if
 * an element matches the value .
 * < img class = " marble " src = " doc - files / marbles / hasElementForFlux . svg " alt = " " >
 * @ param value constant compared to incoming signals
 * @ return a new { @ link Flux } with < code > true < / code > if any element is equal to a given value and < code > false < / code >
 * otherwise */
public final Mono < Boolean > hasElement ( T value ) { } }

implementation: Objects . requireNonNull ( value , "value" ) ; return any ( t -> Objects . equals ( value , t ) ) ;


signature:
public class Flux { /** * Keep information about the order in which source values were received by
 * indexing them internally with a 0 - based incrementing long then combining this
 * information with the source value into a { @ code I } using the provided { @ link BiFunction } ,
 * returning a { @ link Flux Flux & lt ; I & gt ; } .
 * Typical usage would be to produce a { @ link Tuple2 } similar to { @ link # index ( ) } , but
 * 1 - based instead of 0 - based :
 * { @ code index ( ( i , v ) - > Tuples . of ( i + 1 , v ) ) }
 * < img class = " marble " src = " doc - files / marbles / indexWithMapper . svg " alt = " " >
 * @ param indexMapper the { @ link BiFunction } to use to combine elements and their index .
 * @ return an indexed { @ link Flux } with each source value combined with its computed index . */
public final < I > Flux < I > index ( BiFunction < ? super Long , ? super T , ? extends I > indexMapper ) { } }

implementation: if ( this instanceof Fuseable ) { return onAssembly ( new FluxIndexFuseable < > ( this , indexMapper ) ) ; } return onAssembly ( new FluxIndex < > ( this , indexMapper ) ) ;


signature:
public class Flux { /** * Map values from two Publishers into time windows and emit combination of values
 * in case their windows overlap . The emitted elements are obtained by passing the
 * values from this { @ link Flux } and the other { @ link Publisher } to a { @ link BiFunction } .
 * There are no guarantees in what order the items get combined when multiple items from
 * one or both source Publishers overlap .
 * < img class = " marble " src = " doc - files / marbles / join . svg " alt = " " >
 * @ param other the other { @ link Publisher } to correlate items with
 * @ param leftEnd a function that returns a Publisher whose emissions indicate the
 * time window for the source value to be considered
 * @ param rightEnd a function that returns a Publisher whose emissions indicate the
 * time window for the { @ code right } Publisher value to be considered
 * @ param resultSelector a function that takes an item emitted by each Publisher and returns the
 * value to be emitted by the resulting { @ link Flux }
 * @ param < TRight > the type of the elements from the right { @ link Publisher }
 * @ param < TLeftEnd > the type for this { @ link Flux } window signals
 * @ param < TRightEnd > the type for the right { @ link Publisher } window signals
 * @ param < R > the combined result type
 * @ return a joining { @ link Flux }
 * @ see # groupJoin ( Publisher , Function , Function , BiFunction ) */
public final < TRight , TLeftEnd , TRightEnd , R > Flux < R > join ( Publisher < ? extends TRight > other , Function < ? super T , ? extends Publisher < TLeftEnd > > leftEnd , Function < ? super TRight , ? extends Publisher < TRightEnd > > rightEnd , BiFunction < ? super T , ? super TRight , ? extends R > resultSelector ) { } }

implementation: return onAssembly ( new FluxJoin < T , TRight , TLeftEnd , TRightEnd , R > ( this , other , leftEnd , rightEnd , resultSelector ) ) ;


signature:
public class Flux { /** * Emit the last element observed before complete signal as a { @ link Mono } , or emit
 * { @ link NoSuchElementException } error if the source was empty .
 * For a passive version use { @ link # takeLast ( int ) }
 * < img class = " marble " src = " doc - files / marbles / last . svg " alt = " " >
 * @ reactor . discard This operator discards elements before the last .
 * @ return a { @ link Mono } with the last value in this { @ link Flux } */
public final Mono < T > last ( ) { } }

implementation: if ( this instanceof Callable ) { @ SuppressWarnings ( "unchecked" ) Callable < T > thiz = ( Callable < T > ) this ; Mono < T > callableMono = convertToMono ( thiz ) ; if ( callableMono == Mono . empty ( ) ) { return Mono . error ( new NoSuchElementException ( "Flux#last() didn't observe any onNext signal from Callable flux" ) ) ; } return callableMono ; } return Mono . onAssembly ( new MonoTakeLastOne < > ( this ) ) ;


signature:
public class Flux { /** * Emit the last element observed before complete signal as a { @ link Mono } , or emit
 * the { @ code defaultValue } if the source was empty .
 * For a passive version use { @ link # takeLast ( int ) }
 * < img class = " marble " src = " doc - files / marbles / lastWithDefault . svg " alt = " " >
 * @ reactor . discard This operator discards elements before the last .
 * @ param defaultValue a single fallback item if this { @ link Flux } is empty
 * @ return a { @ link Mono } with the last value in this { @ link Flux } */
public final Mono < T > last ( T defaultValue ) { } }

implementation: if ( this instanceof Callable ) { @ SuppressWarnings ( "unchecked" ) Callable < T > thiz = ( Callable < T > ) this ; if ( thiz instanceof Fuseable . ScalarCallable ) { @ SuppressWarnings ( "unchecked" ) Fuseable . ScalarCallable < T > c = ( Fuseable . ScalarCallable < T > ) thiz ; T v ; try { v = c . call ( ) ; } catch ( Exception e ) { return Mono . error ( e ) ; } if ( v == null ) { return Mono . just ( defaultValue ) ; } return Mono . just ( v ) ; } Mono . onAssembly ( new MonoCallable < > ( thiz ) ) ; } return Mono . onAssembly ( new MonoTakeLastOne < > ( this , defaultValue ) ) ;


signature:
public class Flux { /** * Ensure that backpressure signals from downstream subscribers are split into batches
 * capped at the provided { @ code highTide } first , then replenishing at the provided
 * { @ code lowTide } , effectively rate limiting the upstream { @ link Publisher } .
 * Note that this is an upper bound , and that this operator uses a prefetch - and - replenish
 * strategy , requesting a replenishing amount when 75 % of the prefetch amount has been
 * emitted .
 * Typically used for scenarios where consumer ( s ) request a large amount of data
 * ( eg . { @ code Long . MAX _ VALUE } ) but the data source behaves better or can be optimized
 * with smaller requests ( eg . database paging , etc . . . ) . All data is still processed ,
 * unlike with { @ link # limitRequest ( long ) } which will cap the grand total request
 * amount .
 * Similar to { @ code flux . publishOn ( Schedulers . immediate ( ) , prefetchRate ) . subscribe ( ) } ,
 * except with a customized " low tide " instead of the default 75 % .
 * Note that the smaller the lowTide is , the higher the potential for concurrency
 * between request and data production . And thus the more extraneous replenishment
 * requests this operator could make . For example , for a global downstream
 * request of 14 , with a highTide of 10 and a lowTide of 2 , the operator would perform
 * 7 low tide requests , whereas with the default lowTide of 8 it would only perform one .
 * Using a { @ code lowTide } equal to { @ code highTide } reverts to the default 75 % strategy ,
 * while using a { @ code lowTide } of { @ literal 0 } disables the lowTide , resulting in
 * all requests strictly adhering to the highTide .
 * @ param highTide the initial request amount
 * @ param lowTide the subsequent ( or replenishing ) request amount , { @ literal 0 } to
 * disable early replenishing , { @ literal highTide } to revert to a 75 % replenish strategy .
 * @ return a { @ link Flux } limiting downstream ' s backpressure and customizing the
 * replenishment request amount
 * @ see # publishOn ( Scheduler , int )
 * @ see # limitRequest ( long ) */
public final Flux < T > limitRate ( int highTide , int lowTide ) { } }

implementation: return onAssembly ( this . publishOn ( Schedulers . immediate ( ) , true , highTide , lowTide ) ) ;


signature:
public class Flux { /** * Observe Reactive Streams signals matching the passed filter { @ code options } and
 * trace them using { @ link Logger } support . Default will use { @ link Level # INFO } and
 * { @ code 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 >
 * flux . log ( " category " , Level . INFO , SignalType . ON _ NEXT , SignalType . ON _ ERROR )
 * < / pre >
 * < img class = " marble " src = " doc - files / marbles / logForFlux . 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 be added , e . g . " reactor . Flux . Map " .
 * @ 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 Flux } that logs signals */
public final Flux < 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 FluxLogFuseable < > ( this , log ) ) ; } return onAssembly ( new FluxLog < > ( this , log ) ) ;


signature:
public class Flux { /** * 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 / logForFlux . svg " alt = " " >
 * @ param logger the { @ link Logger } to use , instead of resolving one through a category .
 * @ return a new { @ link Flux } that logs signals */
public final Flux < T > log ( Logger logger ) { } }

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


signature:
public class Flux { /** * 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 / logForFlux . 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 ( default in overload is false ) .
 * @ param options a vararg { @ link SignalType } option to filter log messages
 * @ return a new { @ link Flux } that logs signals */
public final Flux < 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 FluxLogFuseable < > ( this , log ) ) ; } return onAssembly ( new FluxLog < > ( this , log ) ) ;


signature:
public class Flux { /** * Transform the items emitted by this { @ link Flux } by applying a synchronous function
 * to each item .
 * < img class = " marble " src = " doc - files / marbles / mapForFlux . svg " alt = " " >
 * @ param mapper the synchronous transforming { @ link Function }
 * @ param < V > the transformed type
 * @ reactor . errorMode This operator supports { @ link # onErrorContinue ( BiConsumer ) resuming on errors }
 * ( including when fusion is enabled ) . Exceptions thrown by the mapper then cause the
 * source value to be dropped and a new element ( { @ code request ( 1 ) } ) being requested
 * from upstream .
 * @ return a transformed { @ link Flux } */
public final < V > Flux < V > map ( Function < ? super T , ? extends V > mapper ) { } }

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


signature:
public class Flux { /** * Merge data from this { @ link Flux } and a { @ link Publisher } into a reordered merge
 * sequence , by picking the smallest value from each sequence as defined by a provided
 * { @ link Comparator } . Note that subsequent calls are combined , and their comparators are
 * in lexicographic order as defined by { @ link Comparator # thenComparing ( Comparator ) } .
 * The combination step is avoided if the two { @ link Comparator Comparators } are
 * { @ link Comparator # equals ( Object ) equal } ( which can easily be achieved by using the
 * same reference , and is also always true of { @ link Comparator # naturalOrder ( ) } ) .
 * Note that merge is tailored to work with asynchronous sources or finite sources . When dealing with
 * an infinite source that doesn ' t already publish on a dedicated Scheduler , you must isolate that source
 * in its own Scheduler , as merge would otherwise attempt to drain it before subscribing to
 * another source .
 * < img class = " marble " src = " doc - files / marbles / mergeOrderedWith . svg " alt = " " >
 * @ param other the { @ link Publisher } to merge with
 * @ param otherComparator the { @ link Comparator } to use for merging
 * @ return a new { @ link Flux } */
public final Flux < T > mergeOrderedWith ( Publisher < ? extends T > other , Comparator < ? super T > otherComparator ) { } }

implementation: if ( this instanceof FluxMergeOrdered ) { FluxMergeOrdered < T > fluxMerge = ( FluxMergeOrdered < T > ) this ; return fluxMerge . mergeAdditionalSource ( other , otherComparator ) ; } return mergeOrdered ( otherComparator , this , other ) ;


signature:
public class Flux { /** * Merge data from this { @ link Flux } and a { @ link Publisher } into an interleaved merged
 * sequence . Unlike { @ link # concatWith ( Publisher ) concat } , inner sources are subscribed
 * to eagerly .
 * < img class = " marble " src = " doc - files / marbles / mergeWithForFlux . svg " alt = " " >
 * Note that merge is tailored to work with asynchronous sources or finite sources . When dealing with
 * an infinite source that doesn ' t already publish on a dedicated Scheduler , you must isolate that source
 * in its own Scheduler , as merge would otherwise attempt to drain it before subscribing to
 * another source .
 * @ param other the { @ link Publisher } to merge with
 * @ return a new { @ link Flux } */
public final Flux < T > mergeWith ( Publisher < ? extends T > other ) { } }

implementation: if ( this instanceof FluxMerge ) { FluxMerge < T > fluxMerge = ( FluxMerge < T > ) this ; return fluxMerge . mergeAdditionalSource ( other , Queues :: get ) ; } return merge ( this , other ) ;


signature:
public class Flux { /** * 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 Flux } */
public final Flux < T > metrics ( ) { } }

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


signature:
public class Flux { /** * Emit only the first item emitted by this { @ link Flux } , into a new { @ link Mono } .
 * < img class = " marble " src = " doc - files / marbles / next . svg " alt = " " >
 * @ return a new { @ link Mono } emitting the first value in this { @ link Flux } */
public final Mono < T > next ( ) { } }

implementation: if ( this instanceof Callable ) { @ SuppressWarnings ( "unchecked" ) Callable < T > m = ( Callable < T > ) this ; return convertToMono ( m ) ; } return Mono . onAssembly ( new MonoNext < > ( this ) ) ;


signature:
public class Flux { /** * Request an unbounded demand and push to the returned { @ link Flux } , or park up to
 * { @ code maxSize } elements when not enough demand is requested downstream .
 * The first element past this buffer to arrive out of sync with the downstream
 * subscriber ' s demand ( the " overflowing " element ) is immediately passed to a
 * { @ link Consumer } and the source is cancelled .
 * The { @ link Flux } is going to terminate with an overflow error , but this error is
 * delayed , which lets the subscriber make more requests for the content of the buffer .
 * Note that should the cancelled source produce further overflowing elements , these
 * would be passed to the { @ link Hooks # onNextDropped ( Consumer ) onNextDropped hook } .
 * < img class = " marble " src = " doc - files / marbles / onBackpressureBufferWithMaxSizeConsumer . svg " alt = " " >
 * @ reactor . discard This operator discards the buffered overflow elements upon cancellation or error triggered by a data signal ,
 * as well as elements that are rejected by the buffer due to { @ code maxSize } ( even though
 * they are passed to the { @ code onOverflow } { @ link Consumer } first ) .
 * @ param maxSize maximum buffer backlog size before overflow callback is called and source is cancelled
 * @ param onOverflow callback to invoke on overflow
 * @ return a backpressured { @ link Flux } that buffers with a bounded capacity */
public final Flux < T > onBackpressureBuffer ( int maxSize , Consumer < ? super T > onOverflow ) { } }

implementation: Objects . requireNonNull ( onOverflow , "onOverflow" ) ; return onAssembly ( new FluxOnBackpressureBuffer < > ( this , maxSize , false , onOverflow ) ) ;


signature:
public class Flux { /** * Request an unbounded demand and push to the returned { @ link Flux } , or park the observed
 * elements if not enough demand is requested downstream , within a { @ code maxSize }
 * limit . Over that limit , the overflow strategy is applied ( see { @ link BufferOverflowStrategy } ) .
 * Note that for the { @ link BufferOverflowStrategy # ERROR ERROR } strategy , the overflow
 * error will be delayed after the current backlog is consumed .
 * < img class = " marble " src = " doc - files / marbles / onBackpressureBufferWithMaxSizeStrategyDropOldest . svg " alt = " " >
 * @ reactor . discard This operator discards the buffered overflow elements upon cancellation or error triggered by a data signal ,
 * as well as elements that are rejected by the buffer due to { @ code maxSize } ( even though
 * they are passed to the { @ code bufferOverflowStrategy } first ) .
 * @ param maxSize maximum buffer backlog size before overflow strategy is applied
 * @ param bufferOverflowStrategy strategy to apply to overflowing elements
 * @ return a backpressured { @ link Flux } that buffers up to a capacity then applies an
 * overflow strategy */
public final Flux < T > onBackpressureBuffer ( int maxSize , BufferOverflowStrategy bufferOverflowStrategy ) { } }

implementation: Objects . requireNonNull ( bufferOverflowStrategy , "bufferOverflowStrategy" ) ; return onAssembly ( new FluxOnBackpressureBufferStrategy < > ( this , maxSize , null , bufferOverflowStrategy ) ) ;


signature:
public class Flux { /** * Request an unbounded demand and push to the returned { @ link Flux } , or park the observed
 * elements if not enough demand is requested downstream , within a { @ code maxSize }
 * limit . Over that limit , the overflow strategy is applied ( see { @ link BufferOverflowStrategy } ) .
 * A { @ link Consumer } is immediately invoked when there is an overflow , receiving the
 * value that was discarded because of the overflow ( which can be different from the
 * latest element emitted by the source in case of a
 * { @ link BufferOverflowStrategy # DROP _ LATEST DROP _ LATEST } strategy ) .
 * Note that for the { @ link BufferOverflowStrategy # ERROR ERROR } strategy , the overflow
 * error will be delayed after the current backlog is consumed . The consumer is still
 * invoked immediately .
 * < img class = " marble " src = " doc - files / marbles / onBackpressureBufferWithMaxSizeStrategyDropOldest . svg " alt = " " >
 * @ reactor . discard This operator discards the buffered overflow elements upon cancellation or error triggered by a data signal ,
 * as well as elements that are rejected by the buffer due to { @ code maxSize } ( even though
 * they are passed to the { @ code onOverflow } { @ link Consumer } AND the { @ code bufferOverflowStrategy } first ) .
 * @ param maxSize maximum buffer backlog size before overflow callback is called
 * @ param onBufferOverflow callback to invoke on overflow
 * @ param bufferOverflowStrategy strategy to apply to overflowing elements
 * @ return a backpressured { @ link Flux } that buffers up to a capacity then applies an
 * overflow strategy */
public final Flux < T > onBackpressureBuffer ( int maxSize , Consumer < ? super T > onBufferOverflow , BufferOverflowStrategy bufferOverflowStrategy ) { } }

implementation: Objects . requireNonNull ( onBufferOverflow , "onBufferOverflow" ) ; Objects . requireNonNull ( bufferOverflowStrategy , "bufferOverflowStrategy" ) ; return onAssembly ( new FluxOnBackpressureBufferStrategy < > ( this , maxSize , onBufferOverflow , bufferOverflowStrategy ) ) ;


signature:
public class Flux { /** * Request an unbounded demand and push to the returned { @ link Flux } , or park the observed
 * elements if not enough demand is requested downstream , within a { @ code maxSize }
 * limit and for a maximum { @ link Duration } of { @ code ttl } ( as measured on the
 * { @ link Schedulers # elastic ( ) elastic Scheduler } ) . Over that limit , oldest
 * elements from the source are dropped .
 * Elements evicted based on the TTL are passed to a cleanup { @ link Consumer } , which
 * is also immediately invoked when there is an overflow .
 * < img class = " marble " src = " doc - files / marbles / onBackpressureBufferWithDurationAndMaxSize . svg " alt = " " >
 * @ reactor . discard This operator discards its internal buffer of elements that overflow ,
 * after having applied the { @ code onBufferEviction } handler .
 * @ param ttl maximum { @ link Duration } for which an element is kept in the backlog
 * @ param maxSize maximum buffer backlog size before overflow callback is called
 * @ param onBufferEviction callback to invoke once TTL is reached or on overflow
 * @ return a backpressured { @ link Flux } that buffers with a TTL and up to a capacity then applies an
 * overflow strategy */
public final Flux < T > onBackpressureBuffer ( Duration ttl , int maxSize , Consumer < ? super T > onBufferEviction ) { } }

implementation: return onBackpressureBuffer ( ttl , maxSize , onBufferEviction , Schedulers . parallel ( ) ) ;


signature:
public class Flux { /** * Request an unbounded demand and push to the returned { @ link Flux } , or park the observed
 * elements if not enough demand is requested downstream , within a { @ code maxSize }
 * limit and for a maximum { @ link Duration } of { @ code ttl } ( as measured on the provided
 * { @ link Scheduler } ) . Over that limit , oldest elements from the source are dropped .
 * Elements evicted based on the TTL are passed to a cleanup { @ link Consumer } , which
 * is also immediately invoked when there is an overflow .
 * < img class = " marble " src = " doc - files / marbles / onBackpressureBufferWithDurationAndMaxSize . svg " alt = " " >
 * @ reactor . discard This operator discards its internal buffer of elements that overflow ,
 * after having applied the { @ code onBufferEviction } handler .
 * @ param ttl maximum { @ link Duration } for which an element is kept in the backlog
 * @ param maxSize maximum buffer backlog size before overflow callback is called
 * @ param onBufferEviction callback to invoke once TTL is reached or on overflow
 * @ param scheduler the scheduler on which to run the timeout check
 * @ return a backpressured { @ link Flux } that buffers with a TTL and up to a capacity then applies an
 * overflow strategy */
public final Flux < T > onBackpressureBuffer ( Duration ttl , int maxSize , Consumer < ? super T > onBufferEviction , Scheduler scheduler ) { } }

implementation: Objects . requireNonNull ( ttl , "ttl" ) ; Objects . requireNonNull ( onBufferEviction , "onBufferEviction" ) ; return onAssembly ( new FluxOnBackpressureBufferTimeout < > ( this , ttl , scheduler , maxSize , onBufferEviction ) ) ;


signature:
public class Flux { /** * Let compatible operators < strong > upstream < / strong > recover from errors by dropping the
 * incriminating element from the sequence and continuing with subsequent elements .
 * The recovered error and associated value are notified via the provided { @ link BiConsumer } .
 * Alternatively , throwing from that biconsumer will propagate the thrown exception downstream
 * in place of the original error , which is added as a suppressed exception to the new one .
 * < img class = " marble " src = " doc - files / marbles / onErrorContinue . svg " alt = " " >
 * Note that this error handling mode is not necessarily implemented by all operators
 * ( look for the { @ code Error Mode Support } javadoc section to find operators that
 * support it ) .
 * @ param errorConsumer a { @ link BiConsumer } fed with errors matching the predicate and the value
 * that triggered the error .
 * @ return a { @ link Flux } that attempts to continue processing on errors . */
public final Flux < T > onErrorContinue ( BiConsumer < Throwable , Object > errorConsumer ) { } }

implementation: BiConsumer < Throwable , Object > genericConsumer = errorConsumer ; return subscriberContext ( Context . of ( OnNextFailureStrategy . KEY_ON_NEXT_ERROR_STRATEGY , OnNextFailureStrategy . resume ( genericConsumer ) ) ) ;


signature:
public class Flux { /** * Let compatible operators < strong > upstream < / strong > recover from errors by dropping the
 * incriminating element from the sequence and continuing with subsequent elements .
 * Only errors matching the specified { @ code type } are recovered from .
 * The recovered error and associated value are notified via the provided { @ link BiConsumer } .
 * Alternatively , throwing from that biconsumer will propagate the thrown exception downstream
 * in place of the original error , which is added as a suppressed exception to the new one .
 * < img class = " marble " src = " doc - files / marbles / onErrorContinueWithClassPredicate . svg " alt = " " >
 * Note that this error handling mode is not necessarily implemented by all operators
 * ( look for the { @ code Error Mode Support } javadoc section to find operators that
 * support it ) .
 * @ param type the { @ link Class } of { @ link Exception } that are resumed from .
 * @ param errorConsumer a { @ link BiConsumer } fed with errors matching the { @ link Class }
 * and the value that triggered the error .
 * @ return a { @ link Flux } that attempts to continue processing on some errors . */
public final < E extends Throwable > Flux < T > onErrorContinue ( Class < E > type , BiConsumer < Throwable , Object > errorConsumer ) { } }

implementation: return onErrorContinue ( type :: isInstance , errorConsumer ) ;


signature:
public class Flux { /** * Let compatible operators < strong > upstream < / strong > recover from errors by dropping the
 * incriminating element from the sequence and continuing with subsequent elements .
 * Only errors matching the { @ link Predicate } are recovered from ( note that this
 * predicate can be applied several times and thus must be idempotent ) .
 * The recovered error and associated value are notified via the provided { @ link BiConsumer } .
 * Alternatively , throwing from that biconsumer will propagate the thrown exception downstream
 * in place of the original error , which is added as a suppressed exception to the new one .
 * < img class = " marble " src = " doc - files / marbles / onErrorContinueWithPredicate . svg " alt = " " >
 * Note that this error handling mode is not necessarily implemented by all operators
 * ( look for the { @ code Error Mode Support } javadoc section to find operators that
 * support it ) .
 * @ param errorPredicate a { @ link Predicate } used to filter which errors should be resumed from .
 * This MUST be idempotent , as it can be used several times .
 * @ param errorConsumer a { @ link BiConsumer } fed with errors matching the predicate and the value
 * that triggered the error .
 * @ return a { @ link Flux } that attempts to continue processing on some errors . */
public final < E extends Throwable > Flux < T > onErrorContinue ( Predicate < E > errorPredicate , BiConsumer < Throwable , Object > errorConsumer ) { } }

implementation:
// this cast is ok as only T values will be propagated in this sequence
@ SuppressWarnings ( "unchecked" ) Predicate < Throwable > genericPredicate = ( Predicate < Throwable > ) errorPredicate ; BiConsumer < Throwable , Object > genericErrorConsumer = errorConsumer ; return subscriberContext ( Context . of ( OnNextFailureStrategy . KEY_ON_NEXT_ERROR_STRATEGY , OnNextFailureStrategy . resumeIf ( genericPredicate , genericErrorConsumer ) ) ) ;


signature:
public class Flux { /** * If an { @ link # onErrorContinue ( BiConsumer ) } variant has been used before , reverts to the default
 * ' STOP ' mode where errors are terminal events . It can be used for easier scoping of
 * the on next failure strategy or to override the inherited strategy in a sub - stream
 * ( for example in a flatMap ) . It has no effect if { @ link # onErrorContinue ( BiConsumer ) } has not
 * been used .
 * @ return a { @ link Flux } that terminates on errors , even if { @ link # onErrorContinue ( BiConsumer ) }
 * was used before */
public final Flux < T > onErrorStop ( ) { } }

implementation: return subscriberContext ( Context . of ( OnNextFailureStrategy . KEY_ON_NEXT_ERROR_STRATEGY , OnNextFailureStrategy . stop ( ) ) ) ;


signature:
public class Flux { /** * Transform an error emitted by this { @ link Flux } by synchronously applying a function
 * to it if the error matches the given type . Otherwise let the error pass through .
 * < img class = " marble " src = " doc - files / marbles / onErrorMapWithClassPredicateForFlux . svg " alt = " " >
 * @ param type the class of the exception type to react to
 * @ param mapper the error transforming { @ link Function }
 * @ param < E > the error type
 * @ return a { @ link Flux } that transforms some source errors to other errors */
public final < E extends Throwable > Flux < T > onErrorMap ( Class < E > type , Function < ? super E , ? extends Throwable > mapper ) { } }

implementation: @ SuppressWarnings ( "unchecked" ) Function < Throwable , Throwable > handler = ( Function < Throwable , Throwable > ) mapper ; return onErrorMap ( type :: isInstance , handler ) ;


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

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


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

implementation: Objects . requireNonNull ( type , "type" ) ; @ SuppressWarnings ( "unchecked" ) Function < ? super Throwable , Publisher < ? extends T > > handler = ( Function < ? super Throwable , Publisher < ? extends T > > ) fallback ; return onErrorResume ( type :: isInstance , handler ) ;


signature:
public class Flux { /** * Subscribe to a fallback publisher when an error matching a given predicate
 * occurs .
 * < img class = " marble " src = " doc - files / marbles / onErrorResumeForFlux . svg " alt = " " >
 * @ param predicate the error predicate to match
 * @ param fallback the function to choose the fallback to an alternative { @ link Publisher }
 * @ return a { @ link Flux } falling back upon source onError */
public final Flux < T > onErrorResume ( Predicate < ? super Throwable > predicate , Function < ? super Throwable , ? extends Publisher < ? extends T > > fallback ) { } }

implementation: Objects . requireNonNull ( predicate , "predicate" ) ; return onErrorResume ( e -> predicate . test ( e ) ? fallback . apply ( e ) : error ( e ) ) ;


signature:
public class Flux { /** * Simply emit a captured fallback value when an error of the specified type is
 * observed on this { @ link Flux } .
 * < img class = " marble " src = " doc - files / marbles / onErrorReturnForFlux . 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 Flux } */
public final < E extends Throwable > Flux < T > onErrorReturn ( Class < E > type , T fallbackValue ) { } }

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


signature:
public class Flux { /** * Simply emit a captured fallback value when an error matching the given predicate is
 * observed on this { @ link Flux } .
 * < img class = " marble " src = " doc - files / marbles / onErrorReturnForFlux . 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 falling back { @ link Flux } */
public final Flux < T > onErrorReturn ( Predicate < ? super Throwable > predicate , T fallbackValue ) { } }

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


signature:
public class Flux { /** * Pick the first { @ link Publisher } between this { @ link Flux } and another publisher
 * to emit any signal ( onNext / onError / onComplete ) and replay all signals from that
 * { @ link Publisher } , effectively behaving like the fastest of these competing sources .
 * < img class = " marble " src = " doc - files / marbles / orForFlux . svg " alt = " " >
 * @ param other the { @ link Publisher } to race with
 * @ return the fastest sequence
 * @ see # first */
public final Flux < T > or ( Publisher < ? extends T > other ) { } }

implementation: if ( this instanceof FluxFirstEmitting ) { FluxFirstEmitting < T > publisherAmb = ( FluxFirstEmitting < T > ) this ; FluxFirstEmitting < T > result = publisherAmb . ambAdditionalSource ( other ) ; if ( result != null ) { return result ; } } return first ( this , other ) ;


signature:
public class Flux { /** * Prepare this { @ link Flux } by dividing data on a number of ' rails ' matching the
 * provided { @ code parallelism } parameter , in a round - robin fashion and using a
 * custom prefetch amount and queue for dealing with the source { @ link Flux } ' s values .
 * Note that to actually perform the work in parallel , you should call
 * { @ link ParallelFlux # runOn ( Scheduler ) } afterward .
 * < img class = " marble " src = " doc - files / marbles / parallel . svg " alt = " " >
 * @ param parallelism the number of parallel rails
 * @ param prefetch the number of values to prefetch from the source
 * @ return a new { @ link ParallelFlux } instance */
public final ParallelFlux < T > parallel ( int parallelism , int prefetch ) { } }

implementation: return ParallelFlux . from ( this , parallelism , prefetch , Queues . get ( prefetch ) ) ;


signature:
public class Flux { /** * Prepare a { @ link ConnectableFlux } which shares this { @ link Flux } sequence and
 * dispatches values to subscribers in a backpressure - aware manner . This will
 * effectively turn any type of sequence into a hot sequence .
 * Backpressure will be coordinated on { @ link Subscription # request } and if any
 * { @ link Subscriber } is missing demand ( requested = 0 ) , multicast will pause
 * pushing / pulling .
 * < img class = " marble " src = " doc - files / marbles / publish . svg " alt = " " >
 * @ param prefetch bounded requested demand
 * @ return a new { @ link ConnectableFlux } */
public final ConnectableFlux < T > publish ( int prefetch ) { } }

implementation: return onAssembly ( new FluxPublish < > ( this , prefetch , Queues . get ( prefetch ) ) ) ;


signature:
public class Flux { /** * Shares a sequence 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 Flux } */
public final < R > Flux < R > publish ( Function < ? super Flux < T > , ? extends Publisher < ? extends R > > transform ) { } }

implementation: return publish ( transform , Queues . SMALL_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Shares a sequence 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 prefetch the request size
 * @ param < R > the output value type
 * @ return a new { @ link Flux } */
public final < R > Flux < R > publish ( Function < ? super Flux < T > , ? extends Publisher < ? extends R > > transform , int prefetch ) { } }

implementation: return onAssembly ( new FluxPublishMulticast < > ( this , transform , prefetch , Queues . get ( prefetch ) ) ) ;


signature:
public class Flux { /** * Run onNext , onComplete and onError on a supplied { @ link Scheduler }
 * { @ link 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 / publishOnForFlux . svg " alt = " " >
 * Typically used for fast publisher , slow consumer ( s ) scenarios .
 * < blockquote > < pre >
 * { @ code flux . publishOn ( Schedulers . single ( ) ) . subscribe ( ) }
 * < / pre > < / blockquote >
 * @ reactor . discard This operator discards elements it internally queued for backpressure upon cancellation or error triggered by a data signal .
 * @ param scheduler a { @ link Scheduler } providing the { @ link Worker } where to publish
 * @ param prefetch the asynchronous boundary capacity
 * @ return a { @ link Flux } producing asynchronously */
public final Flux < T > publishOn ( Scheduler scheduler , int prefetch ) { } }

implementation: return publishOn ( scheduler , true , prefetch ) ;


signature:
public class Flux { /** * Run onNext , onComplete and onError on a supplied { @ link Scheduler }
 * { @ link 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 / publishOnForFlux . svg " alt = " " >
 * Typically used for fast publisher , slow consumer ( s ) scenarios .
 * < blockquote > < pre >
 * { @ code flux . publishOn ( Schedulers . single ( ) ) . subscribe ( ) }
 * < / pre > < / blockquote >
 * @ reactor . discard This operator discards elements it internally queued for backpressure upon cancellation or error triggered by a data signal .
 * @ param scheduler a { @ link Scheduler } providing the { @ link Worker } where to publish
 * @ param delayError should the buffer be consumed before forwarding any error
 * @ param prefetch the asynchronous boundary capacity
 * @ return a { @ link Flux } producing asynchronously */
public final Flux < T > publishOn ( Scheduler scheduler , boolean delayError , int prefetch ) { } }

implementation: return publishOn ( scheduler , delayError , prefetch , prefetch ) ;


signature:
public class Flux { /** * Reduce the values from this { @ link Flux } sequence into a single object of the same
 * type than the emitted items . Reduction is performed using a { @ link BiFunction } that
 * takes the intermediate result of the reduction and the current value and returns
 * the next intermediate value of the reduction . Note , { @ link BiFunction } will not
 * be invoked for a sequence with 0 or 1 elements . In case of one element ' s
 * sequence , the result will be directly sent to the subscriber .
 * < img class = " marble " src = " doc - files / marbles / reduceWithSameReturnType . svg " alt = " " >
 * @ reactor . discard This operator discards the internally accumulated value upon cancellation or error .
 * @ param aggregator the reducing { @ link BiFunction }
 * @ return a reduced { @ link Flux } */
public final Mono < T > reduce ( BiFunction < T , T , T > aggregator ) { } }

implementation: if ( this instanceof Callable ) { @ SuppressWarnings ( "unchecked" ) Callable < T > thiz = ( Callable < T > ) this ; return convertToMono ( thiz ) ; } return Mono . onAssembly ( new MonoReduce < > ( this , aggregator ) ) ;


signature:
public class Flux { /** * Reduce the values from this { @ link Flux } sequence into a single object matching the
 * type of a seed value . Reduction is performed using a { @ link BiFunction } that
 * takes the intermediate result of the reduction and the current value and returns
 * the next intermediate value of the reduction . First element is paired with the seed
 * value , { @ literal initial } .
 * < img class = " marble " src = " doc - files / marbles / reduce . svg " alt = " " >
 * @ reactor . discard This operator discards the internally accumulated value upon cancellation or error .
 * @ param accumulator the reducing { @ link BiFunction }
 * @ param initial the seed , the initial leftmost argument to pass to the reducing { @ link BiFunction }
 * @ param < A > the type of the seed and the reduced object
 * @ return a reduced { @ link Flux } */
public final < A > Mono < A > reduce ( A initial , BiFunction < A , ? super T , A > accumulator ) { } }

implementation: return reduceWith ( ( ) -> initial , accumulator ) ;


signature:
public class Flux { /** * Reduce the values from this { @ link Flux } sequence into a single object matching the
 * type of a lazily supplied seed value . Reduction is performed using a
 * { @ link BiFunction } that takes the intermediate result of the reduction and the
 * current value and returns the next intermediate value of the reduction . First
 * element is paired with the seed value , supplied via { @ literal initial } .
 * < img class = " marble " src = " doc - files / marbles / reduceWith . svg " alt = " " >
 * @ reactor . discard This operator discards the internally accumulated value upon cancellation or error .
 * @ param accumulator the reducing { @ link BiFunction }
 * @ param initial a { @ link Supplier } of the seed , called on subscription and passed to the the reducing { @ link BiFunction }
 * @ param < A > the type of the seed and the reduced object
 * @ return a reduced { @ link Flux } */
public final < A > Mono < A > reduceWith ( Supplier < A > initial , BiFunction < A , ? super T , A > accumulator ) { } }

implementation: return Mono . onAssembly ( new MonoReduceSeed < > ( this , initial , accumulator ) ) ;


signature:
public class Flux { /** * Repeatedly subscribe to the source if the predicate returns true after completion of the previous
 * subscription . A specified maximum of repeat will limit the number of re - subscribe .
 * < img class = " marble " src = " doc - files / marbles / repeatWithAttemptsAndPredicateForFlux . svg " alt = " " >
 * @ param numRepeat the number of times to re - subscribe on complete ( positive , or 0 for original sequence only )
 * @ param predicate the boolean to evaluate on onComplete
 * @ return a { @ link Flux } that repeats on onComplete while the predicate matches ,
 * up to the specified number of repetitions */
public final Flux < T > repeat ( long numRepeat , BooleanSupplier predicate ) { } }

implementation: if ( numRepeat < 0L ) { throw new IllegalArgumentException ( "numRepeat >= 0 required" ) ; } if ( numRepeat == 0 ) { return this ; } return defer ( ( ) -> repeat ( countingBooleanSupplier ( predicate , numRepeat ) ) ) ;


signature:
public class Flux { /** * Repeatedly subscribe to this { @ link Flux } when a companion sequence emits elements in
 * response to the flux completion signal . Any terminal signal from the companion
 * sequence will terminate the resulting { @ link Flux } with the same signal immediately .
 * < p > If the companion sequence signals when this { @ link Flux } is active , the repeat
 * attempt is suppressed .
 * < img class = " marble " src = " doc - files / marbles / repeatWhenForFlux . svg " alt = " " >
 * Note that if the companion { @ link Publisher } created by the { @ code repeatFactory }
 * emits { @ link Context } as trigger objects , these { @ link Context } will REPLACE the
 * operator ' s own Context . < strong > Please be careful there < / strong > : replacing the
 * Context means that some keys you don ' t own could be removed , breaking libraries
 * that depend on them . As a result , the recommended approach is to always create such
 * a { @ link Context } trigger by starting from the original Context ( ensuring the trigger
 * contains all the keys from the original , unless you absolutely know you want to
 * remove one of these keys ) :
 * < pre > < code >
 * . repeatWhen ( emittedEachAttempt - > emittedEachAttempt
 * . flatMap ( e - > Mono . subscriberContext ( ) . map ( ctx - > Tuples . of ( e , ctx ) ) )
 * . flatMap ( t2 - > {
 * long lastEmitted = t2 . getT1 ( ) ;
 * Context ctx = t2 . getT2 ( ) ;
 * int rl = ctx . getOrDefault ( " repeatsLeft " , 0 ) ;
 * if ( rl > 0 ) {
 * / / / ! \ THE ctx . put HERE IS THE ESSENTIAL PART / ! \
 * return Mono . just ( ctx . put ( " repeatsLeft " , rl - 1)
 * . put ( " emitted " , lastEmitted ) ) ;
 * } else {
 * return Mono . < Context > error ( new IllegalStateException ( " repeats exhausted " ) ) ;
 * < / code > < / pre >
 * @ param repeatFactory the { @ link Function } that returns the associated { @ link Publisher }
 * companion , given a { @ link Flux } that signals each onComplete as a { @ link Long }
 * representing the number of source elements emitted in the latest attempt .
 * @ return a { @ link Flux } that repeats on onComplete when the companion { @ link Publisher } produces an
 * onNext signal */
public final Flux < T > repeatWhen ( Function < Flux < Long > , ? extends Publisher < ? > > repeatFactory ) { } }

implementation: return onAssembly ( new FluxRepeatWhen < > ( this , repeatFactory ) ) ;


signature:
public class Flux { /** * Turn this { @ link Flux } into a connectable hot source and cache last emitted signals
 * for further { @ link Subscriber } . Will retain up to the given history size onNext
 * signals with a per - item ttl .
 * Completion and Error will also be replayed until { @ code ttl } triggers in which case
 * the next { @ link Subscriber } will start over a new subscription
 * < img class = " marble " src = " doc - files / marbles / replayWithHistoryAndTtl . svg " alt = " " >
 * @ param history number of events retained in history excluding complete and error
 * @ param ttl Per - item and post termination timeout duration
 * @ return a replaying { @ link ConnectableFlux } */
public final ConnectableFlux < T > replay ( int history , Duration ttl ) { } }

implementation: return replay ( history , ttl , Schedulers . parallel ( ) ) ;


signature:
public class Flux { /** * Turn this { @ link Flux } into a connectable hot source and cache last emitted signals
 * for further { @ link Subscriber } . Will retain onNext signal for up to the given
 * { @ link Duration } with a per - item ttl .
 * Completion and Error will also be replayed until { @ code ttl } triggers in which case
 * the next { @ link Subscriber } will start over a new subscription
 * < img class = " marble " src = " doc - files / marbles / replayWithTtl . svg " alt = " " >
 * @ param ttl Per - item and post termination timeout duration
 * @ param timer a time - capable { @ link Scheduler } instance to read current time from
 * @ return a replaying { @ link ConnectableFlux } */
public final ConnectableFlux < T > replay ( Duration ttl , Scheduler timer ) { } }

implementation: return replay ( Integer . MAX_VALUE , ttl , timer ) ;


signature:
public class Flux { /** * Turn this { @ link Flux } into a connectable hot source and cache last emitted signals
 * for further { @ link Subscriber } . Will retain up to the given history size onNext
 * signals with a per - item ttl .
 * Completion and Error will also be replayed until { @ code ttl } triggers in which case
 * the next { @ link Subscriber } will start over a new subscription
 * < img class = " marble " src = " doc - files / marbles / replayWithHistoryAndTtl . svg " alt = " " >
 * @ param history number of events retained in history excluding complete and error
 * @ param ttl Per - item and post termination timeout duration
 * @ param timer a { @ link Scheduler } instance to read current time from
 * @ return a replaying { @ link ConnectableFlux } */
public final ConnectableFlux < T > replay ( int history , Duration ttl , Scheduler timer ) { } }

implementation: Objects . requireNonNull ( timer , "timer" ) ; return onAssembly ( new FluxReplay < > ( this , history , ttl . toMillis ( ) , timer ) ) ;


signature:
public class Flux { /** * Retries this { @ link Flux } when a companion sequence signals
 * an item in response to this { @ link Flux } error signal
 * < p > If the companion sequence signals when the { @ link Flux } is active , the retry
 * attempt is suppressed and any terminal signal will terminate the { @ link Flux } source with the same signal
 * immediately .
 * < img class = " marble " src = " doc - files / marbles / retryWhenForFlux . svg " alt = " " >
 * Note that if the companion { @ link Publisher } created by the { @ code whenFactory }
 * emits { @ link Context } as trigger objects , these { @ link Context } will REPLACE the
 * operator ' s own Context . < strong > Please be careful there < / strong > : replacing the
 * Context means that some keys you don ' t own could be removed , breaking libraries
 * that depend on them . As a result , the recommended approach is to always create such
 * a { @ link Context } trigger by starting from the original Context ( ensuring the trigger
 * contains all the keys from the original , unless you absolutely know you want to
 * remove one of these keys ) :
 * < pre > < code >
 * . retryWhen ( errorCurrentAttempt - > errorCurrentAttempt
 * . flatMap ( e - > Mono . subscriberContext ( ) . map ( ctx - > Tuples . of ( e , ctx ) ) )
 * . flatMap ( t2 - > {
 * Throwable lastError = t2 . getT1 ( ) ;
 * Context ctx = t2 . getT2 ( ) ;
 * int rl = ctx . getOrDefault ( " retriesLeft " , 0 ) ;
 * if ( rl > 0 ) {
 * / / / ! \ THE ctx . put HERE IS THE ESSENTIAL PART / ! \
 * return Mono . just ( ctx . put ( " retriesLeft " , rl - 1)
 * . put ( " lastError " , lastError ) ) ;
 * } else {
 * return Mono . < Context > error ( new IllegalStateException ( " retries exhausted " , lastError ) ) ;
 * < / code > < / pre >
 * @ 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 Flux } that retries on onError when the companion { @ link Publisher } produces an
 * onNext signal */
public final Flux < T > retryWhen ( Function < Flux < Throwable > , ? extends Publisher < ? > > whenFactory ) { } }

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