signature:
public class Flux { /** * Merge data from { @ link Publisher } sequences provided in an { @ link Iterable }
 * into an ordered merged sequence . Unlike concat , sources are subscribed to
 * eagerly ( but at most { @ code maxConcurrency } sources at a time ) . Unlike merge , their
 * emitted values are merged into the final sequence in subscription order .
 * < img class = " marble " src = " doc - files / marbles / mergeSequentialVarSources . svg " alt = " " >
 * @ param sources an { @ link Iterable } of { @ link Publisher } sequences to merge
 * @ param maxConcurrency the request produced to the main source thus limiting concurrent merge backlog
 * @ param prefetch the inner source request size
 * @ param < I > the merged type
 * @ return a merged { @ link Flux } , subscribing early but keeping the original ordering */
public static < I > Flux < I > mergeSequential ( Iterable < ? extends Publisher < ? extends I > > sources , int maxConcurrency , int prefetch ) { } }


signature:
public class Flux { /** * Merge data from { @ link Publisher } sequences provided in an { @ link Iterable }
 * into an ordered merged sequence . Unlike concat , sources are subscribed to
 * eagerly ( but at most { @ code maxConcurrency } sources at a time ) . Unlike merge , their
 * emitted values are merged into the final sequence in subscription order .
 * This variant will delay any error until after the rest of the mergeSequential backlog
 * has been processed .
 * < img class = " marble " src = " doc - files / marbles / mergeSequentialVarSources . svg " alt = " " >
 * @ param sources an { @ link Iterable } of { @ link Publisher } sequences to merge
 * @ param maxConcurrency the request produced to the main source thus limiting concurrent merge backlog
 * @ param prefetch the inner source request size
 * @ param < I > the merged type
 * @ return a merged { @ link Flux } , subscribing early but keeping the original ordering */
public static < I > Flux < I > mergeSequentialDelayError ( Iterable < ? extends Publisher < ? extends I > > sources , int maxConcurrency , int prefetch ) { } }

implementation: return mergeSequential ( sources , true , maxConcurrency , prefetch ) ;


signature:
public class Flux { /** * Build a { @ link Flux } that will only emit a sequence of { @ code count } incrementing integers ,
 * starting from { @ code start } . That is , emit integers between { @ code start } ( included )
 * and { @ code start + count } ( excluded ) then complete .
 * < img class = " marble " src = " doc - files / marbles / range . svg " alt = " " >
 * @ param start the first integer to be emit
 * @ param count the total number of incrementing values to emit , including the first value
 * @ return a ranged { @ link Flux } */
public static Flux < Integer > range ( int start , int count ) { } }

implementation: if ( count == 1 ) { return just ( start ) ; } if ( count == 0 ) { return empty ( ) ; } return onAssembly ( new FluxRange ( start , count ) ) ;


signature:
public class Flux { /** * Creates a { @ link Flux } that mirrors the most recently emitted { @ link Publisher } ,
 * forwarding its data until a new { @ link Publisher } comes in in the source .
 * The resulting { @ link Flux } will complete once there are no new { @ link Publisher } in
 * the source ( source has completed ) and the last mirrored { @ link Publisher } has also
 * completed .
 * < img class = " marble " src = " doc - files / marbles / switchOnNext . svg " alt = " " >
 * @ param mergedPublishers The { @ link Publisher } of { @ link Publisher } to switch on and mirror .
 * @ param < T > the produced type
 * @ return a { @ link FluxProcessor } accepting publishers and producing T */
public static < T > Flux < T > switchOnNext ( Publisher < ? extends Publisher < ? extends T > > mergedPublishers ) { } }

implementation: return switchOnNext ( mergedPublishers , Queues . XS_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Creates a { @ link Flux } that mirrors the most recently emitted { @ link Publisher } ,
 * forwarding its data until a new { @ link Publisher } comes in in the source .
 * The resulting { @ link Flux } will complete once there are no new { @ link Publisher } in
 * the source ( source has completed ) and the last mirrored { @ link Publisher } has also
 * completed .
 * < img class = " marble " src = " doc - files / marbles / switchOnNext . svg " alt = " " >
 * @ param mergedPublishers The { @ link Publisher } of { @ link Publisher } to switch on and mirror .
 * @ param prefetch the inner source request size
 * @ param < T > the produced type
 * @ return a { @ link FluxProcessor } accepting publishers and producing T */
public static < T > Flux < T > switchOnNext ( Publisher < ? extends Publisher < ? extends T > > mergedPublishers , int prefetch ) { } }

implementation: return onAssembly ( new FluxSwitchMap < > ( from ( mergedPublishers ) , identityFunction ( ) , Queues . unbounded ( prefetch ) , prefetch ) ) ;


signature:
public class Flux { /** * Uses a resource , generated by a supplier for each individual Subscriber , while streaming the values from a
 * Publisher derived from the same resource and makes sure the resource is released if the sequence terminates or
 * the Subscriber cancels .
 * Eager resource cleanup happens just before the source termination and exceptions raised by the cleanup Consumer
 * may override the terminal even .
 * < img class = " marble " src = " doc - files / marbles / usingForFlux . svg " alt = " " >
 * For an asynchronous version of the cleanup , with distinct path for onComplete , onError
 * and cancel terminations , see { @ link # usingWhen ( Publisher , Function , Function , Function , Function ) } .
 * @ param resourceSupplier a { @ link Callable } that is called on subscribe to generate the resource
 * @ param sourceSupplier a factory to derive a { @ link Publisher } from the supplied resource
 * @ param resourceCleanup a resource cleanup callback invoked on completion
 * @ param < T > emitted type
 * @ param < D > resource type
 * @ return a new { @ link Flux } built around a disposable resource
 * @ see # usingWhen ( Publisher , Function , Function , Function , Function )
 * @ see # usingWhen ( Publisher , Function , Function ) */
public static < T , D > Flux < T > using ( Callable < ? extends D > resourceSupplier , Function < ? super D , ? extends Publisher < ? extends T > > sourceSupplier , Consumer < ? super D > resourceCleanup ) { } }

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


signature:
public class Flux { /** * Uses a resource , generated by a supplier for each individual Subscriber , while streaming the values from a
 * Publisher derived from the same resource and makes sure the resource is released if the sequence terminates or
 * the Subscriber cancels .
 * < ul > < li > Eager resource cleanup happens just before the source termination and exceptions raised by the cleanup
 * Consumer may override the terminal even . < / li > < li > Non - eager cleanup will drop any exception . < / li > < / ul >
 * < img class = " marble " src = " doc - files / marbles / usingForFlux . svg " alt = " " >
 * For an asynchronous version of the cleanup , with distinct path for onComplete , onError
 * and cancel terminations , see { @ link # usingWhen ( Publisher , Function , Function , Function , Function ) } .
 * @ param resourceSupplier a { @ link Callable } that is called on subscribe to generate the resource
 * @ param sourceSupplier a factory to derive a { @ link Publisher } from the supplied resource
 * @ param resourceCleanup a resource cleanup callback invoked on completion
 * @ param eager true to clean before terminating downstream subscribers
 * @ param < T > emitted type
 * @ param < D > resource type
 * @ return a new { @ link Flux } built around a disposable resource
 * @ see # usingWhen ( Publisher , Function , Function , Function , Function )
 * @ see # usingWhen ( Publisher , Function , Function ) */
public static < T , D > Flux < T > using ( Callable < ? extends D > resourceSupplier , Function < ? super D , ? extends Publisher < ? extends T > > sourceSupplier , Consumer < ? super D > resourceCleanup , boolean eager ) { } }

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


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

implementation:
// null asyncCancel translates to using the ` asyncComplete ` function in the operator
return onAssembly ( new FluxUsingWhen < > ( resourceSupplier , resourceClosure , asyncComplete , asyncError , null ) ) ;


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

implementation: return usingWhen ( resourceSupplier , resourceClosure , asyncCleanup , asyncCleanup ) ;


signature:
public class Flux { /** * Zip two sources together , that is to say wait for all the sources to emit one
 * element and combine these elements once into an output value ( constructed by the provided
 * combinator ) . The operator will continue doing so until any of the sources completes .
 * Errors will immediately be forwarded .
 * This " Step - Merge " processing is especially useful in Scatter - Gather scenarios .
 * < img class = " marble " src = " doc - files / marbles / zipTwoSourcesWithZipperForFlux . svg " alt = " " >
 * @ param source1 The first { @ link Publisher } source to zip .
 * @ param source2 The second { @ link Publisher } source to zip .
 * @ param combinator The aggregate function that will receive a unique value from each upstream and return the
 * value to signal downstream
 * @ param < T1 > type of the value from source1
 * @ param < T2 > type of the value from source2
 * @ param < O > The produced output after transformation by the combinator
 * @ return a zipped { @ link Flux } */
public static < T1 , T2 , O > Flux < O > zip ( Publisher < ? extends T1 > source1 , Publisher < ? extends T2 > source2 , final BiFunction < ? super T1 , ? super T2 , ? extends O > combinator ) { } }

implementation: return onAssembly ( new FluxZip < T1 , O > ( source1 , source2 , combinator , Queues . xs ( ) , Queues . XS_BUFFER_SIZE ) ) ;


