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

implementation: return retryBackoff ( numRetries , firstBackoff , FluxRetryWhen . MAX_BACKOFF , 0.5d ) ;


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

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


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

implementation: return retryWhen ( FluxRetryWhen . randomExponentialBackoffFunction ( numRetries , firstBackoff , maxBackoff , jitterFactor ) ) ;


signature:
public class Flux { /** * Sample this { @ link Flux } by emitting an item corresponding to that { @ link Flux }
 * latest emitted value whenever a companion sampler { @ link Publisher } signals a value .
 * Termination of either { @ link Publisher } will result in termination for the { @ link Subscriber }
 * as well .
 * Note that if some elements are emitted just before source completion and before a
 * last sampler can trigger , the last of these elements will be emitted along with the
 * onComplete signal .
 * Both { @ link Publisher } will run in unbounded mode because the backpressure
 * would interfere with the sampling precision .
 * < img class = " marble " src = " doc - files / marbles / sampleWithSampler . svg " alt = " " >
 * @ reactor . discard This operator discards elements that are not part of the sampling .
 * @ param sampler the sampler companion { @ link Publisher }
 * @ param < U > the type of the sampler sequence
 * @ return a { @ link Flux } sampled to the last item observed each time the sampler { @ link Publisher } signals */
public final < U > Flux < T > sample ( Publisher < U > sampler ) { } }

implementation: return onAssembly ( new FluxSample < > ( this , sampler ) ) ;


signature:
public class Flux { /** * Repeatedly take a value from this { @ link Flux } then skip the values that follow
 * within a given duration .
 * < img class = " marble " src = " doc - files / marbles / sampleFirstAtRegularInterval . svg " alt = " " >
 * @ reactor . discard This operator discards elements that are not part of the sampling .
 * @ param timespan the duration during which to skip values after each sample
 * @ return a { @ link Flux } sampled to the first item of each duration - based window */
public final Flux < T > sampleFirst ( Duration timespan ) { } }

implementation: return sampleFirst ( t -> Mono . delay ( timespan ) ) ;


signature:
public class Flux { /** * Repeatedly take a value from this { @ link Flux } then skip the values that follow
 * before the next signal from a companion sampler { @ link Publisher } .
 * < img class = " marble " src = " doc - files / marbles / sampleFirstWithSamplerFactory . svg " alt = " " >
 * @ reactor . discard This operator discards elements that are not part of the sampling .
 * @ param samplerFactory supply a companion sampler { @ link Publisher } which signals the end of the skip window
 * @ param < U > the companion reified type
 * @ return a { @ link Flux } sampled to the first item observed in each window closed by the sampler signals */
public final < U > Flux < T > sampleFirst ( Function < ? super T , ? extends Publisher < U > > samplerFactory ) { } }

implementation: return onAssembly ( new FluxSampleFirst < > ( this , samplerFactory ) ) ;


signature:
public class Flux { /** * Emit the latest value from this { @ link Flux } only if there were no new values emitted
 * during the window defined by a companion { @ link Publisher } derived from that particular
 * value .
 * Note that this means that the last value in the sequence is always emitted .
 * < img class = " marble " src = " doc - files / marbles / sampleTimeoutWithThrottlerFactory . svg " alt = " " >
 * @ reactor . discard This operator discards elements that are not part of the sampling .
 * @ param throttlerFactory supply a companion sampler { @ link Publisher } which signals
 * the end of the window during which no new emission should occur . If it is the case ,
 * the original value triggering the window is emitted .
 * @ param < U > the companion reified type
 * @ return a { @ link Flux } sampled to items not followed by any other item within a window
 * defined by a companion { @ link Publisher } */
public final < U > Flux < T > sampleTimeout ( Function < ? super T , ? extends Publisher < U > > throttlerFactory ) { } }

implementation: return sampleTimeout ( throttlerFactory , Queues . XS_BUFFER_SIZE ) ;


signature:
public class Flux { /** * FIXME re - evaluate the wording on maxConcurrency , seems more related to request / buffering . If so redo the marble */
public final < U > Flux < T > sampleTimeout ( Function < ? super T , ? extends Publisher < U > > throttlerFactory , int maxConcurrency ) { } }

implementation: return onAssembly ( new FluxSampleTimeout < > ( this , throttlerFactory , Queues . get ( maxConcurrency ) ) ) ;


signature:
public class Flux { /** * Reduce this { @ link Flux } values with an accumulator { @ link BiFunction } and
 * also emit the intermediate results of this function .
 * Unlike { @ link # scan ( Object , BiFunction ) } , this operator doesn ' t take an initial value
 * but treats the first { @ link Flux } value as initial value .
 * < br >
 * The accumulation works as follows :
 * < pre > < code >
 * result [ 0 ] = source [ 0]
 * result [ 1 ] = accumulator ( result [ 0 ] , source [ 1 ] )
 * result [ 2 ] = accumulator ( result [ 1 ] , source [ 2 ] )
 * result [ 3 ] = accumulator ( result [ 2 ] , source [ 3 ] )
 * < / code > < / pre >
 * < img class = " marble " src = " doc - files / marbles / scanWithSameReturnType . svg " alt = " " >
 * @ param accumulator the accumulating { @ link BiFunction }
 * @ return an accumulating { @ link Flux } */
public final Flux < T > scan ( BiFunction < T , T , T > accumulator ) { } }

implementation: return onAssembly ( new FluxScan < > ( this , accumulator ) ) ;


signature:
public class Flux { /** * Reduce this { @ link Flux } values with an accumulator { @ link BiFunction } and
 * also emit the intermediate results of this function .
 * The accumulation works as follows :
 * < pre > < code >
 * result [ 0 ] = initialValue ;
 * result [ 1 ] = accumulator ( result [ 0 ] , source [ 0 ] )
 * result [ 2 ] = accumulator ( result [ 1 ] , source [ 1 ] )
 * result [ 3 ] = accumulator ( result [ 2 ] , source [ 2 ] )
 * < / code > < / pre >
 * < img class = " marble " src = " doc - files / marbles / scan . svg " alt = " " >
 * @ param initial the initial leftmost argument to pass to the reduce function
 * @ param accumulator the accumulating { @ link BiFunction }
 * @ param < A > the accumulated type
 * @ return an accumulating { @ link Flux } starting with initial state */
public final < A > Flux < A > scan ( A initial , BiFunction < A , ? super T , A > accumulator ) { } }

implementation: Objects . requireNonNull ( initial , "seed" ) ; return scanWith ( ( ) -> initial , accumulator ) ;


signature:
public class Flux { /** * Reduce this { @ link Flux } values with the help of an accumulator { @ link BiFunction }
 * and also emits the intermediate results . A seed value is lazily provided by a
 * { @ link Supplier } invoked for each { @ link Subscriber } .
 * The accumulation works as follows :
 * < pre > < code >
 * result [ 0 ] = initialValue ;
 * result [ 1 ] = accumulator ( result [ 0 ] , source [ 0 ] )
 * result [ 2 ] = accumulator ( result [ 1 ] , source [ 1 ] )
 * result [ 3 ] = accumulator ( result [ 2 ] , source [ 2 ] )
 * < / code > < / pre >
 * < img class = " marble " src = " doc - files / marbles / scanWith . svg " alt = " " >
 * @ param initial the supplier providing the seed , the leftmost parameter initially
 * passed to the reduce function
 * @ param accumulator the accumulating { @ link BiFunction }
 * @ param < A > the accumulated type
 * @ return an accumulating { @ link Flux } starting with initial state */
public final < A > Flux < A > scanWith ( Supplier < A > initial , BiFunction < A , ? super T , A > accumulator ) { } }

implementation: return onAssembly ( new FluxScanSeed < > ( this , initial , accumulator ) ) ;


signature:
public class Flux { /** * Returns a new { @ link Flux } that multicasts ( shares ) the original { @ link Flux } .
 * As long as there is at least one { @ link Subscriber } this { @ link Flux } will be subscribed and
 * emitting data .
 * When all subscribers have cancelled it will cancel the source
 * { @ link Flux } .
 * < p > This is an alias for { @ link # publish ( ) } . { @ link ConnectableFlux # refCount ( ) } .
 * @ return a { @ link Flux } that upon first subscribe causes the source { @ link Flux }
 * to subscribe once , late subscribers might therefore miss items . */
public final Flux < T > share ( ) { } }

implementation: return onAssembly ( new FluxRefCount < > ( new FluxPublish < > ( this , Queues . SMALL_BUFFER_SIZE , Queues . small ( ) ) , 1 ) ) ;


signature:
public class Flux { /** * Expect and emit a single item from this { @ link Flux } source , and accept an empty
 * source but signal an { @ link IndexOutOfBoundsException } for a source with more than
 * one element .
 * < img class = " marble " src = " doc - files / marbles / singleOrEmpty . svg " alt = " " >
 * @ return a { @ link Mono } with the expected single item , no item or an error */
public final Mono < T > singleOrEmpty ( ) { } }

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


signature:
public class Flux { /** * Skip elements from this { @ link Flux } emitted within the specified initial duration ,
 * as measured on the provided { @ link Scheduler } .
 * < img class = " marble " src = " doc - files / marbles / skipWithTimespan . svg " alt = " " >
 * @ reactor . discard This operator discards elements that are skipped .
 * @ param timespan the initial time window during which to drop elements
 * @ param timer a time - capable { @ link Scheduler } instance to measure the time window on
 * @ return a { @ link Flux } dropping at the beginning for the given duration */
public final Flux < T > skip ( Duration timespan , Scheduler timer ) { } }

implementation: if ( ! timespan . isZero ( ) ) { return skipUntilOther ( Mono . delay ( timespan , timer ) ) ; } else { return this ; }


signature:
public class Flux { /** * Sort elements from this { @ link Flux } using a { @ link Comparator } function , by
 * collecting and sorting elements in the background then emitting the sorted sequence
 * once this sequence completes .
 * < p > Note that calling { @ code sort } with long , non - terminating or infinite sources
 * might cause { @ link OutOfMemoryError }
 * < img class = " marble " src = " doc - files / marbles / sort . svg " alt = " " >
 * @ param sortFunction a function that compares two items emitted by this { @ link Flux }
 * to indicate their sort order
 * @ return a sorted { @ link Flux } */
public final Flux < T > sort ( Comparator < ? super T > sortFunction ) { } }

implementation: return collectSortedList ( sortFunction ) . flatMapIterable ( identityFunction ( ) ) ;


signature:
public class Flux { /** * Prepend the given { @ link Publisher } sequence to this { @ link Flux } sequence .
 * < img class = " marble " src = " doc - files / marbles / startWithPublisher . svg " alt = " " >
 * @ param publisher the Publisher whose values to prepend
 * @ return a new { @ link Flux } prefixed with the given { @ link Publisher } sequence */
public final Flux < T > startWith ( Publisher < ? extends T > publisher ) { } }

implementation: if ( this instanceof FluxConcatArray ) { FluxConcatArray < T > fluxConcatArray = ( FluxConcatArray < T > ) this ; return fluxConcatArray . concatAdditionalSourceFirst ( publisher ) ; } return concat ( publisher , this ) ;


signature:
public class Flux { /** * Subscribe a { @ link Consumer } to this { @ link Flux } that will consume all the
 * elements in the sequence . It will request an unbounded demand ( { @ code Long . MAX _ VALUE } ) .
 * For a passive version that observe and forward incoming data see { @ link # doOnNext ( java . util . function . Consumer ) } .
 * For a version that gives you more control over backpressure and the request , see
 * { @ link # subscribe ( Subscriber ) } with a { @ link BaseSubscriber } .
 * Keep in mind that since the sequence can be asynchronous , this will immediately
 * return control to the calling thread . This can give the impression the consumer is
 * not invoked when executing in a main thread or a unit test for instance .
 * < img class = " marble " src = " doc - files / marbles / subscribeWithOnNextForFlux . svg " alt = " " >
 * @ param consumer the consumer to invoke on each value ( onNext signal )
 * @ return a new { @ link Disposable } that can be used to cancel the underlying { @ link Subscription } */
public final Disposable subscribe ( Consumer < ? super T > consumer ) { } }

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


signature:
public class Flux { /** * Subscribe to this { @ link Flux } with a { @ link Consumer } that will consume all the
 * elements in the sequence , as well as a { @ link Consumer } that will handle errors .
 * The subscription will request an unbounded demand ( { @ code Long . MAX _ VALUE } ) .
 * For a passive version that observe and forward incoming data see
 * { @ link # doOnNext ( java . util . function . Consumer ) } and { @ link # doOnError ( java . util . function . Consumer ) } .
 * < p > For a version that gives you more control over backpressure and the request , see
 * { @ link # subscribe ( Subscriber ) } with a { @ link BaseSubscriber } .
 * Keep in mind that since the sequence can be asynchronous , this will immediately
 * return control to the calling thread . This can give the impression the consumers are
 * not invoked when executing in a main thread or a unit test for instance .
 * < img class = " marble " src = " doc - files / marbles / subscribeWithOnNextAndOnErrorForFlux . svg " alt = " " >
 * @ param consumer the consumer to invoke on each next signal
 * @ param errorConsumer the consumer to invoke on error signal
 * @ return a new { @ link Disposable } that can be used to cancel the underlying { @ link Subscription } */
public final Disposable subscribe ( @ Nullable Consumer < ? super T > consumer , Consumer < ? super Throwable > errorConsumer ) { } }

implementation: Objects . requireNonNull ( errorConsumer , "errorConsumer" ) ; return subscribe ( consumer , errorConsumer , null ) ;


signature:
public class Flux { /** * Subscribe { @ link Consumer } to this { @ link Flux } that will respectively consume all the
 * elements in the sequence , handle errors and react to completion . The subscription
 * will request unbounded demand ( { @ code Long . MAX _ VALUE } ) .
 * For a passive version that observe and forward incoming data see { @ link # doOnNext ( java . util . function . Consumer ) } ,
 * { @ link # doOnError ( java . util . function . Consumer ) } and { @ link # doOnComplete ( Runnable ) } .
 * < p > For a version that gives you more control over backpressure and the request , see
 * { @ link # subscribe ( Subscriber ) } with a { @ link BaseSubscriber } .
 * Keep in mind that since the sequence can be asynchronous , this will immediately
 * return control to the calling thread . This can give the impression the consumer is
 * not invoked when executing in a main thread or a unit test for instance .
 * < img class = " marble " src = " doc - files / marbles / subscribeWithOnNextAndOnErrorAndOnCompleteForFlux . svg " alt = " " >
 * @ param consumer the consumer to invoke on each value
 * @ param errorConsumer the consumer to invoke on error signal
 * @ param completeConsumer the consumer to invoke on complete signal
 * @ return a new { @ link Disposable } that can be used to cancel the underlying { @ link Subscription } */
public final Disposable subscribe ( @ Nullable Consumer < ? super T > consumer , @ Nullable Consumer < ? super Throwable > errorConsumer , @ Nullable Runnable completeConsumer ) { } }

implementation: return subscribe ( consumer , errorConsumer , completeConsumer , null ) ;


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

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


signature:
public class Flux { /** * Run subscribe and onSubscribe on a specified { @ link Scheduler } ' s { @ link Worker } .
 * Request will be run on that worker too depending on the { @ code requestOnSeparateThread }
 * parameter ( which defaults to true in the { @ link # subscribeOn ( Scheduler ) } version ) .
 * As such , placing this operator anywhere in the chain will also impact the execution
 * context of onNext / onError / onComplete signals from the beginning of the chain up to
 * the next occurrence of a { @ link # publishOn ( Scheduler ) publishOn } .
 * Note that if you are using an eager or blocking
 * { @ link Flux # create ( Consumer , FluxSink . OverflowStrategy ) }
 * as the source , it can lead to deadlocks due to requests piling up behind the emitter .
 * Thus this operator has a { @ code requestOnSeparateThread } parameter , which should be
 * set to { @ code false } in this case .
 * < img class = " marble " src = " doc - files / marbles / subscribeOnForFlux . svg " alt = " " >
 * Typically used for slow publisher e . g . , blocking IO , fast consumer ( s ) scenarios .
 * < blockquote > < pre >
 * { @ code flux . subscribeOn ( Schedulers . single ( ) ) . subscribe ( ) }
 * < / pre > < / blockquote >
 * Note that { @ link Worker # schedule ( Runnable ) } raising
 * { @ link java . util . concurrent . RejectedExecutionException } on late
 * { @ link Subscription # request ( long ) } will be propagated to the request caller .
 * @ param scheduler a { @ link Scheduler } providing the { @ link Worker } where to subscribe
 * @ param requestOnSeparateThread whether or not to also perform requests on the worker .
 * { @ code true } to behave like { @ link # subscribeOn ( Scheduler ) }
 * @ return a { @ link Flux } requesting asynchronously
 * @ see # publishOn ( Scheduler )
 * @ see # subscribeOn ( Scheduler ) */
public final Flux < T > subscribeOn ( Scheduler scheduler , boolean requestOnSeparateThread ) { } }

implementation:
if ( this instanceof Callable ) { if ( this instanceof Fuseable . ScalarCallable ) { try { @ SuppressWarnings ( "unchecked" ) T value = ( ( Fuseable . ScalarCallable < T > ) this ) . call ( ) ; return onAssembly ( new FluxSubscribeOnValue < > ( value , scheduler ) ) ; } catch ( Exception e ) { // leave FluxSubscribeOnCallable defer error
} } @ SuppressWarnings ( "unchecked" ) Callable < T > c = ( Callable < T > ) this ; return onAssembly ( new FluxSubscribeOnCallable < > ( c , scheduler ) ) ; } return onAssembly ( new FluxSubscribeOn < > ( this , scheduler , requestOnSeparateThread ) ) ;


signature:
public class Flux { /** * Subscribe the given { @ link Subscriber } to this { @ link Flux } and return said
 * { @ link Subscriber } ( eg . a { @ link FluxProcessor } ) .
 * < blockquote > < pre >
 * { @ code flux . subscribeWith ( WorkQueueProcessor . create ( ) ) . subscribe ( ) }
 * < / pre > < / blockquote >
 * If you need more control over backpressure and the request , use a { @ link BaseSubscriber } .
 * @ param subscriber the { @ link Subscriber } to subscribe with and return
 * @ param < E > the reified type from the input / output subscriber
 * @ return the passed { @ link Subscriber } */
public final < E extends Subscriber < ? super T > > E subscribeWith ( E subscriber ) { } }

implementation: subscribe ( subscriber ) ; return subscriber ;


signature:
public class Flux { /** * Transform the current { @ link Flux < T > } once it emits its first element , making a
 * conditional transformation possible . This operator first requests one element
 * from the source then applies a transformation derived from the first { @ link Signal }
 * and the source . The whole source ( including the first signal ) is passed as second
 * argument to the { @ link BiFunction } and it is very strongly advised to always build
 * upon with operators ( see below ) .
 * Note that the source might complete or error immediately instead of emitting ,
 * in which case the { @ link Signal } would be onComplete or onError . It is NOT
 * necessarily an onNext Signal , and must be checked accordingly .
 * For example , this operator could be used to define a dynamic transformation that depends
 * on the first element ( which could contain routing metadata for instance ) :
 * < blockquote > < pre >
 * { @ code
 * fluxOfIntegers . switchOnFirst ( ( signal , flux ) - > {
 * if ( signal . hasValue ( ) ) {
 * ColoredShape firstColor = signal . get ( ) ;
 * return flux . filter ( v - > ! v . hasSameColorAs ( firstColor ) )
 * return flux ; / / either early complete or error , this forwards the termination in any case
 * / / ` return flux . onErrorResume ( t - > Mono . empty ( ) ) ; ` instead would suppress an early error
 * / / ` return Flux . just ( 1,2,3 ) ; ` instead would suppress an early error and return 1 , 2 , 3.
 * / / It would also only cancel the original ` flux ` at the completion of ` just ` .
 * < / pre > < / blockquote >
 * < img class = " marble " src = " doc - files / marbles / switchOnFirst . svg " alt = " " >
 * It is advised to return a { @ link Publisher } derived from the original { @ link Flux }
 * in all cases , as not doing so would keep the original { @ link Publisher } open and
 * hanging with a single request until the inner { @ link Publisher } terminates or
 * the whole { @ link Flux } is cancelled .
 * @ param transformer A { @ link BiFunction } executed once the first signal is
 * available and used to transform the source conditionally . The whole source ( including
 * first signal ) is passed as second argument to the BiFunction .
 * @ param < V > the item type in the returned { @ link Flux }
 * @ return a new { @ link Flux } that transform the upstream once a signal is available */
public final < V > Flux < V > switchOnFirst ( BiFunction < Signal < ? extends T > , Flux < T > , Publisher < ? extends V > > transformer ) { } }

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