signature:
public class Flux { /** * Zip two sources together , that is to say wait for all the sources to emit one
 * element and combine these elements once into a { @ link Tuple2 } .
 * The operator will continue doing so until any of the sources completes .
 * Errors will immediately be forwarded .
 * This " Step - Merge " processing is especially useful in Scatter - Gather scenarios .
 * < img class = " marble " src = " doc - files / marbles / zipFixedSourcesForFlux . svg " alt = " " >
 * @ param source1 The first { @ link Publisher } source to zip .
 * @ param source2 The second { @ link Publisher } source to zip .
 * @ param < T1 > type of the value from source1
 * @ param < T2 > type of the value from source2
 * @ return a zipped { @ link Flux } */
public static < T1 , T2 > Flux < Tuple2 < T1 , T2 > > zip ( Publisher < ? extends T1 > source1 , Publisher < ? extends T2 > source2 ) { } }

implementation: return zip ( source1 , source2 , tuple2Function ( ) ) ;


signature:
public class Flux { /** * Zip three sources together , that is to say wait for all the sources to emit one
 * element and combine these elements once into a { @ link Tuple3 } .
 * The operator will continue doing so until any of the sources completes .
 * Errors will immediately be forwarded .
 * This " Step - Merge " processing is especially useful in Scatter - Gather scenarios .
 * < img class = " marble " src = " doc - files / marbles / zipFixedSourcesForFlux . svg " alt = " " >
 * @ param source1 The first upstream { @ link Publisher } to subscribe to .
 * @ param source2 The second upstream { @ link Publisher } to subscribe to .
 * @ param source3 The third upstream { @ link Publisher } to subscribe to .
 * @ param < T1 > type of the value from source1
 * @ param < T2 > type of the value from source2
 * @ param < T3 > type of the value from source3
 * @ return a zipped { @ link Flux } */
public static < T1 , T2 , T3 > Flux < Tuple3 < T1 , T2 , T3 > > zip ( Publisher < ? extends T1 > source1 , Publisher < ? extends T2 > source2 , Publisher < ? extends T3 > source3 ) { } }

implementation: return zip ( Tuples . fn3 ( ) , source1 , source2 , source3 ) ;


signature:
public class Flux { /** * Zip four sources together , that is to say wait for all the sources to emit one
 * element and combine these elements once into a { @ link Tuple4 } .
 * The operator will continue doing so until any of the sources completes .
 * Errors will immediately be forwarded .
 * This " Step - Merge " processing is especially useful in Scatter - Gather scenarios .
 * < img class = " marble " src = " doc - files / marbles / zipFixedSourcesForFlux . svg " alt = " " >
 * @ param source1 The first upstream { @ link Publisher } to subscribe to .
 * @ param source2 The second upstream { @ link Publisher } to subscribe to .
 * @ param source3 The third upstream { @ link Publisher } to subscribe to .
 * @ param source4 The fourth upstream { @ link Publisher } to subscribe to .
 * @ param < T1 > type of the value from source1
 * @ param < T2 > type of the value from source2
 * @ param < T3 > type of the value from source3
 * @ param < T4 > type of the value from source4
 * @ return a zipped { @ link Flux } */
public static < T1 , T2 , T3 , T4 > Flux < Tuple4 < T1 , T2 , T3 , T4 > > zip ( Publisher < ? extends T1 > source1 , Publisher < ? extends T2 > source2 , Publisher < ? extends T3 > source3 , Publisher < ? extends T4 > source4 ) { } }

implementation: return zip ( Tuples . fn4 ( ) , source1 , source2 , source3 , source4 ) ;


signature:
public class Flux { /** * Zip five sources together , that is to say wait for all the sources to emit one
 * element and combine these elements once into a { @ link Tuple5 } .
 * The operator will continue doing so until any of the sources completes .
 * Errors will immediately be forwarded .
 * This " Step - Merge " processing is especially useful in Scatter - Gather scenarios .
 * < img class = " marble " src = " doc - files / marbles / zipFixedSourcesForFlux . svg " alt = " " >
 * @ param source1 The first upstream { @ link Publisher } to subscribe to .
 * @ param source2 The second upstream { @ link Publisher } to subscribe to .
 * @ param source3 The third upstream { @ link Publisher } to subscribe to .
 * @ param source4 The fourth upstream { @ link Publisher } to subscribe to .
 * @ param source5 The fifth upstream { @ link Publisher } to subscribe to .
 * @ param < T1 > type of the value from source1
 * @ param < T2 > type of the value from source2
 * @ param < T3 > type of the value from source3
 * @ param < T4 > type of the value from source4
 * @ param < T5 > type of the value from source5
 * @ return a zipped { @ link Flux } */
public static < T1 , T2 , T3 , T4 , T5 > Flux < Tuple5 < T1 , T2 , T3 , T4 , T5 > > zip ( Publisher < ? extends T1 > source1 , Publisher < ? extends T2 > source2 , Publisher < ? extends T3 > source3 , Publisher < ? extends T4 > source4 , Publisher < ? extends T5 > source5 ) { } }

implementation: return zip ( Tuples . fn5 ( ) , source1 , source2 , source3 , source4 , source5 ) ;


signature:
public class Flux { /** * Zip six sources together , that is to say wait for all the sources to emit one
 * element and combine these elements once into a { @ link Tuple6 } .
 * The operator will continue doing so until any of the sources completes .
 * Errors will immediately be forwarded .
 * This " Step - Merge " processing is especially useful in Scatter - Gather scenarios .
 * < img class = " marble " src = " doc - files / marbles / zipFixedSourcesForFlux . svg " alt = " " >
 * @ param source1 The first upstream { @ link Publisher } to subscribe to .
 * @ param source2 The second upstream { @ link Publisher } to subscribe to .
 * @ param source3 The third upstream { @ link Publisher } to subscribe to .
 * @ param source4 The fourth upstream { @ link Publisher } to subscribe to .
 * @ param source5 The fifth upstream { @ link Publisher } to subscribe to .
 * @ param source6 The sixth upstream { @ link Publisher } to subscribe to .
 * @ param < T1 > type of the value from source1
 * @ param < T2 > type of the value from source2
 * @ param < T3 > type of the value from source3
 * @ param < T4 > type of the value from source4
 * @ param < T5 > type of the value from source5
 * @ param < T6 > type of the value from source6
 * @ return a zipped { @ link Flux } */
public static < T1 , T2 , T3 , T4 , T5 , T6 > Flux < Tuple6 < T1 , T2 , T3 , T4 , T5 , T6 > > zip ( Publisher < ? extends T1 > source1 , Publisher < ? extends T2 > source2 , Publisher < ? extends T3 > source3 , Publisher < ? extends T4 > source4 , Publisher < ? extends T5 > source5 , Publisher < ? extends T6 > source6 ) { } }

implementation: return zip ( Tuples . fn6 ( ) , source1 , source2 , source3 , source4 , source5 , source6 ) ;


signature:
public class Flux { /** * Zip seven sources together , that is to say wait for all the sources to emit one
 * element and combine these elements once into a { @ link Tuple7 } .
 * The operator will continue doing so until any of the sources completes .
 * Errors will immediately be forwarded .
 * This " Step - Merge " processing is especially useful in Scatter - Gather scenarios .
 * < img class = " marble " src = " doc - files / marbles / zipFixedSourcesForFlux . svg " alt = " " >
 * @ param source1 The first upstream { @ link Publisher } to subscribe to .
 * @ param source2 The second upstream { @ link Publisher } to subscribe to .
 * @ param source3 The third upstream { @ link Publisher } to subscribe to .
 * @ param source4 The fourth upstream { @ link Publisher } to subscribe to .
 * @ param source5 The fifth upstream { @ link Publisher } to subscribe to .
 * @ param source6 The sixth upstream { @ link Publisher } to subscribe to .
 * @ param source7 The seventh upstream { @ link Publisher } to subscribe to .
 * @ param < T1 > type of the value from source1
 * @ param < T2 > type of the value from source2
 * @ param < T3 > type of the value from source3
 * @ param < T4 > type of the value from source4
 * @ param < T5 > type of the value from source5
 * @ param < T6 > type of the value from source6
 * @ param < T7 > type of the value from source7
 * @ return a zipped { @ link Flux } */
public static < T1 , T2 , T3 , T4 , T5 , T6 , T7 > Flux < Tuple7 < T1 , T2 , T3 , T4 , T5 , T6 , T7 > > zip ( Publisher < ? extends T1 > source1 , Publisher < ? extends T2 > source2 , Publisher < ? extends T3 > source3 , Publisher < ? extends T4 > source4 , Publisher < ? extends T5 > source5 , Publisher < ? extends T6 > source6 , Publisher < ? extends T7 > source7 ) { } }

implementation: return zip ( Tuples . fn7 ( ) , source1 , source2 , source3 , source4 , source5 , source6 , source7 ) ;