signature:
public class Flux { /** * Switch to a new { @ link Publisher } generated via a { @ link Function } whenever
 * this { @ link Flux } produces an item . As such , the elements from each generated
 * Publisher are emitted in the resulting { @ link Flux } .
 * < img class = " marble " src = " doc - files / marbles / switchMap . svg " alt = " " >
 * @ param fn the { @ link Function } to generate a { @ link Publisher } for each source value
 * @ param < V > the type of the return value of the transformation function
 * @ return a new { @ link Flux } that emits values from an alternative { @ link Publisher }
 * for each source onNext */
public final < V > Flux < V > switchMap ( Function < ? super T , Publisher < ? extends V > > fn ) { } }

implementation: return switchMap ( fn , Queues . XS_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Switch to a new { @ link Publisher } generated via a { @ link Function } whenever
 * this { @ link Flux } produces an item . As such , the elements from each generated
 * Publisher are emitted in the resulting { @ link Flux } .
 * < img class = " marble " src = " doc - files / marbles / switchMap . svg " alt = " " >
 * @ param fn the { @ link Function } to generate a { @ link Publisher } for each source value
 * @ param prefetch the produced demand for inner sources
 * @ param < V > the type of the return value of the transformation function
 * @ return a new { @ link Flux } that emits values from an alternative { @ link Publisher }
 * for each source onNext */
public final < V > Flux < V > switchMap ( Function < ? super T , Publisher < ? extends V > > fn , int prefetch ) { } }

implementation: return onAssembly ( new FluxSwitchMap < > ( this , fn , Queues . unbounded ( prefetch ) , prefetch ) ) ;


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

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


signature:
public class Flux { /** * Take only the first N values from this { @ link Flux } , if available .
 * If N is zero , the resulting { @ link Flux } completes as soon as this { @ link Flux }
 * signals its first value ( which is not not relayed , though ) .
 * < img class = " marble " src = " doc - files / marbles / take . svg " alt = " " >
 * Note that this operator doesn ' t manipulate the backpressure requested amount .
 * Rather , it merely lets requests from downstream propagate as is and cancels once
 * N elements have been emitted . As a result , the source could produce a lot of
 * extraneous elements in the meantime . If that behavior is undesirable and you do
 * not own the request from downstream ( e . g . prefetching operators ) , consider
 * using { @ link # limitRequest ( long ) } instead .
 * @ param n the number of items to emit from this { @ link Flux }
 * @ return a { @ link Flux } limited to size N
 * @ see # limitRequest ( long ) */
public final Flux < T > take ( long n ) { } }

implementation: if ( this instanceof Fuseable ) { return onAssembly ( new FluxTakeFuseable < > ( this , n ) ) ; } return onAssembly ( new FluxTake < > ( this , n ) ) ;


signature:
public class Flux { /** * Relay values from this { @ link Flux } until the specified { @ link Duration } elapses ,
 * as measured on the specified { @ link Scheduler } .
 * If the duration is zero , the resulting { @ link Flux } completes as soon as this { @ link Flux }
 * signals its first value ( which is not not relayed , though ) .
 * < img class = " marble " src = " doc - files / marbles / takeWithTimespanForFlux . svg " alt = " " >
 * @ param timespan the { @ link Duration } of the time window during which to emit elements
 * from this { @ link Flux }
 * @ param timer a time - capable { @ link Scheduler } instance to run on
 * @ return a { @ link Flux } limited to elements emitted within a specific duration */
public final Flux < T > take ( Duration timespan , Scheduler timer ) { } }

implementation: if ( ! timespan . isZero ( ) ) { return takeUntilOther ( Mono . delay ( timespan , timer ) ) ; } else { return take ( 0 ) ; }


signature:
public class Flux { /** * Emit the last N values this { @ link Flux } emitted before its completion .
 * < img class = " marble " src = " doc - files / marbles / takeLast . svg " alt = " " >
 * @ param n the number of items from this { @ link Flux } to retain and emit on onComplete
 * @ return a terminating { @ link Flux } sub - sequence */
public final Flux < T > takeLast ( int n ) { } }

implementation: if ( n == 1 ) { return onAssembly ( new FluxTakeLastOne < > ( this ) ) ; } return onAssembly ( new FluxTakeLast < > ( this , n ) ) ;


signature:
public class Flux { /** * Return a { @ code Mono < Void > } that completes when this { @ link Flux } completes .
 * This will actively ignore the sequence and only replay completion or error signals .
 * < img class = " marble " src = " doc - files / marbles / thenForFlux . svg " alt = " " >
 * @ reactor . discard This operator discards elements from the source .
 * @ return a new { @ link Mono } representing the termination of this { @ link Flux } */
public final Mono < Void > then ( ) { } }

implementation: @ SuppressWarnings ( "unchecked" ) Mono < Void > then = ( Mono < Void > ) new MonoIgnoreElements < > ( this ) ; return Mono . onAssembly ( then ) ;


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

implementation: return Mono . onAssembly ( new MonoIgnoreThen < > ( new Publisher [ ] { this } , other ) ) ;


signature:
public class Flux { /** * Return a { @ code Mono < Void > } that waits for this { @ link Flux } to complete then
 * for a supplied { @ link Publisher Publisher & lt ; Void & gt ; } to also complete . The
 * second completion signal is replayed , or any error signal that occurs instead .
 * < img class = " marble " src = " doc - files / marbles / thenEmptyForFlux . svg " alt = " " >
 * @ reactor . discard This operator discards elements from the source .
 * @ param other a { @ link Publisher } to wait for after this Flux ' s termination
 * @ return a new { @ link Mono } completing when both publishers have completed in
 * sequence */
public final Mono < Void > thenEmpty ( Publisher < Void > other ) { } }

implementation: return then ( Mono . fromDirect ( other ) ) ;


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

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


signature:
public class Flux { /** * Propagate a { @ link TimeoutException } as soon as no item is emitted within the
 * given { @ link Duration } from the previous emission ( or the subscription for the first item ) .
 * < img class = " marble " src = " doc - files / marbles / timeoutForFlux . svg " alt = " " >
 * @ param timeout the timeout between two signals from this { @ link Flux }
 * @ return a { @ link Flux } that can time out on a per - item basis */
public final Flux < T > timeout ( Duration timeout ) { } }

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


signature:
public class Flux { /** * Switch to a fallback { @ link Flux } as soon as no item is emitted within the
 * given { @ link Duration } from the previous emission ( or the subscription for the first item ) .
 * If the given { @ link Publisher } is null , signal a { @ link TimeoutException } instead .
 * < img class = " marble " src = " doc - files / marbles / timeoutFallbackForFlux . svg " alt = " " >
 * @ param timeout the timeout between two signals from this { @ link Flux }
 * @ param fallback the fallback { @ link Publisher } to subscribe when a timeout occurs
 * @ return a { @ link Flux } that will fallback to a different { @ link Publisher } in case of a per - item timeout */
public final Flux < T > timeout ( Duration timeout , @ Nullable Publisher < ? extends T > fallback ) { } }

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


signature:
public class Flux { /** * Propagate a { @ link TimeoutException } as soon as no item is emitted within the
 * given { @ link Duration } from the previous emission ( or the subscription for the first
 * item ) , as measured by the specified { @ link Scheduler } .
 * < img class = " marble " src = " doc - files / marbles / timeoutForFlux . svg " alt = " " >
 * @ param timeout the timeout { @ link Duration } between two signals from this { @ link Flux }
 * @ param timer a time - capable { @ link Scheduler } instance to run on
 * @ return a { @ link Flux } that can time out on a per - item basis */
public final Flux < T > timeout ( Duration timeout , Scheduler timer ) { } }

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


signature:
public class Flux { /** * Switch to a fallback { @ link Flux } as soon as no item is emitted within the
 * given { @ link Duration } from the previous emission ( or the subscription for the
 * first item ) , as measured on the specified { @ link Scheduler } .
 * If the given { @ link Publisher } is null , signal a { @ link TimeoutException } instead .
 * < img class = " marble " src = " doc - files / marbles / timeoutFallbackForFlux . svg " alt = " " >
 * @ param timeout the timeout { @ link Duration } between two signals from this { @ link Flux }
 * @ param fallback the fallback { @ link Publisher } to subscribe when a timeout occurs
 * @ param timer a time - capable { @ link Scheduler } instance to run on
 * @ return a { @ link Flux } that will fallback to a different { @ link Publisher } in case of a per - item timeout */
public final Flux < T > timeout ( Duration timeout , @ Nullable Publisher < ? extends T > fallback , Scheduler timer ) { } }

implementation: final Mono < Long > _timer = Mono . delay ( timeout , timer ) . onErrorReturn ( 0L ) ; final Function < T , Publisher < Long > > rest = o -> _timer ; if ( fallback == null ) { return timeout ( _timer , rest , timeout . toMillis ( ) + "ms" ) ; } return timeout ( _timer , rest , fallback ) ;


signature:
public class Flux { /** * Signal a { @ link TimeoutException } in case the first item from this { @ link Flux } has
 * not been emitted before the given { @ link Publisher } emits .
 * < img class = " marble " src = " doc - files / marbles / timeoutPublisher . svg " alt = " " >
 * @ param firstTimeout the companion { @ link Publisher } that will trigger a timeout if
 * emitting before the first signal from this { @ link Flux }
 * @ param < U > the type of the timeout Publisher
 * @ return a { @ link Flux } that can time out if the first item does not come before
 * a signal from a companion { @ link Publisher } */
public final < U > Flux < T > timeout ( Publisher < U > firstTimeout ) { } }

implementation: return timeout ( firstTimeout , t -> never ( ) ) ;


signature:
public class Flux { /** * Signal a { @ link TimeoutException } in case the first item from this { @ link Flux } has
 * not been emitted before the { @ code firstTimeout } { @ link Publisher } emits , and whenever
 * each subsequent elements is not emitted before a { @ link Publisher } generated from
 * the latest element signals .
 * < img class = " marble " src = " doc - files / marbles / timeoutPublisherFunctionForFlux . svg " alt = " " >
 * @ param firstTimeout the timeout { @ link Publisher } that must not emit before the first signal from this { @ link Flux }
 * @ param nextTimeoutFactory the timeout { @ link Publisher } factory for each next item
 * @ param < U > the type of the elements of the first timeout Publisher
 * @ param < V > the type of the elements of the subsequent timeout Publishers
 * @ return a { @ link Flux } that can time out if each element does not come before
 * a signal from a per - item companion { @ link Publisher } */
public final < U , V > Flux < T > timeout ( Publisher < U > firstTimeout , Function < ? super T , ? extends Publisher < V > > nextTimeoutFactory ) { } }

implementation: return timeout ( firstTimeout , nextTimeoutFactory , "first signal from a Publisher" ) ;


signature:
public class Flux { /** * Switch to a fallback { @ link Publisher } in case the first item from this { @ link Flux } has
 * not been emitted before the { @ code firstTimeout } { @ link Publisher } emits , and whenever
 * each subsequent elements is not emitted before a { @ link Publisher } generated from
 * the latest element signals .
 * < img class = " marble " src = " doc - files / marbles / timeoutPublisherFunctionAndFallbackForFlux . svg " alt = " " >
 * @ param firstTimeout the timeout { @ link Publisher } that must not emit before the first signal from this { @ link Flux }
 * @ param nextTimeoutFactory the timeout { @ link Publisher } factory for each next item
 * @ param fallback the fallback { @ link Publisher } to subscribe when a timeout occurs
 * @ param < U > the type of the elements of the first timeout Publisher
 * @ param < V > the type of the elements of the subsequent timeout Publishers
 * @ return a { @ link Flux } that can time out if each element does not come before
 * a signal from a per - item companion { @ link Publisher } */
public final < U , V > Flux < T > timeout ( Publisher < U > firstTimeout , Function < ? super T , ? extends Publisher < V > > nextTimeoutFactory , Publisher < ? extends T > fallback ) { } }

implementation: return onAssembly ( new FluxTimeout < > ( this , firstTimeout , nextTimeoutFactory , fallback ) ) ;


signature:
public class Flux { /** * Emit a { @ link reactor . util . function . Tuple2 } pair of T1 the current clock time in
 * millis ( as a { @ link Long } measured by the provided { @ link Scheduler } ) and T2
 * the emitted data ( as a { @ code T } ) , for each item from this { @ link Flux } .
 * < img class = " marble " src = " doc - files / marbles / timestampForFlux . svg " alt = " " >
 * @ param scheduler the { @ link Scheduler } to read time from
 * @ return a timestamped { @ link Flux } */
public final Flux < Tuple2 < Long , T > > timestamp ( Scheduler scheduler ) { } }

implementation: Objects . requireNonNull ( scheduler , "scheduler" ) ; return map ( d -> Tuples . of ( scheduler . now ( TimeUnit . MILLISECONDS ) , d ) ) ;


signature:
public class Flux { /** * Transform this { @ link Flux } into a lazy { @ link Iterable } blocking on
 * { @ link Iterator # next ( ) } calls .
 * < img class = " marble " src = " doc - files / marbles / toIterableWithBatchSize . svg " alt = " " >
 * Note that iterating from within threads marked as " non - blocking only " is illegal and will
 * cause an { @ link IllegalStateException } to be thrown , but obtaining the { @ link Iterable }
 * itself within these threads is ok .
 * @ param batchSize the bounded capacity to prefetch from this { @ link Flux } or
 * { @ code Integer . MAX _ VALUE } for unbounded demand
 * @ param queueProvider the supplier of the queue implementation to be used for storing
 * elements emitted faster than the iteration
 * @ return a blocking { @ link Iterable } */
public final Iterable < T > toIterable ( int batchSize , @ Nullable Supplier < Queue < T > > queueProvider ) { } }

implementation: final Supplier < Queue < T > > provider ; if ( queueProvider == null ) { provider = Queues . get ( batchSize ) ; } else { provider = queueProvider ; } return new BlockingIterable < > ( this , batchSize , provider ) ;


signature:
public class Flux { /** * Transform this { @ link Flux } into a lazy { @ link Stream } blocking for each source
 * { @ link Subscriber # onNext ( Object ) onNext } call .
 * < img class = " marble " src = " doc - files / marbles / toStreamWithBatchSize . svg " alt = " " >
 * Note that iterating from within threads marked as " non - blocking only " is illegal and will
 * cause an { @ link IllegalStateException } to be thrown , but obtaining the { @ link Stream }
 * itself or applying lazy intermediate operation on the stream within these threads is ok .
 * @ param batchSize the bounded capacity to prefetch from this { @ link Flux } or
 * { @ code Integer . MAX _ VALUE } for unbounded demand
 * @ return a { @ link Stream } of unknown size with onClose attached to { @ link Subscription # cancel ( ) } */
public final Stream < T > toStream ( int batchSize ) { } }

implementation: final Supplier < Queue < T > > provider ; provider = Queues . get ( batchSize ) ; return new BlockingIterable < > ( this , batchSize , provider ) . stream ( ) ;


signature:
public class Flux { /** * Split this { @ link Flux } sequence into multiple { @ link Flux } windows containing
 * { @ code maxSize } elements ( or less for the final window ) and starting from the first item .
 * Each { @ link Flux } window will onComplete after { @ code maxSize } items have been routed .
 * < img class = " marble " src = " doc - files / marbles / windowWithMaxSize . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure
 * upon cancellation or error triggered by a data signal .
 * @ param maxSize the maximum number of items to emit in the window before closing it
 * @ return a { @ link Flux } of { @ link Flux } windows based on element count */
public final Flux < Flux < T > > window ( int maxSize ) { } }

implementation: return onAssembly ( new FluxWindow < > ( this , maxSize , Queues . get ( maxSize ) ) ) ;


signature:
public class Flux { /** * Split this { @ link Flux } sequence into multiple { @ link Flux } windows of size
 * { @ code maxSize } , that each open every { @ code skip } elements in the source .
 * When maxSize < skip : dropping windows
 * < img class = " marble " src = " doc - files / marbles / windowWithMaxSizeLessThanSkipSize . svg " alt = " " >
 * When maxSize > skip : overlapping windows
 * < img class = " marble " src = " doc - files / marbles / windowWithMaxSizeGreaterThanSkipSize . svg " alt = " " >
 * When maxSize = = skip : exact windows
 * < img class = " marble " src = " doc - files / marbles / windowWithMaxSizeEqualsSkipSize . svg " alt = " " >
 * @ reactor . discard The overlapping variant DOES NOT discard elements , as they might be part of another still valid window .
 * The exact window and dropping window variants bot discard elements they internally queued for backpressure
 * upon cancellation or error triggered by a data signal . The dropping window variant also discards elements in between windows .
 * @ param maxSize the maximum number of items to emit in the window before closing it
 * @ param skip the number of items to count before opening and emitting a new window
 * @ return a { @ link Flux } of { @ link Flux } windows based on element count and opened every skipCount */
public final Flux < Flux < T > > window ( int maxSize , int skip ) { } }

implementation: return onAssembly ( new FluxWindow < > ( this , maxSize , skip , Queues . unbounded ( Queues . XS_BUFFER_SIZE ) , Queues . unbounded ( Queues . XS_BUFFER_SIZE ) ) ) ;


signature:
public class Flux { /** * Split this { @ link Flux } sequence into continuous , non - overlapping windows
 * where the window boundary is signalled by another { @ link Publisher }
 * < img class = " marble " src = " doc - files / marbles / windowWithBoundary . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure
 * upon cancellation or error triggered by a data signal .
 * @ param boundary a { @ link Publisher } to emit any item for a split signal and complete to terminate
 * @ return a { @ link Flux } of { @ link Flux } windows delimited by a given { @ link Publisher } */
public final Flux < Flux < T > > window ( Publisher < ? > boundary ) { } }

implementation: return onAssembly ( new FluxWindowBoundary < > ( this , boundary , Queues . unbounded ( Queues . XS_BUFFER_SIZE ) ) ) ;


signature:
public class Flux { /** * Split this { @ link Flux } sequence into continuous , non - overlapping windows that open
 * for a { @ code windowingTimespan } { @ link Duration } ( as measured on the { @ link Schedulers # parallel ( ) parallel }
 * Scheduler ) .
 * < img class = " marble " src = " doc - files / marbles / windowWithTimespan . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure
 * upon cancellation or error triggered by a data signal .
 * @ param windowingTimespan the { @ link Duration } to delimit { @ link Flux } windows
 * @ return a { @ link Flux } of { @ link Flux } windows continuously opened for a given { @ link Duration } */
public final Flux < Flux < T > > window ( Duration windowingTimespan ) { } }

implementation: return window ( windowingTimespan , Schedulers . parallel ( ) ) ;


signature:
public class Flux { /** * Split this { @ link Flux } sequence into continuous , non - overlapping windows that open
 * for a { @ code windowingTimespan } { @ link Duration } ( as measured on the provided { @ link Scheduler } ) .
 * < img class = " marble " src = " doc - files / marbles / windowWithTimespan . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure
 * upon cancellation or error triggered by a data signal .
 * @ param windowingTimespan the { @ link Duration } to delimit { @ link Flux } windows
 * @ param timer a time - capable { @ link Scheduler } instance to run on
 * @ return a { @ link Flux } of { @ link Flux } windows continuously opened for a given { @ link Duration } */
public final Flux < Flux < T > > window ( Duration windowingTimespan , Scheduler timer ) { } }

implementation: return window ( interval ( windowingTimespan , timer ) ) ;


signature:
public class Flux { /** * Split this { @ link Flux } sequence into multiple { @ link Flux } windows that open
 * for a given { @ code windowingTimespan } { @ link Duration } , after which it closes with onComplete .
 * Each window is opened at a regular { @ code timeShift } interval , starting from the
 * first item .
 * Both durations are measured on the provided { @ link Scheduler } .
 * When windowingTimespan < openWindowEvery : dropping windows
 * < img class = " marble " src = " doc - files / marbles / windowWithTimespanLessThanOpenWindowEvery . svg " alt = " " >
 * When windowingTimespan > openWindowEvery : overlapping windows
 * < img class = " marble " src = " doc - files / marbles / windowWithTimespanGreaterThanOpenWindowEvery . svg " alt = " " >
 * When openWindowEvery = = openWindowEvery : exact windows
 * < img class = " marble " src = " doc - files / marbles / windowWithTimespanEqualsOpenWindowEvery . svg " alt = " " >
 * @ reactor . discard The overlapping variant DOES NOT discard elements , as they might be part of another still valid window .
 * The exact window and dropping window variants bot discard elements they internally queued for backpressure
 * upon cancellation or error triggered by a data signal . The dropping window variant also discards elements in between windows .
 * @ param windowingTimespan the maximum { @ link Flux } window { @ link Duration }
 * @ param openWindowEvery the period of time at which to create new { @ link Flux } windows
 * @ param timer a time - capable { @ link Scheduler } instance to run on
 * @ return a { @ link Flux } of { @ link Flux } windows opened at regular intervals and
 * closed after a { @ link Duration } */
public final Flux < Flux < T > > window ( Duration windowingTimespan , Duration openWindowEvery , Scheduler timer ) { } }