signature:
public class Flux { /** * Zip eight sources together , that is to say wait for all the sources to emit one
 * element and combine these elements once into a { @ link Tuple8 } .
 * The operator will continue doing so until any of the sources completes .
 * Errors will immediately be forwarded .
 * This " Step - Merge " processing is especially useful in Scatter - Gather scenarios .
 * < img class = " marble " src = " doc - files / marbles / zipFixedSourcesForFlux . svg " alt = " " >
 * @ param source1 The first upstream { @ link Publisher } to subscribe to .
 * @ param source2 The second upstream { @ link Publisher } to subscribe to .
 * @ param source3 The third upstream { @ link Publisher } to subscribe to .
 * @ param source4 The fourth upstream { @ link Publisher } to subscribe to .
 * @ param source5 The fifth upstream { @ link Publisher } to subscribe to .
 * @ param source6 The sixth upstream { @ link Publisher } to subscribe to .
 * @ param source7 The seventh upstream { @ link Publisher } to subscribe to .
 * @ param source8 The eight upstream { @ link Publisher } to subscribe to .
 * @ param < T1 > type of the value from source1
 * @ param < T2 > type of the value from source2
 * @ param < T3 > type of the value from source3
 * @ param < T4 > type of the value from source4
 * @ param < T5 > type of the value from source5
 * @ param < T6 > type of the value from source6
 * @ param < T7 > type of the value from source7
 * @ param < T8 > type of the value from source8
 * @ return a zipped { @ link Flux } */
public static < T1 , T2 , T3 , T4 , T5 , T6 , T7 , T8 > Flux < Tuple8 < T1 , T2 , T3 , T4 , T5 , T6 , T7 , T8 > > zip ( Publisher < ? extends T1 > source1 , Publisher < ? extends T2 > source2 , Publisher < ? extends T3 > source3 , Publisher < ? extends T4 > source4 , Publisher < ? extends T5 > source5 , Publisher < ? extends T6 > source6 , Publisher < ? extends T7 > source7 , Publisher < ? extends T8 > source8 ) { } }

implementation: return zip ( Tuples . fn8 ( ) , source1 , source2 , source3 , source4 , source5 , source6 , source7 , source8 ) ;


signature:
public class Flux { /** * Zip multiple sources together , that is to say wait for all the sources to emit one
 * element and combine these elements once into an output value ( constructed by the provided
 * combinator ) .
 * The operator will continue doing so until any of the sources completes .
 * Errors will immediately be forwarded .
 * This " Step - Merge " processing is especially useful in Scatter - Gather scenarios .
 * The { @ link Iterable # iterator ( ) } will be called on each { @ link Publisher # subscribe ( Subscriber ) } .
 * < img class = " marble " src = " doc - files / marbles / zipIterableSourcesForFlux . svg " alt = " " >
 * @ param sources the { @ link Iterable } providing sources to zip
 * @ param combinator The aggregate function that will receive a unique value from each upstream and return the value
 * to signal downstream
 * @ param < O > the combined produced type
 * @ return a zipped { @ link Flux } */
public static < O > Flux < O > zip ( Iterable < ? extends Publisher < ? > > sources , final Function < ? super Object [ ] , ? extends O > combinator ) { } }

implementation: return zip ( sources , Queues . XS_BUFFER_SIZE , combinator ) ;


signature:
public class Flux { /** * Zip multiple sources together , that is to say wait for all the sources to emit one
 * element and combine these elements once into an output value ( constructed by the provided
 * combinator ) .
 * The operator will continue doing so until any of the sources completes .
 * Errors will immediately be forwarded .
 * This " Step - Merge " processing is especially useful in Scatter - Gather scenarios .
 * The { @ link Iterable # iterator ( ) } will be called on each { @ link Publisher # subscribe ( Subscriber ) } .
 * < img class = " marble " src = " doc - files / marbles / zipIterableSourcesForFlux . svg " alt = " " >
 * @ param sources the { @ link Iterable } providing sources to zip
 * @ param prefetch the inner source request size
 * @ param combinator The aggregate function that will receive a unique value from each upstream and return the value
 * to signal downstream
 * @ param < O > the combined produced type
 * @ return a zipped { @ link Flux } */
public static < O > Flux < O > zip ( Iterable < ? extends Publisher < ? > > sources , int prefetch , final Function < ? super Object [ ] , ? extends O > combinator ) { } }

implementation: return onAssembly ( new FluxZip < Object , O > ( sources , combinator , Queues . get ( prefetch ) , prefetch ) ) ;


signature:
public class Flux { /** * Zip multiple sources together , that is to say wait for all the sources to emit one
 * element and combine these elements once into an output value ( constructed by the provided
 * combinator ) .
 * The operator will continue doing so until any of the sources completes .
 * Errors will immediately be forwarded .
 * This " Step - Merge " processing is especially useful in Scatter - Gather scenarios .
 * < img class = " marble " src = " doc - files / marbles / zipIterableSourcesForFlux . svg " alt = " " >
 * @ param combinator The aggregate function that will receive a unique value from each upstream and return the
 * value to signal downstream
 * @ param sources the array providing sources to zip
 * @ param < I > the type of the input sources
 * @ param < O > the combined produced type
 * @ return a zipped { @ link Flux } */
@ SafeVarargs public static < I , O > Flux < O > zip ( final Function < ? super Object [ ] , ? extends O > combinator , Publisher < ? extends I > ... sources ) { } }

implementation: return zip ( combinator , Queues . XS_BUFFER_SIZE , sources ) ;


signature:
public class Flux { /** * Zip multiple sources together , that is to say wait for all the sources to emit one
 * element and combine these elements once into an output value ( constructed by the provided
 * combinator ) .
 * The operator will continue doing so until any of the sources completes .
 * Errors will immediately be forwarded .
 * This " Step - Merge " processing is especially useful in Scatter - Gather scenarios .
 * < img class = " marble " src = " doc - files / marbles / zipIterableSourcesForFlux . svg " alt = " " >
 * @ param combinator The aggregate function that will receive a unique value from each upstream and return the
 * value to signal downstream
 * @ param prefetch individual source request size
 * @ param sources the array providing sources to zip
 * @ param < I > the type of the input sources
 * @ param < O > the combined produced type
 * @ return a zipped { @ link Flux } */
@ SafeVarargs public static < I , O > Flux < O > zip ( final Function < ? super Object [ ] , ? extends O > combinator , int prefetch , Publisher < ? extends I > ... sources ) { } }

implementation: if ( sources . length == 0 ) { return empty ( ) ; } if ( sources . length == 1 ) { Publisher < ? extends I > source = sources [ 0 ] ; if ( source instanceof Fuseable ) { return onAssembly ( new FluxMapFuseable < > ( from ( source ) , v -> combinator . apply ( new Object [ ] { v } ) ) ) ; } return onAssembly ( new FluxMap < > ( from ( source ) , v -> combinator . apply ( new Object [ ] { v } ) ) ) ; } return onAssembly ( new FluxZip < > ( sources , combinator , Queues . get ( prefetch ) , prefetch ) ) ;


signature:
public class Flux { /** * Zip multiple sources together , that is to say wait for all the sources to emit one
 * element and combine these elements once into an output value ( constructed by the provided
 * combinator ) .
 * The operator will continue doing so until any of the sources completes .
 * Errors will immediately be forwarded .
 * This " Step - Merge " processing is especially useful in Scatter - Gather scenarios .
 * Note that the { @ link Publisher } sources from the outer { @ link Publisher } will
 * accumulate into an exhaustive list before starting zip operation .
 * < img class = " marble " src = " doc - files / marbles / zipAsyncSourcesForFlux . svg " alt = " " >
 * @ param sources The { @ link Publisher } of { @ link Publisher } sources to zip . A finite publisher is required .
 * @ param combinator The aggregate function that will receive a unique value from each upstream and return the value
 * to signal downstream
 * @ param < TUPLE > the raw tuple type
 * @ param < V > The produced output after transformation by the given combinator
 * @ return a { @ link Flux } based on the produced value */
@ SuppressWarnings ( { } }

implementation: "rawtypes" , "unchecked" } ) public static < TUPLE extends Tuple2 , V > Flux < V > zip ( Publisher < ? extends Publisher < ? > > sources , final Function < ? super TUPLE , ? extends V > combinator ) { return onAssembly ( new FluxBuffer < > ( from ( sources ) , Integer . MAX_VALUE , listSupplier ( ) ) . flatMap ( new Function < List < ? extends Publisher < ? > > , Publisher < V > > ( ) { @ Override public Publisher < V > apply ( List < ? extends Publisher < ? > > publishers ) { return zip ( Tuples . fnAny ( ( Function < Tuple2 , V > ) combinator ) , publishers . toArray ( new Publisher [ publishers . size ( ) ] ) ) ; } } ) ) ;


signature:
public class Flux { /** * Emit a single boolean true if any of the values of this { @ link Flux } sequence match
 * the predicate .
 * The implementation uses short - circuit logic and completes with false if any value
 * doesn ' t match the predicate .
 * < img class = " marble " src = " doc - files / marbles / any . svg " alt = " " >
 * @ param predicate the { @ link Predicate } that needs to apply to at least one emitted item
 * @ return a new { @ link Mono } with < code > true < / code > if any value satisfies a predicate and < code > false < / code >
 * otherwise */
public final Mono < Boolean > any ( Predicate < ? super T > predicate ) { } }

implementation: return Mono . onAssembly ( new MonoAny < > ( this , predicate ) ) ;


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

implementation: return transformer . apply ( this ) ;


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

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


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

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


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

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


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

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


signature:
public class Flux { /** * Collect incoming values into multiple { @ link List } buffers that will be emitted
 * by the returned { @ link Flux } each time the given max size is reached or once this
 * Flux completes . Buffers can be created with gaps , as a new buffer will be created
 * every time { @ code skip } values have been emitted by the source .
 * When maxSize < skip : dropping buffers
 * < img class = " marble " src = " doc - files / marbles / bufferWithMaxSizeLessThanSkipSize . svg " alt = " " >
 * When maxSize > skip : overlapping buffers
 * < img class = " marble " src = " doc - files / marbles / bufferWithMaxSizeGreaterThanSkipSize . svg " alt = " " >
 * When maxSize = = skip : exact buffers
 * < img class = " marble " src = " doc - files / marbles / bufferWithMaxSizeEqualsSkipSize . svg " alt = " " >
 * @ reactor . discard This operator discards elements in between buffers ( in the case of
 * dropping buffers ) . It also discards the currently open buffer upon cancellation or error triggered by a data signal .
 * Note however that overlapping buffer variant DOES NOT discard , as this might result in an element
 * being discarded from an early buffer while it is still valid in a more recent buffer .
 * @ param skip the number of items to count before creating a new buffer
 * @ param maxSize the max collected size
 * @ return a microbatched { @ link Flux } of possibly overlapped or gapped { @ link List } */
public final Flux < List < T > > buffer ( int maxSize , int skip ) { } }

implementation: return buffer ( maxSize , skip , listSupplier ( ) ) ;


signature:
public class Flux { /** * Collect incoming values into multiple user - defined { @ link Collection } buffers that
 * will be emitted by the returned { @ link Flux } each time the given max size is reached
 * or once this Flux completes . Buffers can be created with gaps , as a new buffer will
 * be created every time { @ code skip } values have been emitted by the source
 * When maxSize < skip : dropping buffers
 * < img class = " marble " src = " doc - files / marbles / bufferWithMaxSizeLessThanSkipSize . svg " alt = " " >
 * When maxSize > skip : overlapping buffers
 * < img class = " marble " src = " doc - files / marbles / bufferWithMaxSizeGreaterThanSkipSize . svg " alt = " " >
 * When maxSize = = skip : exact buffers
 * < img class = " marble " src = " doc - files / marbles / bufferWithMaxSizeEqualsSkipSize . svg " alt = " " >
 * @ reactor . discard This operator discards elements in between buffers ( in the case of
 * dropping buffers ) . It also discards the currently open buffer upon cancellation or error triggered by a data signal .
 * Note however that overlapping buffer variant DOES NOT discard , as this might result in an element
 * being discarded from an early buffer while it is still valid in a more recent buffer .
 * @ param skip the number of items to count before creating a new buffer
 * @ param maxSize the max collected size
 * @ param bufferSupplier a { @ link Supplier } of the concrete { @ link Collection } to use for each buffer
 * @ param < C > the { @ link Collection } buffer type
 * @ return a microbatched { @ link Flux } of possibly overlapped or gapped
 * { @ link Collection } */
public final < C extends Collection < ? super T > > Flux < C > buffer ( int maxSize , int skip , Supplier < C > bufferSupplier ) { } }

implementation: return onAssembly ( new FluxBuffer < > ( this , maxSize , skip , bufferSupplier ) ) ;


signature:
public class Flux { /** * Collect incoming values into multiple user - defined { @ link Collection } buffers , as
 * delimited by the signals of a companion { @ link Publisher } this operator will
 * subscribe to .
 * < img class = " marble " src = " doc - files / marbles / bufferWithBoundary . svg " alt = " " >
 * @ reactor . discard This operator discards the currently open buffer upon cancellation or error triggered by a data signal ,
 * and the last received element when the bufferSupplier fails .
 * @ param other the companion { @ link Publisher } whose signals trigger new buffers
 * @ param bufferSupplier a { @ link Supplier } of the concrete { @ link Collection } to use for each buffer
 * @ param < C > the { @ link Collection } buffer type
 * @ return a microbatched { @ link Flux } of { @ link Collection } delimited by signals from a { @ link Publisher } */
public final < C extends Collection < ? super T > > Flux < C > buffer ( Publisher < ? > other , Supplier < C > bufferSupplier ) { } }

implementation: return onAssembly ( new FluxBufferBoundary < > ( this , other , bufferSupplier ) ) ;


signature:
public class Flux { /** * Collect incoming values into multiple { @ link List } buffers that will be emitted by
 * the returned { @ link Flux } every { @ code bufferingTimespan } .
 * < img class = " marble " src = " doc - files / marbles / bufferWithTimespan . svg " alt = " " >
 * @ reactor . discard This operator discards the currently open buffer upon cancellation or error triggered by a data signal .
 * @ param bufferingTimespan the duration from buffer creation until a buffer is closed and emitted
 * @ return a microbatched { @ link Flux } of { @ link List } delimited by the given time span */
public final Flux < List < T > > buffer ( Duration bufferingTimespan ) { } }

implementation: return buffer ( bufferingTimespan , Schedulers . parallel ( ) ) ;


signature:
public class Flux { /** * Collect incoming values into multiple { @ link List } buffers that will be emitted by
 * the returned { @ link Flux } every { @ code bufferingTimespan } , as measured on the provided { @ link Scheduler } .
 * < img class = " marble " src = " doc - files / marbles / bufferWithTimespan . svg " alt = " " >
 * @ reactor . discard This operator discards the currently open buffer upon cancellation or error triggered by a data signal .
 * @ param bufferingTimespan the duration from buffer creation until a buffer is closed and emitted
 * @ param timer a time - capable { @ link Scheduler } instance to run on
 * @ return a microbatched { @ link Flux } of { @ link List } delimited by the given period */
public final Flux < List < T > > buffer ( Duration bufferingTimespan , Scheduler timer ) { } }

implementation: return buffer ( interval ( bufferingTimespan , timer ) ) ;


signature:
public class Flux { /** * Collect incoming values into multiple { @ link List } buffers created at a given
 * { @ code openBufferEvery } period , as measured on the provided { @ link Scheduler } . Each
 * buffer will last until the { @ code bufferingTimespan } has elapsed ( also measured on the scheduler ) ,
 * thus emitting the bucket in the resulting { @ link Flux } .
 * When bufferingTimespan < openBufferEvery : dropping buffers
 * < img class = " marble " src = " doc - files / marbles / bufferWithTimespanLessThanOpenBufferEvery . svg " alt = " " >
 * When bufferingTimespan > openBufferEvery : overlapping buffers
 * < img class = " marble " src = " doc - files / marbles / bufferWithTimespanGreaterThanOpenBufferEvery . svg " alt = " " >
 * When bufferingTimespan = = openBufferEvery : exact buffers
 * < img class = " marble " src = " doc - files / marbles / bufferWithTimespanEqualsOpenBufferEvery . svg " alt = " " >
 * @ reactor . discard This operator discards the currently open buffer upon cancellation or error triggered by a data signal .
 * It DOES NOT provide strong guarantees in the case of overlapping buffers , as elements
 * might get discarded too early ( from the first of two overlapping buffers for instance ) .
 * @ param bufferingTimespan the duration from buffer creation until a buffer is closed and emitted
 * @ param openBufferEvery the interval at which to create a new buffer
 * @ param timer a time - capable { @ link Scheduler } instance to run on
 * @ return a microbatched { @ link Flux } of { @ link List } delimited by the given period openBufferEvery and sized by bufferingTimespan */
public final Flux < List < T > > buffer ( Duration bufferingTimespan , Duration openBufferEvery , Scheduler timer ) { } }

implementation: if ( bufferingTimespan . equals ( openBufferEvery ) ) { return buffer ( bufferingTimespan , timer ) ; } return bufferWhen ( interval ( Duration . ZERO , openBufferEvery , timer ) , aLong -> Mono . delay ( bufferingTimespan , timer ) ) ;


signature:
public class Flux { /** * Collect incoming values into multiple user - defined { @ link Collection } buffers that
 * will be emitted by the returned { @ link Flux } each time the buffer reaches a maximum
 * size OR the maxTime { @ link Duration } elapses .
 * < img class = " marble " src = " doc - files / marbles / bufferTimeoutWithMaxSizeAndTimespan . svg " alt = " " >
 * @ reactor . discard This operator discards the currently open buffer upon cancellation or error triggered by a data signal .
 * @ param maxSize the max collected size
 * @ param maxTime the timeout enforcing the release of a partial buffer
 * @ param bufferSupplier a { @ link Supplier } of the concrete { @ link Collection } to use for each buffer
 * @ param < C > the { @ link Collection } buffer type
 * @ return a microbatched { @ link Flux } of { @ link Collection } delimited by given size or a given period timeout */
public final < C extends Collection < ? super T > > Flux < C > bufferTimeout ( int maxSize , Duration maxTime , Supplier < C > bufferSupplier ) { } }

implementation: return bufferTimeout ( maxSize , maxTime , Schedulers . parallel ( ) , bufferSupplier ) ;


signature:
public class Flux { /** * Collect incoming values into multiple { @ link List } buffers that will be emitted
 * by the returned { @ link Flux } each time the buffer reaches a maximum size OR the
 * maxTime { @ link Duration } elapses , as measured on the provided { @ link Scheduler } .
 * < img class = " marble " src = " doc - files / marbles / bufferTimeoutWithMaxSizeAndTimespan . svg " alt = " " >
 * @ reactor . discard This operator discards the currently open buffer upon cancellation or error triggered by a data signal .
 * @ param maxSize the max collected size
 * @ param maxTime the timeout enforcing the release of a partial buffer
 * @ param timer a time - capable { @ link Scheduler } instance to run on
 * @ return a microbatched { @ link Flux } of { @ link List } delimited by given size or a given period timeout */
public final Flux < List < T > > bufferTimeout ( int maxSize , Duration maxTime , Scheduler timer ) { } }

implementation: return bufferTimeout ( maxSize , maxTime , timer , listSupplier ( ) ) ;


signature:
public class Flux { /** * Collect incoming values into multiple user - defined { @ link Collection } buffers that
 * will be emitted by the returned { @ link Flux } each time the buffer reaches a maximum
 * size OR the maxTime { @ link Duration } elapses , as measured on the provided { @ link Scheduler } .
 * < img class = " marble " src = " doc - files / marbles / bufferTimeoutWithMaxSizeAndTimespan . svg " alt = " " >
 * @ reactor . discard This operator discards the currently open buffer upon cancellation or error triggered by a data signal .
 * @ param maxSize the max collected size
 * @ param maxTime the timeout enforcing the release of a partial buffer
 * @ param timer a time - capable { @ link Scheduler } instance to run on
 * @ param bufferSupplier a { @ link Supplier } of the concrete { @ link Collection } to use for each buffer
 * @ param < C > the { @ link Collection } buffer type
 * @ return a microbatched { @ link Flux } of { @ link Collection } delimited by given size or a given period timeout */
public final < C extends Collection < ? super T > > Flux < C > bufferTimeout ( int maxSize , Duration maxTime , Scheduler timer , Supplier < C > bufferSupplier ) { } }

implementation: return onAssembly ( new FluxBufferTimeout < > ( this , maxSize , maxTime . toMillis ( ) , timer , bufferSupplier ) ) ;


signature:
public class Flux { /** * Collect incoming values into multiple { @ link List } buffers that will be emitted by
 * the resulting { @ link Flux } each time the given predicate returns true . Note that
 * the element that triggers the predicate to return true ( and thus closes a buffer )
 * is included as last element in the emitted buffer .
 * < img class = " marble " src = " doc - files / marbles / bufferUntil . svg " alt = " " >
 * On completion , if the latest buffer is non - empty and has not been closed it is
 * emitted . However , such a " partial " buffer isn ' t emitted in case of onError
 * termination .
 * @ reactor . discard This operator discards the currently open buffer upon cancellation or error triggered by a data signal .
 * @ param predicate a predicate that triggers the next buffer when it becomes true .
 * @ return a microbatched { @ link Flux } of { @ link List } */
public final Flux < List < T > > bufferUntil ( Predicate < ? super T > predicate ) { } }

implementation: return onAssembly ( new FluxBufferPredicate < > ( this , predicate , listSupplier ( ) , FluxBufferPredicate . Mode . UNTIL ) ) ;


signature:
public class Flux { /** * Collect incoming values into multiple { @ link List } buffers that will be emitted by
 * the resulting { @ link Flux } . Each buffer continues aggregating values while the
 * given predicate returns true , and a new buffer is created as soon as the
 * predicate returns false . . . Note that the element that triggers the predicate
 * to return false ( and thus closes a buffer ) is NOT included in any emitted buffer .
 * < img class = " marble " src = " doc - files / marbles / bufferWhile . svg " alt = " " >
 * On completion , if the latest buffer is non - empty and has not been closed it is
 * emitted . However , such a " partial " buffer isn ' t emitted in case of onError
 * termination .
 * @ reactor . discard This operator discards the currently open buffer upon cancellation or error triggered by a data signal ,
 * as well as the buffer - triggering element .
 * @ param predicate a predicate that triggers the next buffer when it becomes false .
 * @ return a microbatched { @ link Flux } of { @ link List } */
public final Flux < List < T > > bufferWhile ( Predicate < ? super T > predicate ) { } }

implementation: return onAssembly ( new FluxBufferPredicate < > ( this , predicate , listSupplier ( ) , FluxBufferPredicate . Mode . WHILE ) ) ;


signature:
public class Flux { /** * Collect incoming values into multiple { @ link List } buffers started each time an opening
 * companion { @ link Publisher } emits . Each buffer will last until the corresponding
 * closing companion { @ link Publisher } emits , thus releasing the buffer to the resulting { @ link Flux } .
 * When Open signal is strictly not overlapping Close signal : dropping buffers ( see green marbles in diagram below ) .
 * When Open signal is strictly more frequent than Close signal : overlapping buffers ( see second and third buffers in diagram below ) .
 * When Open signal is exactly coordinated with Close signal : exact buffers
 * < img class = " marble " src = " doc - files / marbles / bufferWhen . svg " alt = " " >
 * @ reactor . discard This operator discards the currently open buffer upon cancellation or error triggered by a data signal .
 * It DOES NOT provide strong guarantees in the case of overlapping buffers , as elements
 * might get discarded too early ( from the first of two overlapping buffers for instance ) .
 * @ param bucketOpening a companion { @ link Publisher } to subscribe for buffer creation signals .
 * @ param closeSelector a factory that , given a buffer opening signal , returns a companion
 * { @ link Publisher } to subscribe to for buffer closure and emission signals .
 * @ param < U > the element type of the buffer - opening sequence
 * @ param < V > the element type of the buffer - closing sequence
 * @ return a microbatched { @ link Flux } of { @ link List } delimited by an opening { @ link Publisher } and a relative
 * closing { @ link Publisher } */
public final < U , V > Flux < List < T > > bufferWhen ( Publisher < U > bucketOpening , Function < ? super U , ? extends Publisher < V > > closeSelector ) { } }

implementation: return bufferWhen ( bucketOpening , closeSelector , listSupplier ( ) ) ;


signature:
public class Flux { /** * Collect incoming values into multiple user - defined { @ link Collection } buffers started each time an opening
 * companion { @ link Publisher } emits . Each buffer will last until the corresponding
 * closing companion { @ link Publisher } emits , thus releasing the buffer to the resulting { @ link Flux } .
 * When Open signal is strictly not overlapping Close signal : dropping buffers ( see green marbles in diagram below ) .
 * When Open signal is strictly more frequent than Close signal : overlapping buffers ( see second and third buffers in diagram below ) .
 * < img class = " marble " src = " doc - files / marbles / bufferWhenWithSupplier . svg " alt = " " >
 * @ reactor . discard This operator discards the currently open buffer upon cancellation or error triggered by a data signal .
 * It DOES NOT provide strong guarantees in the case of overlapping buffers , as elements
 * might get discarded too early ( from the first of two overlapping buffers for instance ) .
 * @ param bucketOpening a companion { @ link Publisher } to subscribe for buffer creation signals .
 * @ param closeSelector a factory that , given a buffer opening signal , returns a companion
 * { @ link Publisher } to subscribe to for buffer closure and emission signals .
 * @ param bufferSupplier a { @ link Supplier } of the concrete { @ link Collection } to use for each buffer
 * @ param < U > the element type of the buffer - opening sequence
 * @ param < V > the element type of the buffer - closing sequence
 * @ param < C > the { @ link Collection } buffer type
 * @ return a microbatched { @ link Flux } of { @ link Collection } delimited by an opening { @ link Publisher } and a relative
 * closing { @ link Publisher } */
public final < U , V , C extends Collection < ? super T > > Flux < C > bufferWhen ( Publisher < U > bucketOpening , Function < ? super U , ? extends Publisher < V > > closeSelector , Supplier < C > bufferSupplier ) { } }

implementation: return onAssembly ( new FluxBufferWhen < > ( this , bucketOpening , closeSelector , bufferSupplier , Queues . unbounded ( Queues . XS_BUFFER_SIZE ) ) ) ;


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

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


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

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


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

implementation: return replay ( history , ttl , timer ) . autoConnect ( ) ;


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

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


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

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


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

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


signature:
public class Flux { /** * Collect all elements emitted by this { @ link Flux } into a user - defined container ,
 * by applying a collector { @ link BiConsumer } taking the container and each element .
 * The collected result will be emitted when this sequence completes .
 * < img class = " marble " src = " doc - files / marbles / collect . svg " alt = " " >
 * @ param < E > the container type
 * @ param containerSupplier the supplier of the container instance for each Subscriber
 * @ param collector a consumer of both the container instance and the value being currently collected
 * @ reactor . discard This operator discards the buffer upon cancellation or error triggered by a data signal .
 * @ return a { @ link Mono } of the collected container on complete */
public final < E > Mono < E > collect ( Supplier < E > containerSupplier , BiConsumer < E , ? super T > collector ) { } }

implementation: return Mono . onAssembly ( new MonoCollect < > ( this , containerSupplier , collector ) ) ;


signature:
public class Flux { /** * Collect all elements emitted by this { @ link Flux } into a container ,
 * by applying a Java 8 Stream API { @ link Collector }
 * The collected result will be emitted when this sequence completes .
 * < img class = " marble " src = " doc - files / marbles / collectWithCollector . svg " alt = " " >
 * @ param collector the { @ link Collector }
 * @ param < A > The mutable accumulation type
 * @ param < R > the container type
 * @ return a { @ link Mono } of the collected container on complete */
public final < R , A > Mono < R > collect ( Collector < ? super T , A , ? extends R > collector ) { } }

implementation: return Mono . onAssembly ( new MonoStreamCollector < > ( this , collector ) ) ;


signature:
public class Flux { /** * Collect all elements emitted by this { @ link Flux } into a { @ link List } that is
 * emitted by the resulting { @ link Mono } when this sequence completes .
 * < img class = " marble " src = " doc - files / marbles / collectList . svg " alt = " " >
 * @ reactor . discard This operator discards the buffer upon cancellation or error triggered by a data signal .
 * @ return a { @ link Mono } of a { @ link List } of all values from this { @ link Flux } */
public final Mono < List < T > > collectList ( ) { } }

implementation: if ( this instanceof Callable ) { if ( this instanceof Fuseable . ScalarCallable ) { @ SuppressWarnings ( "unchecked" ) Fuseable . ScalarCallable < T > scalarCallable = ( Fuseable . ScalarCallable < T > ) this ; T v ; try { v = scalarCallable . call ( ) ; } catch ( Exception e ) { return Mono . error ( e ) ; } if ( v == null ) { return Mono . onAssembly ( new MonoSupplier < > ( listSupplier ( ) ) ) ; } return Mono . just ( v ) . map ( u -> { List < T > list = Flux . < T > listSupplier ( ) . get ( ) ; list . add ( u ) ; return list ; } ) ; } @ SuppressWarnings ( "unchecked" ) Callable < T > thiz = ( Callable < T > ) this ; return Mono . onAssembly ( new MonoCallable < > ( thiz ) . map ( u -> { List < T > list = Flux . < T > listSupplier ( ) . get ( ) ; list . add ( u ) ; return list ; } ) ) ; } return Mono . onAssembly ( new MonoCollectList < > ( this , listSupplier ( ) ) ) ;


signature:
public class Flux { /** * Collect all elements emitted by this { @ link Flux } into a hashed { @ link Map } that is
 * emitted by the resulting { @ link Mono } when this sequence completes .
 * The key is extracted from each element by applying the { @ code keyExtractor }
 * { @ link Function } . In case several elements map to the same key , the associated value
 * will be the most recently emitted element .
 * < img class = " marble " src = " doc - files / marbles / collectMapWithKeyExtractor . svg " alt = " " >
 * @ reactor . discard This operator discards the buffer upon cancellation or error triggered by a data signal .
 * @ param keyExtractor a { @ link Function } to map elements to a key for the { @ link Map }
 * @ param < K > the type of the key extracted from each source element
 * @ return a { @ link Mono } of a { @ link Map } of key - element pairs ( only including latest
 * element in case of key conflicts ) */
public final < K > Mono < Map < K , T > > collectMap ( Function < ? super T , ? extends K > keyExtractor ) { } }

implementation: return collectMap ( keyExtractor , identityFunction ( ) ) ;


signature:
public class Flux { /** * Collect all elements emitted by this { @ link Flux } into a hashed { @ link Map } that is
 * emitted by the resulting { @ link Mono } when this sequence completes .
 * The key is extracted from each element by applying the { @ code keyExtractor }
 * { @ link Function } , and the value is extracted by the { @ code valueExtractor } Function .
 * In case several elements map to the same key , the associated value will be derived
 * from the most recently emitted element .
 * < img class = " marble " src = " doc - files / marbles / collectMapWithKeyAndValueExtractors . svg " alt = " " >
 * @ reactor . discard This operator discards the buffer upon cancellation or error triggered by a data signal .
 * @ param keyExtractor a { @ link Function } to map elements to a key for the { @ link Map }
 * @ param valueExtractor a { @ link Function } to map elements to a value for the { @ link Map }
 * @ param < K > the type of the key extracted from each source element
 * @ param < V > the type of the value extracted from each source element
 * @ return a { @ link Mono } of a { @ link Map } of key - element pairs ( only including latest
 * element ' s value in case of key conflicts ) */
public final < K , V > Mono < Map < K , V > > collectMap ( Function < ? super T , ? extends K > keyExtractor , Function < ? super T , ? extends V > valueExtractor ) { } }

implementation: return collectMap ( keyExtractor , valueExtractor , ( ) -> new HashMap < > ( ) ) ;


signature:
public class Flux { /** * Collect all elements emitted by this { @ link Flux } into a user - defined { @ link Map } that is
 * emitted by the resulting { @ link Mono } when this sequence completes .
 * The key is extracted from each element by applying the { @ code keyExtractor }
 * { @ link Function } , and the value is extracted by the { @ code valueExtractor } Function .
 * In case several elements map to the same key , the associated value will be derived
 * from the most recently emitted element .
 * < img class = " marble " src = " doc - files / marbles / collectMapWithKeyAndValueExtractors . svg " alt = " " >
 * @ reactor . discard This operator discards the buffer upon cancellation or error triggered by a data signal .
 * @ param keyExtractor a { @ link Function } to map elements to a key for the { @ link Map }
 * @ param valueExtractor a { @ link Function } to map elements to a value for the { @ link Map }
 * @ param mapSupplier a { @ link Map } factory called for each { @ link Subscriber }
 * @ param < K > the type of the key extracted from each source element
 * @ param < V > the type of the value extracted from each source element
 * @ return a { @ link Mono } of a { @ link Map } of key - value pairs ( only including latest
 * element ' s value in case of key conflicts ) */
public final < K , V > Mono < Map < K , V > > collectMap ( final Function < ? super T , ? extends K > keyExtractor , final Function < ? super T , ? extends V > valueExtractor , Supplier < Map < K , V > > mapSupplier ) { } }

implementation: Objects . requireNonNull ( keyExtractor , "Key extractor is null" ) ; Objects . requireNonNull ( valueExtractor , "Value extractor is null" ) ; Objects . requireNonNull ( mapSupplier , "Map supplier is null" ) ; return collect ( mapSupplier , ( m , d ) -> m . put ( keyExtractor . apply ( d ) , valueExtractor . apply ( d ) ) ) ;


signature:
public class Flux { /** * Collect all elements emitted by this { @ link Flux } into a { @ link Map multimap } that is
 * emitted by the resulting { @ link Mono } when this sequence completes .
 * The key is extracted from each element by applying the { @ code keyExtractor }
 * { @ link Function } , and every element mapping to the same key is stored in the { @ link List }
 * associated to said key .
 * < img class = " marble " src = " doc - files / marbles / collectMultiMapWithKeyExtractor . svg " alt = " " >
 * @ reactor . discard This operator discards the buffer upon cancellation or error triggered by a data signal .
 * @ param keyExtractor a { @ link Function } to map elements to a key for the { @ link Map }
 * @ param < K > the type of the key extracted from each source element
 * @ return a { @ link Mono } of a { @ link Map } of key - List ( elements ) pairs */
public final < K > Mono < Map < K , Collection < T > > > collectMultimap ( Function < ? super T , ? extends K > keyExtractor ) { } }

implementation: return collectMultimap ( keyExtractor , identityFunction ( ) ) ;


signature:
public class Flux { /** * Collect all elements emitted by this { @ link Flux } into a { @ link Map multimap } that is
 * emitted by the resulting { @ link Mono } when this sequence completes .
 * The key is extracted from each element by applying the { @ code keyExtractor }
 * { @ link Function } , and every element mapping to the same key is converted by the
 * { @ code valueExtractor } Function to a value stored in the { @ link List } associated to
 * said key .
 * < img class = " marble " src = " doc - files / marbles / collectMultiMapWithKeyAndValueExtractors . svg " alt = " " >
 * @ reactor . discard This operator discards the buffer upon cancellation or error triggered by a data signal .
 * @ param keyExtractor a { @ link Function } to map elements to a key for the { @ link Map }
 * @ param valueExtractor a { @ link Function } to map elements to a value for the { @ link Map }
 * @ param < K > the type of the key extracted from each source element
 * @ param < V > the type of the value extracted from each source element
 * @ return a { @ link Mono } of a { @ link Map } of key - List ( values ) pairs */
public final < K , V > Mono < Map < K , Collection < V > > > collectMultimap ( Function < ? super T , ? extends K > keyExtractor , Function < ? super T , ? extends V > valueExtractor ) { } }

implementation: return collectMultimap ( keyExtractor , valueExtractor , ( ) -> new HashMap < > ( ) ) ;


signature:
public class Flux { /** * Collect all elements emitted by this { @ link Flux } into a user - defined { @ link Map multimap } that is
 * emitted by the resulting { @ link Mono } when this sequence completes .
 * The key is extracted from each element by applying the { @ code keyExtractor }
 * { @ link Function } , and every element mapping to the same key is converted by the
 * { @ code valueExtractor } Function to a value stored in the { @ link Collection } associated to
 * said key .
 * < img class = " marble " src = " doc - files / marbles / collectMultiMapWithKeyAndValueExtractors . svg " alt = " " >
 * @ reactor . discard This operator discards the buffer upon cancellation or error triggered by a data signal .
 * @ param keyExtractor a { @ link Function } to map elements to a key for the { @ link Map }
 * @ param valueExtractor a { @ link Function } to map elements to a value for the { @ link Map }
 * @ param mapSupplier a multimap ( { @ link Map } of { @ link Collection } ) factory called
 * for each { @ link Subscriber }
 * @ param < K > the type of the key extracted from each source element
 * @ param < V > the type of the value extracted from each source element
 * @ return a { @ link Mono } of a { @ link Map } of key - Collection ( values ) pairs */
public final < K , V > Mono < Map < K , Collection < V > > > collectMultimap ( final Function < ? super T , ? extends K > keyExtractor , final Function < ? super T , ? extends V > valueExtractor , Supplier < Map < K , Collection < V > > > mapSupplier ) { } }

implementation: Objects . requireNonNull ( keyExtractor , "Key extractor is null" ) ; Objects . requireNonNull ( valueExtractor , "Value extractor is null" ) ; Objects . requireNonNull ( mapSupplier , "Map supplier is null" ) ; return collect ( mapSupplier , ( m , d ) -> { K key = keyExtractor . apply ( d ) ; Collection < V > values = m . computeIfAbsent ( key , k -> new ArrayList < > ( ) ) ; values . add ( valueExtractor . apply ( d ) ) ; } ) ;


signature:
public class Flux { /** * Collect all elements emitted by this { @ link Flux } until this sequence completes ,
 * and then sort them using a { @ link Comparator } into a { @ link List } that is emitted
 * by the resulting { @ link Mono } .
 * < img class = " marble " src = " doc - files / marbles / collectSortedListWithComparator . svg " alt = " " >
 * @ reactor . discard This operator discards the buffer upon cancellation or error triggered by a data signal .
 * @ param comparator a { @ link Comparator } to sort the items of this sequences
 * @ return a { @ link Mono } of a sorted { @ link List } of all values from this { @ link Flux } */
@ SuppressWarnings ( { } }

implementation:
"unchecked" , "rawtypes" } ) public final Mono < List < T > > collectSortedList ( @ Nullable Comparator < ? super T > comparator ) { return collectList ( ) . map ( list -> { // Note : this assumes the list emitted by buffer ( ) is mutable
if ( comparator != null ) { list . sort ( comparator ) ; } else { List < Comparable > l = ( List < Comparable > ) list ; Collections . sort ( l ) ; } return list ; } ) ;


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

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


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 } , sequentially and
 * preserving order using concatenation .
 * There are three dimensions to this operator that can be compared with
 * { @ link # flatMap ( Function ) flatMap } and { @ link # flatMapSequential ( Function ) flatMapSequential } :
 * < ul >
 * < li > < b > Generation of inners and subscription < / b > : this operator waits for one
 * inner to complete before generating the next one and subscribing to it . < / li >
 * < li > < b > Ordering of the flattened values < / b > : this operator naturally preserves
 * the same order as the source elements , concatenating the inners from each source
 * element sequentially . < / li >
 * < li > < b > Interleaving < / b > : this operator does not let values from different inners
 * interleave ( concatenation ) . < / li >
 * < / ul >
 * Errors will immediately short circuit current concat backlog . The prefetch argument
 * allows to give an arbitrary prefetch size to the inner { @ link Publisher } .
 * < img class = " marble " src = " doc - files / marbles / concatMap . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure upon cancellation .
 * @ param mapper the function to transform this sequence of T into concatenated sequences of V
 * @ param prefetch the inner source produced demand
 * @ param < V > the produced concatenated type
 * @ return a concatenated { @ link Flux } */
public final < V > Flux < V > concatMap ( Function < ? super T , ? extends Publisher < ? extends V > > mapper , int prefetch ) { } }