implementation: if ( openWindowEvery . equals ( windowingTimespan ) ) { return window ( windowingTimespan ) ; } return windowWhen ( interval ( Duration . ZERO , openWindowEvery , timer ) , aLong -> Mono . delay ( windowingTimespan , timer ) ) ;


signature:
public class Flux { /** * Split this { @ link Flux } sequence into multiple { @ link Flux } windows containing
 * { @ code maxSize } elements ( or less for the final window ) and starting from the first item .
 * Each { @ link Flux } window will onComplete once it contains { @ code maxSize } elements
 * OR it has been open for the given { @ link Duration } ( as measured on the { @ link Schedulers # parallel ( ) parallel }
 * Scheduler ) .
 * < img class = " marble " src = " doc - files / marbles / windowTimeout . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure
 * upon cancellation or error triggered by a data signal .
 * @ param maxSize the maximum number of items to emit in the window before closing it
 * @ param maxTime the maximum { @ link Duration } since the window was opened before closing it
 * @ return a { @ link Flux } of { @ link Flux } windows based on element count and duration */
public final Flux < Flux < T > > windowTimeout ( int maxSize , Duration maxTime ) { } }

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


signature:
public class Flux { /** * Split this { @ link Flux } sequence into multiple { @ link Flux } windows containing
 * { @ code maxSize } elements ( or less for the final window ) and starting from the first item .
 * Each { @ link Flux } window will onComplete once it contains { @ code maxSize } elements
 * OR it has been open for the given { @ link Duration } ( as measured on the provided
 * { @ link Scheduler } ) .
 * < img class = " marble " src = " doc - files / marbles / windowTimeout . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure
 * upon cancellation or error triggered by a data signal .
 * @ param maxSize the maximum number of items to emit in the window before closing it
 * @ param maxTime the maximum { @ link Duration } since the window was opened before closing it
 * @ param timer a time - capable { @ link Scheduler } instance to run on
 * @ return a { @ link Flux } of { @ link Flux } windows based on element count and duration */
public final Flux < Flux < T > > windowTimeout ( int maxSize , Duration maxTime , Scheduler timer ) { } }

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


signature:
public class Flux { /** * Split this { @ link Flux } sequence into multiple { @ link Flux } windows delimited by the
 * given predicate . A new window is opened each time the predicate returns true , at which
 * point the previous window will receive the triggering element then onComplete .
 * Windows are lazily made available downstream at the point where they receive their
 * first event ( an element is pushed , the window errors ) . This variant shouldn ' t
 * expose empty windows , as the separators are emitted into
 * the windows they close .
 * < img class = " marble " src = " doc - files / marbles / windowUntil . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure
 * upon cancellation or error triggered by a data signal .
 * @ param boundaryTrigger a predicate that triggers the next window when it becomes true .
 * @ return a { @ link Flux } of { @ link Flux } windows , bounded depending
 * on the predicate . */
public final Flux < Flux < T > > windowUntil ( Predicate < T > boundaryTrigger ) { } }

implementation: return windowUntil ( boundaryTrigger , false ) ;


signature:
public class Flux { /** * Split this { @ link Flux } sequence into multiple { @ link Flux } windows delimited by the
 * given predicate . A new window is opened each time the predicate returns true .
 * Windows are lazily made available downstream at the point where they receive their
 * first event ( an element is pushed , the window completes or errors ) .
 * If { @ code cutBefore } is true , the old window will onComplete and the triggering
 * element will be emitted in the new window , which becomes immediately available .
 * This variant can emit an empty window if the sequence starts with a separator .
 * Otherwise , the triggering element will be emitted in the old window before it does
 * onComplete , similar to { @ link # windowUntil ( Predicate ) } . This variant shouldn ' t
 * expose empty windows , as the separators are emitted into the windows they close .
 * < img class = " marble " src = " doc - files / marbles / windowUntilWithCutBefore . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure
 * upon cancellation or error triggered by a data signal .
 * @ param boundaryTrigger a predicate that triggers the next window when it becomes true .
 * @ param cutBefore set to true to include the triggering element in the new window rather than the old .
 * @ return a { @ link Flux } of { @ link Flux } windows , bounded depending
 * on the predicate . */
public final Flux < Flux < T > > windowUntil ( Predicate < T > boundaryTrigger , boolean cutBefore ) { } }