implementation: return onAssembly ( new FluxConcatMap < > ( this , mapper , Queues . get ( prefetch ) , prefetch , FluxConcatMap . ErrorMode . IMMEDIATE ) ) ;


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 } , sequentially and
 * preserving order using concatenation .
 * There are three dimensions to this operator that can be compared with
 * { @ link # flatMap ( Function ) flatMap } and { @ link # flatMapSequential ( Function ) flatMapSequential } :
 * < ul >
 * < li > < b > Generation of inners and subscription < / b > : this operator waits for one
 * inner to complete before generating the next one and subscribing to it . < / li >
 * < li > < b > Ordering of the flattened values < / b > : this operator naturally preserves
 * the same order as the source elements , concatenating the inners from each source
 * element sequentially . < / li >
 * < li > < b > Interleaving < / b > : this operator does not let values from different inners
 * interleave ( concatenation ) . < / li >
 * < / ul >
 * Errors in the individual publishers will be delayed at the end of the whole concat
 * sequence ( possibly getting combined into a { @ link Exceptions # isMultiple ( Throwable ) composite }
 * if several sources error .
 * < img class = " marble " src = " doc - files / marbles / concatMap . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure upon cancellation .
 * @ param mapper the function to transform this sequence of T into concatenated sequences of V
 * @ param < V > the produced concatenated type
 * @ return a concatenated { @ link Flux } */
public final < V > Flux < V > concatMapDelayError ( Function < ? super T , ? extends Publisher < ? extends V > > mapper ) { } }

implementation: return concatMapDelayError ( mapper , 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 } , sequentially and
 * preserving order using concatenation .
 * There are three dimensions to this operator that can be compared with
 * { @ link # flatMap ( Function ) flatMap } and { @ link # flatMapSequential ( Function ) flatMapSequential } :
 * < ul >
 * < li > < b > Generation of inners and subscription < / b > : this operator waits for one
 * inner to complete before generating the next one and subscribing to it . < / li >
 * < li > < b > Ordering of the flattened values < / b > : this operator naturally preserves
 * the same order as the source elements , concatenating the inners from each source
 * element sequentially . < / li >
 * < li > < b > Interleaving < / b > : this operator does not let values from different inners
 * interleave ( concatenation ) . < / li >
 * < / ul >
 * Errors in the individual publishers will be delayed at the end of the whole concat
 * sequence ( possibly getting combined into a { @ link Exceptions # isMultiple ( Throwable ) composite }
 * if several sources error .
 * The prefetch argument allows to give an arbitrary prefetch size to the inner { @ link Publisher } .
 * < img class = " marble " src = " doc - files / marbles / concatMap . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure upon cancellation .
 * @ param mapper the function to transform this sequence of T into concatenated sequences of V
 * @ param prefetch the inner source produced demand
 * @ param < V > the produced concatenated type
 * @ return a concatenated { @ link Flux } */
public final < V > Flux < V > concatMapDelayError ( Function < ? super T , ? extends Publisher < ? extends V > > mapper , int prefetch ) { } }

implementation: return onAssembly ( new FluxConcatMap < > ( this , mapper , Queues . get ( prefetch ) , prefetch , FluxConcatMap . ErrorMode . END ) ) ;


signature:
public class Flux { /** * Transform the items emitted by this { @ link Flux } into { @ link Iterable } , then flatten the elements from those by
 * concatenating them into a single { @ link Flux } .
 * < img class = " marble " src = " doc - files / marbles / concatMapIterable . 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 it internally queued for backpressure upon cancellation .
 * @ 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 > concatMapIterable ( Function < ? super T , ? extends Iterable < ? extends R > > mapper ) { } }

implementation: return concatMapIterable ( mapper , Queues . XS_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Transform the items emitted by this { @ link Flux } into { @ link Iterable } , then flatten the emissions from those by
 * concatenating them into a single { @ link Flux } . The prefetch argument allows to give an arbitrary prefetch size to the merged { @ link Iterable } .
 * < img class = " marble " src = " doc - files / marbles / concatMapIterable . 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 it internally queued for backpressure upon cancellation .
 * @ param mapper the { @ link Function } to transform input sequence into N { @ link Iterable }
 * @ param prefetch the maximum in - flight elements from each inner { @ link Iterable } sequence
 * @ 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 > concatMapIterable ( Function < ? super T , ? extends Iterable < ? extends R > > mapper , int prefetch ) { } }

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


signature:
public class Flux { /** * Concatenate emissions of this { @ link Flux } with the provided { @ link Publisher } ( no interleave ) .
 * < img class = " marble " src = " doc - files / marbles / concatWithForFlux . svg " alt = " " >
 * @ param other the { @ link Publisher } sequence to concat after this { @ link Flux }
 * @ return a concatenated { @ link Flux } */
public final Flux < T > concatWith ( Publisher < ? extends T > other ) { } }

implementation: if ( this instanceof FluxConcatArray ) { @ SuppressWarnings ( { "unchecked" } ) FluxConcatArray < T > fluxConcatArray = ( FluxConcatArray < T > ) this ; return fluxConcatArray . concatAdditionalSourceLast ( other ) ; } return concat ( this , other ) ;


signature:
public class Flux { /** * Delay each of this { @ link Flux } elements ( { @ link Subscriber # onNext } signals )
 * by a given { @ link Duration } . Signals are delayed and continue on an user - specified
 * { @ link Scheduler } , but empty sequences or immediate error signals are not delayed .
 * < img class = " marble " src = " doc - files / marbles / delayElements . svg " alt = " " >
 * @ param delay period to delay each { @ link Subscriber # onNext } signal
 * @ param timer a time - capable { @ link Scheduler } instance to delay each signal on
 * @ return a delayed { @ link Flux } */
public final Flux < T > delayElements ( Duration delay , Scheduler timer ) { } }

implementation: return delayUntil ( d -> Mono . delay ( delay , timer ) ) ;


signature:
public class Flux { /** * Shift this { @ link Flux } forward in time by a given { @ link Duration } .
 * Unlike with { @ link # delayElements ( Duration , Scheduler ) } , elements are shifted forward in time
 * as they are emitted , always resulting in the delay between two elements being
 * the same as in the source ( only the first element is visibly delayed from the
 * previous event , that is the subscription ) .
 * Signals are delayed and continue on an user - specified { @ link Scheduler } , but empty
 * sequences or immediate error signals are not delayed .
 * With this operator , a source emitting at 10Hz with a delaySequence { @ link Duration }
 * of 1s will still emit at 10Hz , with an initial " hiccup " of 1s .
 * On the other hand , { @ link # delayElements ( Duration , Scheduler ) } would end up emitting
 * at 1Hz .
 * This is closer to { @ link # delaySubscription ( Duration , Scheduler ) } , except the source
 * is subscribed to immediately .
 * < img class = " marble " src = " doc - files / marbles / delaySequence . svg " alt = " " >
 * @ param delay { @ link Duration } to shift the sequence by
 * @ param timer a time - capable { @ link Scheduler } instance to delay signals on
 * @ return an shifted { @ link Flux } emitting at the same frequency as the source */
public final Flux < T > delaySequence ( Duration delay , Scheduler timer ) { } }

implementation: return onAssembly ( new FluxDelaySequence < > ( this , delay , timer ) ) ;


signature:
public class Flux { /** * Subscribe to this { @ link Flux } and generate a { @ link Publisher } from each of this
 * Flux elements , each acting as a trigger for relaying said element .
 * That is to say , the resulting { @ link Flux } delays each of its emission until the
 * associated trigger Publisher terminates .
 * In case of an error either in the source or in a trigger , that error is propagated
 * immediately downstream .
 * Note that unlike with the { @ link Mono # delayUntil ( Function ) Mono variant } there is
 * no fusion of subsequent calls .
 * < img class = " marble " src = " doc - files / marbles / delayUntilForFlux . svg " alt = " " >
 * @ param triggerProvider a { @ link Function } that maps each element into a
 * { @ link Publisher } whose termination will trigger relaying the value .
 * @ return this Flux , but with elements delayed until their derived publisher terminates . */
public final Flux < T > delayUntil ( Function < ? super T , ? extends Publisher < ? > > triggerProvider ) { } }

implementation: return concatMap ( v -> Mono . just ( v ) . delayUntil ( triggerProvider ) ) ;


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

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


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

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


signature:
public class Flux { /** * For each { @ link Subscriber } , track elements from this { @ link Flux } that have been
 * seen and filter out duplicates , as compared by a key extracted through the user
 * provided { @ link Function } .
 * < img class = " marble " src = " doc - files / marbles / distinctWithKey . svg " alt = " " >
 * @ reactor . discard This operator discards elements that don ' t match the distinct predicate ,
 * but you should use the version with a cleanup if you need discarding of keys
 * categorized by the operator as " seen " . See { @ link # distinct ( Function , Supplier , BiPredicate , Consumer ) } .
 * @ param keySelector function to compute comparison key for each element
 * @ param < V > the type of the key extracted from each value in this sequence
 * @ return a filtering { @ link Flux } only emitting values with distinct keys */
public final < V > Flux < T > distinct ( Function < ? super T , ? extends V > keySelector ) { } }

implementation: return distinct ( keySelector , hashSetSupplier ( ) ) ;


signature:
public class Flux { /** * For each { @ link Subscriber } , track elements from this { @ link Flux } that have been
 * seen and filter out duplicates , as compared by a key extracted through the user
 * provided { @ link Function } and by the { @ link Collection # add ( Object ) add method }
 * of the { @ link Collection } supplied ( typically a { @ link Set } ) .
 * < img class = " marble " src = " doc - files / marbles / distinctWithKey . svg " alt = " " >
 * @ reactor . discard This operator discards elements that don ' t match the distinct predicate ,
 * but you should use the version with a cleanup if you need discarding of keys
 * categorized by the operator as " seen " . See { @ link # distinct ( Function , Supplier , BiPredicate , Consumer ) } .
 * @ param keySelector function to compute comparison key for each element
 * @ param distinctCollectionSupplier supplier of the { @ link Collection } used for distinct
 * check through { @ link Collection # add ( Object ) add } of the key .
 * @ param < V > the type of the key extracted from each value in this sequence
 * @ param < C > the type of Collection used for distinct checking of keys
 * @ return a filtering { @ link Flux } only emitting values with distinct keys */
public final < V , C extends Collection < ? super V > > Flux < T > distinct ( Function < ? super T , ? extends V > keySelector , Supplier < C > distinctCollectionSupplier ) { } }

implementation: return this . distinct ( keySelector , distinctCollectionSupplier , Collection :: add , Collection :: clear ) ;


signature:
public class Flux { /** * For each { @ link Subscriber } , track elements from this { @ link Flux } that have been
 * seen and filter out duplicates , as compared by applying a { @ link BiPredicate } on
 * an arbitrary user - supplied { @ code < C > } store and a key extracted through the user
 * provided { @ link Function } . The BiPredicate should typically add the key to the
 * arbitrary store for further comparison . A cleanup callback is also invoked on the
 * store upon termination of the sequence .
 * < img class = " marble " src = " doc - files / marbles / distinctWithKey . svg " alt = " " >
 * @ reactor . discard This operator discards elements that don ' t match the distinct predicate ,
 * but you should use the { @ code cleanup } as well if you need discarding of keys
 * categorized by the operator as " seen " .
 * @ param keySelector function to compute comparison key for each element
 * @ param distinctStoreSupplier supplier of the arbitrary store object used in distinct
 * checks along the extracted key .
 * @ param distinctPredicate the { @ link BiPredicate } to apply to the arbitrary store +
 * extracted key to perform a distinct check . Since nothing is assumed of the store ,
 * this predicate should also add the key to the store as necessary .
 * @ param cleanup the cleanup callback to invoke on the store upon termination .
 * @ param < V > the type of the key extracted from each value in this sequence
 * @ param < C > the type of store backing the { @ link BiPredicate }
 * @ return a filtering { @ link Flux } only emitting values with distinct keys */
public final < V , C > Flux < T > distinct ( Function < ? super T , ? extends V > keySelector , Supplier < C > distinctStoreSupplier , BiPredicate < C , V > distinctPredicate , Consumer < C > cleanup ) { } }

implementation: if ( this instanceof Fuseable ) { return onAssembly ( new FluxDistinctFuseable < > ( this , keySelector , distinctStoreSupplier , distinctPredicate , cleanup ) ) ; } return onAssembly ( new FluxDistinct < > ( this , keySelector , distinctStoreSupplier , distinctPredicate , cleanup ) ) ;


signature:
public class Flux { /** * Filter out subsequent repetitions of an element ( that is , if they arrive right after
 * one another ) , as compared by a key extracted through the user provided { @ link Function }
 * using equality .
 * < img class = " marble " src = " doc - files / marbles / distinctUntilChangedWithKey . svg " alt = " " >
 * @ reactor . discard This operator discards elements that are considered as " already seen " .
 * The keys themselves are not discarded .
 * @ param keySelector function to compute comparison key for each element
 * @ param < V > the type of the key extracted from each value in this sequence
 * @ return a filtering { @ link Flux } with only one occurrence in a row of each element of
 * the same key ( yet element keys can repeat in the overall sequence ) */
public final < V > Flux < T > distinctUntilChanged ( Function < ? super T , ? extends V > keySelector ) { } }

implementation: return distinctUntilChanged ( keySelector , equalPredicate ( ) ) ;


signature:
public class Flux { /** * Filter out subsequent repetitions of an element ( that is , if they arrive right
 * after one another ) , as compared by a key extracted through the user provided { @ link
 * Function } and then comparing keys with the supplied { @ link BiPredicate } .
 * < img class = " marble " src = " doc - files / marbles / distinctUntilChangedWithKey . svg " alt = " " >
 * @ reactor . discard This operator discards elements that are considered as " already seen "
 * ( for which the { @ code keyComparator } returns { @ literal true } ) . The keys themselves
 * are not discarded .
 * @ param keySelector function to compute comparison key for each element
 * @ param keyComparator predicate used to compare keys .
 * @ param < V > the type of the key extracted from each value in this sequence
 * @ return a filtering { @ link Flux } with only one occurrence in a row of each element
 * of the same key for which the predicate returns true ( yet element keys can repeat
 * in the overall sequence ) */
public final < V > Flux < T > distinctUntilChanged ( Function < ? super T , ? extends V > keySelector , BiPredicate < ? super V , ? super V > keyComparator ) { } }

implementation: return onAssembly ( new FluxDistinctUntilChanged < > ( this , keySelector , keyComparator ) ) ;


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

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