implementation: return windowUntil ( boundaryTrigger , cutBefore , Queues . SMALL_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Split this { @ link Flux } sequence into multiple { @ link Flux } windows delimited by the given
 * predicate and using a prefetch . A new window is opened each time the predicate
 * returns true .
 * Windows are lazily made available downstream at the point where they receive their
 * first event ( an element is pushed , the window completes or errors ) .
 * If { @ code cutBefore } is true , the old window will onComplete and the triggering
 * element will be emitted in the new window . This variant can emit an empty window
 * if the sequence starts with a separator .
 * Otherwise , the triggering element will be emitted in the old window before it does
 * onComplete , similar to { @ link # windowUntil ( Predicate ) } . This variant shouldn ' t
 * expose empty windows , as the separators are emitted into the windows they close .
 * < img class = " marble " src = " doc - files / marbles / windowUntilWithCutBefore . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure
 * upon cancellation or error triggered by a data signal .
 * @ param boundaryTrigger a predicate that triggers the next window when it becomes true .
 * @ param cutBefore set to true to include the triggering element in the new window rather than the old .
 * @ param prefetch the request size to use for this { @ link Flux } .
 * @ return a { @ link Flux } of { @ link Flux } windows , bounded depending
 * on the predicate . */
public final Flux < Flux < T > > windowUntil ( Predicate < T > boundaryTrigger , boolean cutBefore , int prefetch ) { } }

implementation: return onAssembly ( new FluxWindowPredicate < > ( this , Queues . unbounded ( prefetch ) , Queues . unbounded ( prefetch ) , prefetch , boundaryTrigger , cutBefore ? FluxBufferPredicate . Mode . UNTIL_CUT_BEFORE : FluxBufferPredicate . Mode . UNTIL ) ) ;


signature:
public class Flux { /** * Split this { @ link Flux } sequence into multiple { @ link Flux } windows that stay open
 * while a given predicate matches the source elements . Once the predicate returns
 * false , the window closes with an onComplete and the triggering element is discarded .
 * Windows are lazily made available downstream at the point where they receive their
 * first event ( an element is pushed , the window completes or errors ) . Empty windows
 * can happen when a sequence starts with a separator or contains multiple separators ,
 * but a sequence that finishes with a separator won ' t cause a remainder empty window
 * to be emitted .
 * < img class = " marble " src = " doc - files / marbles / windowWhile . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure
 * upon cancellation or error triggered by a data signal , as well as the triggering element ( s ) ( that doesn ' t match
 * the predicate ) .
 * @ param inclusionPredicate a predicate that triggers the next window when it becomes false .
 * @ return a { @ link Flux } of { @ link Flux } windows , each containing
 * subsequent elements that all passed a predicate . */
public final Flux < Flux < T > > windowWhile ( Predicate < T > inclusionPredicate ) { } }

implementation: return windowWhile ( inclusionPredicate , Queues . SMALL_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Split this { @ link Flux } sequence into multiple { @ link Flux } windows that stay open
 * while a given predicate matches the source elements . Once the predicate returns
 * false , the window closes with an onComplete and the triggering element is discarded .
 * Windows are lazily made available downstream at the point where they receive their
 * first event ( an element is pushed , the window completes or errors ) . Empty windows
 * can happen when a sequence starts with a separator or contains multiple separators ,
 * but a sequence that finishes with a separator won ' t cause a remainder empty window
 * to be emitted .
 * < img class = " marble " src = " doc - files / marbles / windowWhile . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure
 * upon cancellation or error triggered by a data signal , as well as the triggering element ( s ) ( that doesn ' t match
 * the predicate ) .
 * @ param inclusionPredicate a predicate that triggers the next window when it becomes false .
 * @ param prefetch the request size to use for this { @ link Flux } .
 * @ return a { @ link Flux } of { @ link Flux } windows , each containing
 * subsequent elements that all passed a predicate . */
public final Flux < Flux < T > > windowWhile ( Predicate < T > inclusionPredicate , int prefetch ) { } }

implementation: return onAssembly ( new FluxWindowPredicate < > ( this , Queues . unbounded ( prefetch ) , Queues . unbounded ( prefetch ) , prefetch , inclusionPredicate , FluxBufferPredicate . Mode . WHILE ) ) ;


signature:
public class Flux { /** * Split this { @ link Flux } sequence into potentially overlapping windows controlled by items of a
 * start { @ link Publisher } and end { @ link Publisher } derived from the start values .
 * When Open signal is strictly not overlapping Close signal : dropping windows
 * When Open signal is strictly more frequent than Close signal : overlapping windows
 * When Open signal is exactly coordinated with Close signal : exact windows
 * < img class = " marble " src = " doc - files / marbles / windowWhen . svg " alt = " " >
 * @ reactor . discard This operator DOES NOT discard elements .
 * @ param bucketOpening a { @ link Publisher } that opens a new window when it emits any item
 * @ param closeSelector a { @ link Function } given an opening signal and returning a { @ link Publisher } that
 * will close the window when emitting
 * @ param < U > the type of the sequence opening windows
 * @ param < V > the type of the sequence closing windows opened by the bucketOpening Publisher ' s elements
 * @ return a { @ link Flux } of { @ link Flux } windows opened by signals from a first
 * { @ link Publisher } and lasting until a selected second { @ link Publisher } emits */
public final < U , V > Flux < Flux < T > > windowWhen ( Publisher < U > bucketOpening , final Function < ? super U , ? extends Publisher < V > > closeSelector ) { } }

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


signature:
public class Flux { /** * Zip this { @ link Flux } with another { @ link Publisher } source , that is to say wait
 * for both to emit one element and combine these elements using a { @ code combinator }
 * { @ link BiFunction }
 * 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 / zipWithOtherUsingZipperForFlux . svg " alt = " " >
 * @ param source2 The second source { @ link Publisher } to zip with this { @ link Flux } .
 * @ param combinator The aggregate function that will receive a unique value from each
 * source and return the value to signal downstream
 * @ param < T2 > type of the value from source2
 * @ param < V > The produced output after transformation by the combinator
 * @ return a zipped { @ link Flux } */
public final < T2 , V > Flux < V > zipWith ( Publisher < ? extends T2 > source2 , final BiFunction < ? super T , ? super T2 , ? extends V > combinator ) { } }

implementation: if ( this instanceof FluxZip ) { @ SuppressWarnings ( "unchecked" ) FluxZip < T , V > o = ( FluxZip < T , V > ) this ; Flux < V > result = o . zipAdditionalSource ( source2 , combinator ) ; if ( result != null ) { return result ; } } return zip ( this , source2 , combinator ) ;


signature:
public class Flux { /** * Zip this { @ link Flux } with another { @ link Publisher } source , that is to say wait
 * for both to emit one element and combine these elements using a { @ code combinator }
 * { @ link BiFunction }
 * 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 / zipWithOtherUsingZipperForFlux . svg " alt = " " >
 * @ param source2 The second source { @ link Publisher } to zip with this { @ link Flux } .
 * @ param prefetch the request size to use for this { @ link Flux } and the other { @ link Publisher }
 * @ param combinator The aggregate function that will receive a unique value from each
 * source and return the value to signal downstream
 * @ param < T2 > type of the value from source2
 * @ param < V > The produced output after transformation by the combinator
 * @ return a zipped { @ link Flux } */
@ SuppressWarnings ( "unchecked" ) public final < T2 , V > Flux < V > zipWith ( Publisher < ? extends T2 > source2 , int prefetch , BiFunction < ? super T , ? super T2 , ? extends V > combinator ) { } }

implementation: return zip ( objects -> combinator . apply ( ( T ) objects [ 0 ] , ( T2 ) objects [ 1 ] ) , prefetch , this , source2 ) ;


signature:
public class Flux { /** * Zip this { @ link Flux } with another { @ link Publisher } source , that is to say wait
 * for both 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 / zipWithOtherForFlux . svg " alt = " " >
 * @ param source2 The second source { @ link Publisher } to zip with this { @ link Flux } .
 * @ param prefetch the request size to use for this { @ link Flux } and the other { @ link Publisher }
 * @ param < T2 > type of the value from source2
 * @ return a zipped { @ link Flux } */
public final < T2 > Flux < Tuple2 < T , T2 > > zipWith ( Publisher < ? extends T2 > source2 , int prefetch ) { } }

implementation: return zipWith ( source2 , prefetch , tuple2Function ( ) ) ;


signature:
public class Flux { /** * Zip elements from this { @ link Flux } with the content of an { @ link Iterable } , that is
 * to say combine one element from each , pairwise , into a { @ link Tuple2 } .
 * < img class = " marble " src = " doc - files / marbles / zipWithIterableForFlux . svg " alt = " " >
 * @ param iterable the { @ link Iterable } to zip with
 * @ param < T2 > the value type of the other iterable sequence
 * @ return a zipped { @ link Flux } */
@ SuppressWarnings ( "unchecked" ) public final < T2 > Flux < Tuple2 < T , T2 > > zipWithIterable ( Iterable < ? extends T2 > iterable ) { } }

implementation: return zipWithIterable ( iterable , tuple2Function ( ) ) ;


signature:
public class Flux { /** * Zip elements from this { @ link Flux } with the content of an { @ link Iterable } , that is
 * to say combine one element from each , pairwise , using the given zipper { @ link BiFunction } .
 * < img class = " marble " src = " doc - files / marbles / zipWithIterableUsingZipperForFlux . svg " alt = " " >
 * @ param iterable the { @ link Iterable } to zip with
 * @ param zipper the { @ link BiFunction } pair combinator
 * @ param < T2 > the value type of the other iterable sequence
 * @ param < V > the result type
 * @ return a zipped { @ link Flux } */
public final < T2 , V > Flux < V > zipWithIterable ( Iterable < ? extends T2 > iterable , BiFunction < ? super T , ? super T2 , ? extends V > zipper ) { } }

implementation: return onAssembly ( new FluxZipIterable < > ( this , iterable , zipper ) ) ;


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

implementation: Function < Publisher , Publisher > hook = Hooks . onEachOperatorHook ; if ( hook != null ) { source = ( Flux < T > ) hook . apply ( source ) ; } if ( Hooks . GLOBAL_TRACE ) { AssemblySnapshot stacktrace = new AssemblySnapshot ( null , Traces . callSiteSupplierFactory . get ( ) ) ; source = ( Flux < T > ) Hooks . addAssemblyInfo ( source , stacktrace ) ; } return source ;


signature:
public class Flux { /** * Returns the appropriate Mono instance for a known Supplier Flux .
 * @ param supplier the supplier Flux
 * @ return the mono representing that Flux */
static < T > Mono < T > convertToMono ( Callable < T > supplier ) { } }

implementation: if ( supplier instanceof Fuseable . ScalarCallable ) { Fuseable . ScalarCallable < T > scalarCallable = ( Fuseable . ScalarCallable < T > ) supplier ; T v ; try { v = scalarCallable . call ( ) ; } catch ( Exception e ) { return Mono . error ( e ) ; } if ( v == null ) { return Mono . empty ( ) ; } return Mono . just ( v ) ; } return Mono . onAssembly ( new MonoCallable < > ( supplier ) ) ;


signature:
public class Flux { /** * Unchecked wrap of { @ link Publisher } as { @ link Flux } , supporting { @ link Fuseable } sources
 * @ param source the { @ link Publisher } to wrap
 * @ param < I > input upstream type
 * @ return a wrapped { @ link Flux } */
@ SuppressWarnings ( "unchecked" ) static < I > Flux < I > wrap ( Publisher < ? extends I > source ) { } }

implementation: if ( source instanceof Mono ) { if ( source instanceof Fuseable ) { return new FluxSourceMonoFuseable < > ( ( Mono < I > ) source ) ; } return new FluxSourceMono < > ( ( Mono < I > ) source ) ; } if ( source instanceof Fuseable ) { return new FluxSourceFuseable < > ( source ) ; } return new FluxSource < > ( source ) ;


signature:
public class UnicastProcessor { /** * Create a new { @ link UnicastProcessor } that will buffer on a provided queue in an
 * unbounded fashion .
 * @ param queue the buffering queue
 * @ param endcallback called on any terminal signal
 * @ param < E > the relayed type
 * @ return a unicast { @ link FluxProcessor } */
public static < E > UnicastProcessor < E > create ( Queue < E > queue , Disposable endcallback ) { } }

implementation: return new UnicastProcessor < > ( queue , endcallback ) ;


signature:
public class UnicastProcessor { /** * Create a new { @ link UnicastProcessor } that will buffer on a provided queue in an
 * unbounded fashion .
 * @ param queue the buffering queue
 * @ param endcallback called on any terminal signal
 * @ param onOverflow called when queue . offer return false and unicastProcessor is
 * about to emit onError .
 * @ param < E > the relayed type
 * @ return a unicast { @ link FluxProcessor } */
public static < E > UnicastProcessor < E > create ( Queue < E > queue , Consumer < ? super E > onOverflow , Disposable endcallback ) { } }

implementation: return new UnicastProcessor < > ( queue , onOverflow , endcallback ) ;


signature:
public class Queues { /** * Return the capacity of a given { @ link Queue } in a best effort fashion . Queues that
 * are known to be unbounded will return { @ code Integer . MAX _ VALUE } and queues that
 * have a known bounded capacity will return that capacity . For other { @ link Queue }
 * implementations not recognized by this method or not providing this kind of
 * information , { @ link # CAPACITY _ UNSURE } ( { @ code Integer . MIN _ VALUE } ) is returned .
 * @ param q the { @ link Queue } to try to get a capacity for
 * @ return the capacity of the queue , if discoverable with confidence , or { @ link # CAPACITY _ UNSURE } negative constant . */
public static final int capacity ( Queue q ) { } }

implementation: if ( q instanceof ZeroQueue ) { return 0 ; } if ( q instanceof OneQueue ) { return 1 ; } if ( q instanceof SpscLinkedArrayQueue ) { return Integer . MAX_VALUE ; } else if ( q instanceof SpscArrayQueue ) { return ( ( SpscArrayQueue ) q ) . length ( ) ; } else if ( q instanceof MpscLinkedQueue ) { return Integer . MAX_VALUE ; } else if ( q instanceof BlockingQueue ) { return ( ( BlockingQueue ) q ) . remainingCapacity ( ) ; } else if ( q instanceof ConcurrentLinkedQueue ) { return Integer . MAX_VALUE ; } else { return CAPACITY_UNSURE ; }


signature:
public class Queues { /** * Returns an unbounded , linked - array - based Queue . Integer . max sized link will
 * return the default { @ link # SMALL _ BUFFER _ SIZE } size .
 * @ param linkSize the link size
 * @ param < T > the reified { @ link Queue } generic type
 * @ return an unbounded { @ link Queue } { @ link Supplier } */
@ SuppressWarnings ( "unchecked" ) public static < T > Supplier < Queue < T > > unbounded ( int linkSize ) { } }

implementation: if ( linkSize == XS_BUFFER_SIZE ) { return XS_UNBOUNDED ; } else if ( linkSize == Integer . MAX_VALUE || linkSize == SMALL_BUFFER_SIZE ) { return unbounded ( ) ; } return ( ) -> new SpscLinkedArrayQueue < > ( linkSize ) ;


signature:
public class Tuple3 { /** * Map the 1st part ( T1 ) of this { @ link Tuple3 } into a different value and type ,
 * keeping the other parts .
 * @ param mapper the mapping { @ link Function } for the T1 part
 * @ param < R > the new type for the T1 part
 * @ return a new { @ link Tuple3 } with a different T1 value */
public < R > Tuple3 < R , T2 , T3 > mapT1 ( Function < T1 , R > mapper ) { } }

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


signature:
public class ElasticScheduler { /** * Instantiates the default { @ link ScheduledExecutorService } for the ElasticScheduler
 * ( { @ code Executors . newScheduledThreadPoolExecutor } with core and max pool size of 1 ) . */
@ Override public ScheduledExecutorService get ( ) { } }

implementation: ScheduledThreadPoolExecutor poolExecutor = new ScheduledThreadPoolExecutor ( 1 , factory ) ; poolExecutor . setMaximumPoolSize ( 1 ) ; return poolExecutor ;


signature:
public class ReplayProcessor { /** * Create a { @ link ReplayProcessor } that caches the last element it has pushed ,
 * replaying it to late subscribers . If a { @ link Subscriber } comes in < b > before < / b >
 * any value has been pushed , then the { @ code defaultValue } is emitted instead .
 * This is a buffer - based ReplayProcessor with a history size of 1.
 * < img class = " marble " src = " https : / / raw . githubusercontent . com / reactor / reactor - core / v3.1.3 . RELEASE / src / docs / marble / replaylastd . png "
 * alt = " " >
 * @ param value a default value to start the sequence with in case nothing has been
 * cached yet .
 * @ param < T > the type of the pushed elements
 * @ return a new { @ link ReplayProcessor } that replays its last pushed element to each new
 * { @ link Subscriber } , or a default one if nothing was pushed yet */
public static < T > ReplayProcessor < T > cacheLastOrDefault ( @ Nullable T value ) { } }

implementation: ReplayProcessor < T > b = create ( 1 ) ; if ( value != null ) { b . onNext ( value ) ; } return b ;


signature:
public class ReplayProcessor { /** * Create a new { @ link ReplayProcessor } that either replay all the elements or a
 * limited amount of elements depending on the { @ code unbounded } parameter .
 * @ param historySize maximum items retained if bounded , or initial link size if unbounded
 * @ param unbounded true if " unlimited " data store must be supplied
 * @ param < E > the type of the pushed elements
 * @ return a new { @ link ReplayProcessor } that replays the whole history to each new
 * { @ link Subscriber } if configured as unbounded , a limited history otherwise . */
public static < E > ReplayProcessor < E > create ( int historySize , boolean unbounded ) { } }

implementation: FluxReplay . ReplayBuffer < E > buffer ; if ( unbounded ) { buffer = new FluxReplay . UnboundedReplayBuffer < > ( historySize ) ; } else { buffer = new FluxReplay . SizeBoundReplayBuffer < > ( historySize ) ; } return new ReplayProcessor < > ( buffer ) ;


signature:
public class ReplayProcessor { /** * Creates a time - bounded replay processor .
 * In this setting , the { @ code ReplayProcessor } internally tags each observed item
 * with a timestamp value supplied by the { @ link Schedulers # parallel ( ) } and keeps only
 * those whose age is less than the supplied time value converted to milliseconds . For
 * example , an item arrives at T = 0 and the max age is set to 5 ; at T & gt ; = 5 this first
 * item is then evicted by any subsequent item or termination signal , leaving the
 * buffer empty .
 * Once the processor is terminated , subscribers subscribing to it will receive items
 * that remained in the buffer after the terminal signal , regardless of their age .
 * If an subscriber subscribes while the { @ code ReplayProcessor } is active , it will
 * observe only those items from within the buffer that have an age less than the
 * specified time , and each item observed thereafter , even if the buffer evicts items
 * due to the time constraint in the mean time . In other words , once an subscriber
 * subscribes , it observes items without gaps in the sequence except for any outdated
 * items at the beginning of the sequence .
 * Note that terminal signals ( { @ code onError } and { @ code onComplete } ) trigger
 * eviction as well . For example , with a max age of 5 , the first item is observed at
 * T = 0 , then an { @ code onComplete } signal arrives at T = 10 . If an subscriber subscribes
 * at T = 11 , it will find an empty { @ code ReplayProcessor } with just an { @ code
 * onCompleted } signal .
 * @ param < T > the type of items observed and emitted by the Processor
 * @ param maxAge the maximum age of the contained items
 * @ return a new { @ link ReplayProcessor } that replays elements based on their age . */
public static < T > ReplayProcessor < T > createTimeout ( Duration maxAge ) { } }

implementation: return createTimeout ( maxAge , Schedulers . parallel ( ) ) ;


signature:
public class ReplayProcessor { /** * Creates a time - bounded replay processor .
 * In this setting , the { @ code ReplayProcessor } internally tags each observed item
 * with a timestamp value supplied by the { @ link Scheduler } and keeps only
 * those whose age is less than the supplied time value converted to milliseconds . For
 * example , an item arrives at T = 0 and the max age is set to 5 ; at T & gt ; = 5 this first
 * item is then evicted by any subsequent item or termination signal , leaving the
 * buffer empty .
 * Once the processor is terminated , subscribers subscribing to it will receive items
 * that remained in the buffer after the terminal signal , regardless of their age .
 * If an subscriber subscribes while the { @ code ReplayProcessor } is active , it will
 * observe only those items from within the buffer that have an age less than the
 * specified time , and each item observed thereafter , even if the buffer evicts items
 * due to the time constraint in the mean time . In other words , once an subscriber
 * subscribes , it observes items without gaps in the sequence except for any outdated
 * items at the beginning of the sequence .
 * Note that terminal signals ( { @ code onError } and { @ code onComplete } ) trigger
 * eviction as well . For example , with a max age of 5 , the first item is observed at
 * T = 0 , then an { @ code onComplete } signal arrives at T = 10 . If an subscriber subscribes
 * at T = 11 , it will find an empty { @ code ReplayProcessor } with just an { @ code
 * onCompleted } signal .
 * @ param < T > the type of items observed and emitted by the Processor
 * @ param maxAge the maximum age of the contained items
 * @ return a new { @ link ReplayProcessor } that replays elements based on their age . */
public static < T > ReplayProcessor < T > createTimeout ( Duration maxAge , Scheduler scheduler ) { } }

implementation: return createSizeAndTimeout ( Integer . MAX_VALUE , maxAge , scheduler ) ;


signature:
public class ReplayProcessor { /** * Creates a time - and size - bounded replay processor .
 * In this setting , the { @ code ReplayProcessor } internally tags each received item
 * with a timestamp value supplied by the { @ link Schedulers # parallel ( ) } and holds at
 * most
 * { @ code size } items in its internal buffer . It evicts items from the start of the
 * buffer if their age becomes less - than or equal to the supplied age in milliseconds
 * or the buffer reaches its { @ code size } limit .
 * When subscribers subscribe to a terminated { @ code ReplayProcessor } , they observe
 * the items that remained in the buffer after the terminal signal , regardless of
 * their age , but at most { @ code size } items .
 * If an subscriber subscribes while the { @ code ReplayProcessor } is active , it will
 * observe only those items from within the buffer that have age less than the
 * specified time and each subsequent item , even if the buffer evicts items due to the
 * time constraint in the mean time . In other words , once an subscriber subscribes , it
 * observes items without gaps in the sequence except for the outdated items at the
 * beginning of the sequence .
 * Note that terminal signals ( { @ code onError } and { @ code onComplete } ) trigger
 * eviction as well . For example , with a max age of 5 , the first item is observed at
 * T = 0 , then an { @ code onComplete } signal arrives at T = 10 . If an Subscriber subscribes
 * at T = 11 , it will find an empty { @ code ReplayProcessor } with just an { @ code
 * onCompleted } signal .
 * @ param < T > the type of items observed and emitted by the Processor
 * @ param maxAge the maximum age of the contained items
 * @ param size the maximum number of buffered items
 * @ return a new { @ link ReplayProcessor } that replay up to { @ code size } elements , but
 * will evict them from its history based on their age . */
public static < T > ReplayProcessor < T > createSizeAndTimeout ( int size , Duration maxAge ) { } }

implementation: return createSizeAndTimeout ( size , maxAge , Schedulers . parallel ( ) ) ;


signature:
public class ReplayProcessor { /** * Creates a time - and size - bounded replay processor .
 * In this setting , the { @ code ReplayProcessor } internally tags each received item
 * with a timestamp value supplied by the { @ link Scheduler } and holds at most
 * { @ code size } items in its internal buffer . It evicts items from the start of the
 * buffer if their age becomes less - than or equal to the supplied age in milliseconds
 * or the buffer reaches its { @ code size } limit .
 * When subscribers subscribe to a terminated { @ code ReplayProcessor } , they observe
 * the items that remained in the buffer after the terminal signal , regardless of
 * their age , but at most { @ code size } items .
 * If an subscriber subscribes while the { @ code ReplayProcessor } is active , it will
 * observe only those items from within the buffer that have age less than the
 * specified time and each subsequent item , even if the buffer evicts items due to the
 * time constraint in the mean time . In other words , once an subscriber subscribes , it
 * observes items without gaps in the sequence except for the outdated items at the
 * beginning of the sequence .
 * Note that terminal signals ( { @ code onError } and { @ code onComplete } ) trigger
 * eviction as well . For example , with a max age of 5 , the first item is observed at
 * T = 0 , then an { @ code onComplete } signal arrives at T = 10 . If an Subscriber subscribes
 * at T = 11 , it will find an empty { @ code ReplayProcessor } with just an { @ code
 * onCompleted } signal .
 * @ param < T > the type of items observed and emitted by the Processor
 * @ param maxAge the maximum age of the contained items in milliseconds
 * @ param size the maximum number of buffered items
 * @ param scheduler the { @ link Scheduler } that provides the current time
 * @ return a new { @ link ReplayProcessor } that replay up to { @ code size } elements , but
 * will evict them from its history based on their age . */
public static < T > ReplayProcessor < T > createSizeAndTimeout ( int size , Duration maxAge , Scheduler scheduler ) { } }

implementation: Objects . requireNonNull ( scheduler , "scheduler is null" ) ; if ( size <= 0 ) { throw new IllegalArgumentException ( "size > 0 required but it was " + size ) ; } return new ReplayProcessor < > ( new FluxReplay . SizeAndTimeBoundReplayBuffer < > ( size , maxAge . toMillis ( ) , scheduler ) ) ;


signature:
public class EventLoopProcessor { /** * Create a { @ link Runnable } event loop that will keep monitoring a { @ link
 * LongSupplier } and compare it to a { @ link RingBuffer }
 * @ param upstream the { @ link Subscription } to request / cancel on
 * @ param p parent { @ link EventLoopProcessor }
 * @ param postWaitCallback a { @ link Consumer } notified with the latest sequence read
 * @ param readCount a { @ link LongSupplier } a sequence cursor to wait on
 * @ return a { @ link Runnable } loop to execute to start the requesting loop */
static Runnable createRequestTask ( Subscription upstream , EventLoopProcessor < ? > p , @ Nullable Consumer < Long > postWaitCallback , LongSupplier readCount ) { } }

implementation: return new RequestTask ( upstream , p , postWaitCallback , readCount ) ;


signature:
public class EventLoopProcessor { /** * Spin CPU until the request { @ link LongSupplier } is populated at least once by a
 * strict positive value . To relieve the spin loop , the read sequence itself will be
 * used against so it will wake up only when a signal is emitted upstream or other
 * stopping condition including terminal signals thrown by the { @ link
 * RingBuffer . Reader } waiting barrier .
 * @ param pendingRequest the { @ link LongSupplier } request to observe
 * @ param barrier { @ link RingBuffer . Reader } to wait on
 * @ param isRunning { @ link AtomicBoolean } calling loop running state
 * @ param nextSequence { @ link LongSupplier } ring buffer read cursor
 * @ param waiter an optional extra spin observer for the wait strategy in { @ link
 * RingBuffer . Reader }
 * @ return true if a request has been received , false in any other case . */
static boolean waitRequestOrTerminalEvent ( LongSupplier pendingRequest , RingBuffer . Reader barrier , AtomicBoolean isRunning , LongSupplier nextSequence , Runnable waiter ) { } }

implementation:
try { long waitedSequence ; while ( pendingRequest . getAsLong ( ) <= 0L ) { // pause until first request
waitedSequence = nextSequence . getAsLong ( ) + 1 ; waiter . run ( ) ; barrier . waitFor ( waitedSequence , waiter ) ; if ( ! isRunning . get ( ) ) { WaitStrategy . alert ( ) ; } LockSupport . parkNanos ( 1L ) ; } } catch ( InterruptedException ie ) { Thread . currentThread ( ) . interrupt ( ) ; } catch ( Exception e ) { if ( ! isRunning . get ( ) || WaitStrategy . isAlert ( e ) ) { return false ; } throw e ; } return true ;


signature:
public class EventLoopProcessor { /** * Concurrent addition bound to Long . MAX _ VALUE . Any concurrent write will " happen "
 * before this operation .
 * @ param sequence current sequence to update
 * @ param toAdd delta to add */
static void addCap ( RingBuffer . Sequence sequence , long toAdd ) { } }

implementation: long u , r ; do { r = sequence . getAsLong ( ) ; if ( r == Long . MAX_VALUE ) { return ; } u = Operators . addCap ( r , toAdd ) ; } while ( ! sequence . compareAndSet ( r , u ) ) ;


signature:
public class EventLoopProcessor { /** * Concurrent subtraction bound to 0 and Long . MAX _ VALUE . Any concurrent write will
 * " happen " before this operation .
 * @ param sequence current sequence to update
 * @ param toSub delta to sub
 * @ return value before subscription , 0 or Long . MAX _ VALUE */
static long getAndSub ( RingBuffer . Sequence sequence , long toSub ) { } }

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


signature:
public class EventLoopProcessor { /** * A method to extract a name from the ThreadFactory if it turns out to be a Supplier
 * ( in which case the supplied value string representation is used ) . Otherwise return
 * the current class ' s simpleName .
 * @ param threadFactory the factory to test for a supplied name
 * @ param clazz
 * @ return the name to use in thread pools */
protected static String defaultName ( @ Nullable ThreadFactory threadFactory , Class < ? extends EventLoopProcessor > clazz ) { } }

implementation: String name = threadFactory instanceof Supplier ? ( ( Supplier ) threadFactory ) . get ( ) . toString ( ) : null ; return null != name ? name : clazz . getSimpleName ( ) ;


signature:
public class EventLoopProcessor { /** * A method to create a suitable default { @ link ExecutorService } for use in implementors
 * { @ link # requestTask ( Subscription ) } ( a { @ link Executors # newCachedThreadPool ( ) cached
 * thread pool } ) , reusing a main name and appending { @ code [ request - task ] } suffix .
 * @ param name the main thread name used by the processor .
 * @ return a default { @ link ExecutorService } for requestTask . */
protected static ExecutorService defaultRequestTaskExecutor ( String name ) { } }

implementation: return Executors . newCachedThreadPool ( r -> new Thread ( r , name + "[request-task]" ) ) ;


signature:
public class EventLoopProcessor { /** * Block until all submitted tasks have completed , then do a normal { @ code EventLoopProcessor # dispose ( ) } .
 * @ param timeout the timeout value
 * @ param timeUnit the unit for timeout
 * @ return if the underlying executor terminated and false if the timeout elapsed before termination
 * @ deprecated use { @ link # awaitAndShutdown ( Duration ) } instead */
@ Deprecated public final boolean awaitAndShutdown ( long timeout , TimeUnit timeUnit ) { } }

implementation: try { shutdown ( ) ; return executor . awaitTermination ( timeout , timeUnit ) ; } catch ( InterruptedException ie ) { Thread . currentThread ( ) . interrupt ( ) ; return false ; }


signature:
public class EventLoopProcessor { /** * Block until all submitted tasks have completed , then do a normal { @ code EventLoopProcessor # dispose ( ) } .
 * @ param timeout the timeout value as a { @ link java . time . Duration } . Note this is converted to a { @ link Long }
 * of nanoseconds ( which amounts to roughly 292 years maximum timeout ) .
 * @ return if the underlying executor terminated and false if the timeout elapsed before termination */
public final boolean awaitAndShutdown ( Duration timeout ) { } }

implementation: long nanos = - 1 ; if ( ! timeout . isNegative ( ) ) { nanos = timeout . toNanos ( ) ; } try { shutdown ( ) ; return executor . awaitTermination ( nanos , TimeUnit . NANOSECONDS ) ; } catch ( InterruptedException ie ) { Thread . currentThread ( ) . interrupt ( ) ; return false ; }


signature:
public class EventLoopProcessor { /** * Shutdown this { @ code Processor } , forcibly halting any work currently executing and discarding any tasks that have
 * not yet been executed .
 * @ return a Flux instance with the remaining undelivered values */
final public Flux < IN > forceShutdown ( ) { } }

implementation: int t = terminated ; if ( t != FORCED_SHUTDOWN && TERMINATED . compareAndSet ( this , t , FORCED_SHUTDOWN ) ) { executor . shutdownNow ( ) ; requestTaskExecutor . shutdownNow ( ) ; } return drain ( ) ;


signature:
public class EventLoopProcessor { /** * Shutdown this active { @ code Processor } such that it can no longer be used . If the resource carries any work , it
 * will wait ( but NOT blocking the caller ) for all the remaining tasks to perform before closing the resource . */
public final void shutdown ( ) { } }

implementation: try { onComplete ( ) ; executor . shutdown ( ) ; requestTaskExecutor . shutdown ( ) ; } catch ( Throwable t ) { onError ( Operators . onOperatorError ( t , currentContext ( ) ) ) ; }


signature:
public class MonoProcessor { /** * Block the calling thread for the specified time , waiting for the completion of this { @ code MonoProcessor } . If the
 * { @ link MonoProcessor } is completed with an error a RuntimeException that wraps the error is thrown .
 * @ param timeout the timeout value as a { @ link Duration }
 * @ return the value of this { @ code MonoProcessor } or { @ code null } if the timeout is reached and the { @ code MonoProcessor } has
 * not completed */
@ Override @ Nullable public O block ( Duration timeout ) { } }

implementation: long delay = System . nanoTime ( ) + timeout . toNanos ( ) ; Runnable spinObserver = ( ) -> { if ( delay < System . nanoTime ( ) ) { WaitStrategy . alert ( ) ; } } ; return block ( spinObserver ) ;


signature:
public class MonoProcessor { /** * Returns the internal state as a long fulfilled .
 * < ul >
 * < li > - 1 : cancelled < / li >
 * < li > 0 : ready < / li >
 * < li > 2 : subscribed < / li >
 * < li > 3 : terminated with value < / li >
 * < li > 4 : terminated without value < / li >
 * < li > 5 : terminated with error < / li >
 * < / ul >
 * @ return the internal state from - 1 Cancelled to 5 errored , beyond 3 included is
 * fulfilled .
 * @ deprecated Should not use ( to be removed in 3.2) */
@ Override public long getAsLong ( ) { } }

implementation:
// FIXME Should Be Removed alongside WaitStrategy
NextInner < O > [ ] inners = subscribers ; if ( inners == TERMINATED ) { if ( error != null ) { return 5L ; // terminated with error
} if ( value == null ) { return 4L ; // terminated without value
} return 3L ; // terminated with value
} if ( subscription == Operators . cancelledSubscription ( ) ) { return - 1L ; // cancelled
} if ( subscribers != EMPTY && subscribers != EMPTY_WITH_SOURCE ) { return 2L ; // subscribed
} return 0L ;


signature:
public class MonoProcessor { /** * Returns the value that completed this { @ link MonoProcessor } . Returns { @ code null } if the { @ link MonoProcessor } has not been completed . If the
 * { @ link MonoProcessor } is completed with an error a RuntimeException that wraps the error is thrown .
 * @ return the value that completed the { @ link MonoProcessor } , or { @ code null } if it has not been completed
 * @ throws RuntimeException if the { @ link MonoProcessor } was completed with an error */
@ Nullable public O peek ( ) { } }

implementation: if ( ! isTerminated ( ) ) { return null ; } if ( value != null ) { return value ; } if ( error != null ) { RuntimeException re = Exceptions . propagate ( error ) ; re = Exceptions . addSuppressed ( re , new Exception ( "Mono#peek terminated with an error" ) ) ; throw re ; } return null ;


signature:
public class FluxMerge { /** * Returns a new instance which has the additional source to be merged together with
 * the current array of sources .
 * This operation doesn ' t change the current FluxMerge instance .
 * @ param source the new source to merge with the others
 * @ param newQueueSupplier a function that should return a new queue supplier based on the change in the maxConcurrency value
 * @ return the new FluxMerge instance */
FluxMerge < T > mergeAdditionalSource ( Publisher < ? extends T > source , IntFunction < Supplier < ? extends Queue < T > > > newQueueSupplier ) { } }

implementation:
int n = sources . length ; @ SuppressWarnings ( "unchecked" ) Publisher < ? extends T > [ ] newArray = new Publisher [ n + 1 ] ; System . arraycopy ( sources , 0 , newArray , 0 , n ) ; newArray [ n ] = source ; // increase the maxConcurrency because if merged separately , it would have run concurrently anyway
Supplier < ? extends Queue < T > > newMainQueue ; int mc = maxConcurrency ; if ( mc != Integer . MAX_VALUE ) { mc ++ ; newMainQueue = newQueueSupplier . apply ( mc ) ; } else { newMainQueue = mainQueueSupplier ; } return new FluxMerge < > ( newArray , delayError , mc , newMainQueue , prefetch , innerQueueSupplier ) ;