signature:
public class Schedulers { /** * { @ link Scheduler } that hosts a fixed pool of single - threaded ExecutorService - based
 * workers and is suited for parallel work . This type of { @ link Scheduler } detects and
 * rejects usage of blocking Reactor APIs .
 * @ param name Thread prefix
 * @ param parallelism Number of pooled workers .
 * @ param daemon false if the { @ link Scheduler } requires an explicit { @ link
 * Scheduler # dispose ( ) } to exit the VM .
 * @ return a new { @ link Scheduler } that hosts a fixed pool of single - threaded
 * ExecutorService - based workers and is suited for parallel work */
public static Scheduler newParallel ( String name , int parallelism , boolean daemon ) { } }

implementation: return newParallel ( parallelism , new ReactorThreadFactory ( name , ParallelScheduler . COUNTER , daemon , true , Schedulers :: defaultUncaughtException ) ) ;


signature:
public class Schedulers { /** * { @ link Scheduler } that hosts a single - threaded ExecutorService - based worker and is
 * suited for parallel work . This type of { @ link Scheduler } detects and rejects usage
 * of blocking Reactor APIs .
 * @ param name Component and thread name prefix
 * @ param daemon false if the { @ link Scheduler } requires an explicit { @ link
 * Scheduler # dispose ( ) } to exit the VM .
 * @ return a new { @ link Scheduler } that hosts a single - threaded ExecutorService - based
 * worker */
public static Scheduler newSingle ( String name , boolean daemon ) { } }

implementation: return newSingle ( new ReactorThreadFactory ( name , SingleScheduler . COUNTER , daemon , true , Schedulers :: defaultUncaughtException ) ) ;


signature:
public class Schedulers { /** * Define a hook that is executed when a { @ link Scheduler } has
 * { @ link # handleError ( Throwable ) handled an error } . Note that it is executed after
 * the error has been passed to the thread uncaughtErrorHandler , which is not the
 * case when a fatal error occurs ( see { @ link Exceptions # throwIfJvmFatal ( Throwable ) } ) .
 * @ param c the new hook to set . */
public static void onHandleError ( BiConsumer < Thread , ? super Throwable > c ) { } }

implementation: if ( log . isDebugEnabled ( ) ) { log . debug ( "Hooking new default: onHandleError" ) ; } onHandleErrorHook = Objects . requireNonNull ( c , "onHandleError" ) ;


signature:
public class Schedulers { /** * Set up an additional { @ link ScheduledExecutorService } decorator for a given key
 * only if that key is not already present . Note that the { @ link Factory } ' s legacy
 * { @ link Factory # decorateExecutorService ( String , Supplier ) } method will always be
 * invoked last after applying the decorators added via this method .
 * The decorator is a { @ link BiFunction } taking the Scheduler and the backing
 * { @ link ScheduledExecutorService } as second argument . It returns the
 * decorated { @ link ScheduledExecutorService } .
 * @ param key the key under which to set up the decorator
 * @ param decorator the executor service decorator to add , if key not already present .
 * @ return true if the decorator was added , false if a decorator was already present
 * for this key .
 * @ see # setExecutorServiceDecorator ( String , BiFunction )
 * @ see # removeExecutorServiceDecorator ( String )
 * @ see Schedulers # onScheduleHook ( String , Function ) */
public static boolean addExecutorServiceDecorator ( String key , BiFunction < Scheduler , ScheduledExecutorService , ScheduledExecutorService > decorator ) { } }

implementation: synchronized ( DECORATORS ) { return DECORATORS . putIfAbsent ( key , decorator ) == null ; }


signature:
public class Schedulers { /** * Set up an additional { @ link ScheduledExecutorService } decorator for a given key ,
 * even if that key is already present . Note that the { @ link Factory } ' s legacy
 * { @ link Factory # decorateExecutorService ( String , Supplier ) } method will always be
 * invoked last after applying the decorators added via this method .
 * The decorator is a { @ link BiFunction } taking the Scheduler and the backing
 * { @ link ScheduledExecutorService } as second argument . It returns the
 * decorated { @ link ScheduledExecutorService } .
 * @ param key the key under which to set up the decorator
 * @ param decorator the executor service decorator to add , if key not already present .
 * @ see # addExecutorServiceDecorator ( String , BiFunction )
 * @ see # removeExecutorServiceDecorator ( String )
 * @ see Schedulers # onScheduleHook ( String , Function ) */
public static void setExecutorServiceDecorator ( String key , BiFunction < Scheduler , ScheduledExecutorService , ScheduledExecutorService > decorator ) { } }

implementation: synchronized ( DECORATORS ) { DECORATORS . put ( key , decorator ) ; }


signature:
public class Schedulers { /** * Remove an existing { @ link ScheduledExecutorService } decorator if it has been set up
 * via { @ link # addExecutorServiceDecorator ( String , BiFunction ) } . Note that the { @ link Factory } ' s
 * legacy { @ link Factory # decorateExecutorService ( String , Supplier ) } method is always
 * applied last , even if all other decorators have been removed via this method .
 * In case the decorator implements { @ link Disposable } , it is also
 * { @ link Disposable # dispose ( ) disposed } .
 * @ param key the key for the executor service decorator to remove
 * @ return the removed decorator , or null if none was set for that key
 * @ see # addExecutorServiceDecorator ( String , BiFunction )
 * @ see # setExecutorServiceDecorator ( String , BiFunction ) */
public static BiFunction < Scheduler , ScheduledExecutorService , ScheduledExecutorService > removeExecutorServiceDecorator ( String key ) { } }

implementation: BiFunction < Scheduler , ScheduledExecutorService , ScheduledExecutorService > removed ; synchronized ( DECORATORS ) { removed = DECORATORS . remove ( key ) ; } if ( removed instanceof Disposable ) { ( ( Disposable ) removed ) . dispose ( ) ; } return removed ;


signature:
public class Schedulers { /** * This method is aimed at { @ link Scheduler } implementors , enabling custom implementations
 * that are backed by a { @ link ScheduledExecutorService } to also have said executors
 * decorated ( ie . for instrumentation purposes ) .
 * It < strong > applies < / strong > the decorators added via
 * { @ link # addExecutorServiceDecorator ( String , BiFunction ) } , so it shouldn ' t be added
 * as a decorator . Note also that decorators are not guaranteed to be idempotent , so
 * this method should be called only once per executor .
 * @ param owner a { @ link Scheduler } that owns the { @ link ScheduledExecutorService }
 * @ param original the { @ link ScheduledExecutorService } that the { @ link Scheduler }
 * wants to use originally
 * @ return the decorated { @ link ScheduledExecutorService } , or the original if no decorator is set up
 * @ see # addExecutorServiceDecorator ( String , BiFunction )
 * @ see # removeExecutorServiceDecorator ( String ) */
public static ScheduledExecutorService decorateExecutorService ( Scheduler owner , ScheduledExecutorService original ) { } }

implementation:
synchronized ( DECORATORS ) { for ( BiFunction < Scheduler , ScheduledExecutorService , ScheduledExecutorService > decorator : DECORATORS . values ( ) ) { original = decorator . apply ( owner , original ) ; } } final ScheduledExecutorService beforeFactory = original ; // Backward compatibility
final String schedulerType ; if ( owner instanceof SingleScheduler ) { schedulerType = Schedulers . SINGLE ; } else if ( owner instanceof ParallelScheduler ) { schedulerType = Schedulers . PARALLEL ; } else if ( owner instanceof ElasticScheduler ) { schedulerType = Schedulers . ELASTIC ; } else if ( owner instanceof DelegateServiceScheduler ) { schedulerType = "ExecutorService" ; } else { schedulerType = owner . getClass ( ) . getName ( ) ; } return factory . decorateExecutorService ( schedulerType , ( ) -> beforeFactory ) ;


signature:
public class Schedulers { /** * Add or replace a named scheduling { @ link Function decorator } . With subsequent calls
 * to this method , the onScheduleHook hook can be a composite of several sub - hooks , each
 * with a different key .
 * The sub - hook is a { @ link Function } taking the scheduled { @ link Runnable } .
 * It returns the decorated { @ link Runnable } .
 * @ param key the key under which to set up the onScheduleHook sub - hook
 * @ param decorator the { @ link Runnable } decorator to add ( or replace , if key is already present )
 * @ see # resetOnScheduleHook ( String )
 * @ see # resetOnScheduleHooks ( ) */
public static void onScheduleHook ( String key , Function < Runnable , Runnable > decorator ) { } }

implementation: synchronized ( onScheduleHooks ) { onScheduleHooks . put ( key , decorator ) ; Function < Runnable , Runnable > newHook = null ; for ( Function < Runnable , Runnable > function : onScheduleHooks . values ( ) ) { if ( newHook == null ) { newHook = function ; } else { newHook = newHook . andThen ( function ) ; } } onScheduleHook = newHook ; }


signature:
public class Schedulers { /** * Reset a specific onScheduleHook { @ link Function sub - hook } if it has been set up
 * via { @ link # onScheduleHook ( String , Function ) } .
 * @ param key the key for onScheduleHook sub - hook to remove
 * @ see # onScheduleHook ( String , Function )
 * @ see # resetOnScheduleHooks ( ) */
public static void resetOnScheduleHook ( String key ) { } }

implementation: synchronized ( onScheduleHooks ) { onScheduleHooks . remove ( key ) ; if ( onScheduleHooks . isEmpty ( ) ) { onScheduleHook = Function . identity ( ) ; } else { Function < Runnable , Runnable > newHook = null ; for ( Function < Runnable , Runnable > function : onScheduleHooks . values ( ) ) { if ( newHook == null ) { newHook = function ; } else { newHook = newHook . andThen ( function ) ; } } onScheduleHook = newHook ; } }


signature:
public class Schedulers { /** * Applies the hooks registered with { @ link Schedulers # onScheduleHook ( String , Function ) } .
 * @ param runnable a { @ link Runnable } submitted to a { @ link Scheduler }
 * @ return decorated { @ link Runnable } if any hook is registered , the original otherwise . */
public static Runnable onSchedule ( Runnable runnable ) { } }

implementation: Function < Runnable , Runnable > hook = onScheduleHook ; if ( hook != null ) { return hook . apply ( runnable ) ; } else { return runnable ; }


signature:
public class Schedulers { /** * Clear any cached { @ link Scheduler } and call dispose on them . */
public static void shutdownNow ( ) { } }

implementation: CachedScheduler oldElastic = CACHED_ELASTIC . getAndSet ( null ) ; CachedScheduler oldParallel = CACHED_PARALLEL . getAndSet ( null ) ; CachedScheduler oldSingle = CACHED_SINGLE . getAndSet ( null ) ; if ( oldElastic != null ) oldElastic . _dispose ( ) ; if ( oldParallel != null ) oldParallel . _dispose ( ) ; if ( oldSingle != null ) oldSingle . _dispose ( ) ;


signature:
public class Schedulers { /** * Get a { @ link CachedScheduler } out of the { @ code reference } or create one using the
 * { @ link Supplier } if the reference is empty , effectively creating a single instance
 * to be reused as a default scheduler for the given { @ code key } category .
 * @ param reference the cache reference that holds the scheduler
 * @ param key the " name " for the Scheduler ' s category / type
 * @ param supplier the { @ link Scheduler } generator to use and wrap into a { @ link CachedScheduler } .
 * Note that in case of a race , an extraneous Scheduler can be created , but it ' ll get
 * immediately { @ link Scheduler # dispose ( ) disposed } .
 * @ return a { @ link CachedScheduler } to be reused , either pre - existing or created */
static CachedScheduler cache ( AtomicReference < CachedScheduler > reference , String key , Supplier < Scheduler > supplier ) { } }

implementation:
CachedScheduler s = reference . get ( ) ; if ( s != null ) { return s ; } s = new CachedScheduler ( key , supplier . get ( ) ) ; if ( reference . compareAndSet ( null , s ) ) { return s ; } // the reference was updated in the meantime with a cached scheduler
// fallback to it and dispose the extraneous one
s . _dispose ( ) ; return reference . get ( ) ;


signature:
public class Schedulers { /** * Scan an { @ link Executor } or { @ link ExecutorService } , recognizing several special
 * implementations . Unwraps some decorating schedulers , recognizes { @ link Scannable }
 * schedulers and delegates to their { @ link Scannable # scanUnsafe ( Scannable . Attr ) }
 * method , introspects { @ link ThreadPoolExecutor } instances .
 * If no data can be extracted , defaults to the provided { @ code orElse }
 * { @ link Scannable # scanUnsafe ( Scannable . Attr ) scanUnsafe } .
 * @ param executor the executor to introspect in a best effort manner .
 * @ param key the key to scan for . CAPACITY and BUFFERED mainly .
 * @ return an equivalent of { @ link Scannable # scanUnsafe ( Scannable . Attr ) } but that can
 * also work on some implementations of { @ link Executor } */
@ Nullable static final Object scanExecutor ( Executor executor , Scannable . Attr key ) { } }

implementation: if ( executor instanceof DelegateServiceScheduler . UnsupportedScheduledExecutorService ) { executor = ( ( DelegateServiceScheduler . UnsupportedScheduledExecutorService ) executor ) . get ( ) ; } if ( executor instanceof Scannable ) { return ( ( Scannable ) executor ) . scanUnsafe ( key ) ; } if ( executor instanceof ExecutorService ) { ExecutorService service = ( ExecutorService ) executor ; if ( key == Scannable . Attr . TERMINATED ) return service . isTerminated ( ) ; if ( key == Scannable . Attr . CANCELLED ) return service . isShutdown ( ) ; } if ( executor instanceof ThreadPoolExecutor ) { final ThreadPoolExecutor poolExecutor = ( ThreadPoolExecutor ) executor ; if ( key == Scannable . Attr . CAPACITY ) return poolExecutor . getMaximumPoolSize ( ) ; if ( key == Scannable . Attr . BUFFERED ) return ( ( Long ) ( poolExecutor . getTaskCount ( ) - poolExecutor . getCompletedTaskCount ( ) ) ) . intValue ( ) ; if ( key == Scannable . Attr . LARGE_BUFFERED ) return poolExecutor . getTaskCount ( ) - poolExecutor . getCompletedTaskCount ( ) ; } return null ;


signature:
public class Tuple2 { /** * Map the left - hand part ( T1 ) of this { @ link Tuple2 } into a different value and type ,
 * keeping the right - hand part ( T2 ) .
 * @ param mapper the mapping { @ link Function } for the left - hand part
 * @ param < R > the new type for the left - hand part
 * @ return a new { @ link Tuple2 } with a different left ( T1 ) value */
public < R > Tuple2 < R , T2 > mapT1 ( Function < T1 , R > mapper ) { } }

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


signature:
public class FluxFlatMap { /** * Checks if the source is a Supplier and if the mapper ' s publisher output is also
 * a supplier , thus avoiding subscribing to any of them .
 * @ param source the source publisher
 * @ param s the end consumer
 * @ param mapper the mapper function
 * @ param fuseableExpected if true , the parent class was marked Fuseable thus the
 * mapping output has to signal onSubscribe with a QueueSubscription
 * @ return true if the optimization worked */
@ SuppressWarnings ( "unchecked" ) static < T , R > boolean trySubscribeScalarMap ( Publisher < ? extends T > source , CoreSubscriber < ? super R > s , Function < ? super T , ? extends Publisher < ? extends R > > mapper , boolean fuseableExpected ) { } }

implementation: if ( source instanceof Callable ) { T t ; try { t = ( ( Callable < ? extends T > ) source ) . call ( ) ; } catch ( Throwable e ) { Operators . error ( s , Operators . onOperatorError ( e , s . currentContext ( ) ) ) ; return true ; } if ( t == null ) { Operators . complete ( s ) ; return true ; } Publisher < ? extends R > p ; try { p = Objects . requireNonNull ( mapper . apply ( t ) , "The mapper returned a null Publisher" ) ; } catch ( Throwable e ) { Operators . error ( s , Operators . onOperatorError ( null , e , t , s . currentContext ( ) ) ) ; return true ; } if ( p instanceof Callable ) { R v ; try { v = ( ( Callable < R > ) p ) . call ( ) ; } catch ( Throwable e ) { Operators . error ( s , Operators . onOperatorError ( null , e , t , s . currentContext ( ) ) ) ; return true ; } if ( v != null ) { s . onSubscribe ( Operators . scalarSubscription ( s , v ) ) ; } else { Operators . complete ( s ) ; } } else { if ( ! fuseableExpected || p instanceof Fuseable ) { p . subscribe ( s ) ; } else { p . subscribe ( new FluxHide . SuppressFuseableSubscriber < > ( s ) ) ; } } return true ; } return false ;


signature:
public class RingBuffer { /** * Create a new multiple producer RingBuffer with the specified wait strategy .
 * < p > See { @ code MultiProducerRingBuffer } .
 * @ param < E > the element type
 * @ param factory used to create the events within the ring buffer .
 * @ param bufferSize number of elements to create within the ring buffer .
 * @ param waitStrategy used to determine how to wait for new elements to become available .
 * @ param spinObserver the Runnable to call on a spin loop wait
 * @ return the new RingBuffer instance */
static < E > RingBuffer < E > createMultiProducer ( Supplier < E > factory , int bufferSize , WaitStrategy waitStrategy , Runnable spinObserver ) { } }

implementation: if ( hasUnsafe ( ) ) { MultiProducerRingBuffer sequencer = new MultiProducerRingBuffer ( bufferSize , waitStrategy , spinObserver ) ; return new UnsafeRingBuffer < > ( factory , sequencer ) ; } else { throw new IllegalStateException ( "This JVM does not support sun.misc.Unsafe" ) ; }


signature:
public class RingBuffer { /** * Create a new single producer RingBuffer with the specified wait strategy .
 * < p > See { @ code MultiProducerRingBuffer } .
 * @ param < E > the element type
 * @ param factory used to create the events within the ring buffer .
 * @ param bufferSize number of elements to create within the ring buffer .
 * @ param waitStrategy used to determine how to wait for new elements to become available .
 * @ return the new RingBuffer instance */
static < E > RingBuffer < E > createSingleProducer ( Supplier < E > factory , int bufferSize , WaitStrategy waitStrategy ) { } }

implementation: return createSingleProducer ( factory , bufferSize , waitStrategy , null ) ;


signature:
public class RingBuffer { /** * Create a new single producer RingBuffer with the specified wait strategy .
 * < p > See { @ code MultiProducerRingBuffer } .
 * @ param < E > the element type
 * @ param factory used to create the events within the ring buffer .
 * @ param bufferSize number of elements to create within the ring buffer .
 * @ param waitStrategy used to determine how to wait for new elements to become available .
 * @ param spinObserver called each time the next claim is spinning and waiting for a slot
 * @ return the new RingBuffer instance */
static < E > RingBuffer < E > createSingleProducer ( Supplier < E > factory , int bufferSize , WaitStrategy waitStrategy , @ Nullable Runnable spinObserver ) { } }

implementation: SingleProducerSequencer sequencer = new SingleProducerSequencer ( bufferSize , waitStrategy , spinObserver ) ; if ( hasUnsafe ( ) && Queues . isPowerOfTwo ( bufferSize ) ) { return new UnsafeRingBuffer < > ( factory , sequencer ) ; } else { return new NotFunRingBuffer < > ( factory , sequencer ) ; }


signature:
public class RingBuffer { /** * Get the minimum sequence from an array of { @ link Sequence } s .
 * @ param excludeSequence to exclude from search .
 * @ param sequences to compare .
 * @ param minimum an initial default minimum . If the array is empty this value will be returned .
 * @ return the minimum sequence found or Long . MAX _ VALUE if the array is empty . */
static long getMinimumSequence ( @ Nullable Sequence excludeSequence , final Sequence [ ] sequences , long minimum ) { } }

implementation: for ( int i = 0 , n = sequences . length ; i < n ; i ++ ) { if ( excludeSequence == null || sequences [ i ] != excludeSequence ) { long value = sequences [ i ] . getAsLong ( ) ; minimum = Math . min ( minimum , value ) ; } } return minimum ;


signature:
public class RingBufferProducer { /** * Get the minimum sequence value from all of the gating sequences
 * added to this ringBuffer .
 * @ param excludeSequence to exclude from search
 * @ return The minimum gating sequence or the cursor sequence if
 * no sequences have been added . */
long getMinimumSequence ( @ Nullable RingBuffer . Sequence excludeSequence ) { } }

implementation: return RingBuffer . getMinimumSequence ( excludeSequence , gatingSequences , cursor . getAsLong ( ) ) ;


signature:
public class SingleProducerSequencer { /** * See { @ code RingBufferProducer . next ( int ) } . */
@ Override long next ( int n ) { } }

implementation:
long nextValue = this . nextValue ; long nextSequence = nextValue + n ; long wrapPoint = nextSequence - bufferSize ; long cachedGatingSequence = this . cachedValue ; if ( wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue ) { long minSequence ; while ( wrapPoint > ( minSequence = RingBuffer . getMinimumSequence ( gatingSequences , nextValue ) ) ) { if ( spinObserver != null ) { spinObserver . run ( ) ; } LockSupport . parkNanos ( 1L ) ; // TODO : Use waitStrategy to spin ?
} this . cachedValue = minSequence ; } this . nextValue = nextSequence ; return nextSequence ;


signature:
public class SingleProducerSequencer { /** * See { @ code RingBufferProducer . producerCapacity ( ) } . */
@ Override public long getPending ( ) { } }

implementation: long nextValue = this . nextValue ; long consumed = RingBuffer . getMinimumSequence ( gatingSequences , nextValue ) ; long produced = nextValue ; return produced - consumed ;


signature:
public class MultiProducerRingBuffer { /** * See { @ code RingBufferProducer . next ( int ) } . */
@ Override long next ( int n ) { } }

implementation:
long current ; long next ; do { current = cursor . getAsLong ( ) ; next = current + n ; long wrapPoint = next - bufferSize ; long cachedGatingSequence = gatingSequenceCache . getAsLong ( ) ; if ( wrapPoint > cachedGatingSequence || cachedGatingSequence > current ) { long gatingSequence = RingBuffer . getMinimumSequence ( gatingSequences , current ) ; if ( wrapPoint > gatingSequence ) { if ( spinObserver != null ) { spinObserver . run ( ) ; } LockSupport . parkNanos ( 1 ) ; // TODO , should we spin based on the wait strategy ?
continue ; } gatingSequenceCache . set ( gatingSequence ) ; } else if ( cursor . compareAndSet ( current , next ) ) { break ; } } while ( true ) ; return next ;


signature:
public class MultiProducerRingBuffer { /** * See { @ code RingBufferProducer . producerCapacity ( ) } . */
@ Override long getPending ( ) { } }

implementation: long consumed = RingBuffer . getMinimumSequence ( gatingSequences , cursor . getAsLong ( ) ) ; long produced = cursor . getAsLong ( ) ; return produced - consumed ;


signature:
public class MultiProducerRingBuffer { /** * See { @ code RingBufferProducer . isAvailable ( long ) } */
boolean isAvailable ( long sequence ) { } }

implementation: int index = calculateIndex ( sequence ) ; int flag = calculateAvailabilityFlag ( sequence ) ; long bufferAddress = ( index * SCALE ) + BASE ; return UNSAFE . getIntVolatile ( availableBuffer , bufferAddress ) == flag ;


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

implementation: ScheduledThreadPoolExecutor e = ( ScheduledThreadPoolExecutor ) Executors . newScheduledThreadPool ( 1 , this . factory ) ; e . setRemoveOnCancelPolicy ( true ) ; e . setMaximumPoolSize ( 1 ) ; return e ;


signature:
public class Tuple8 { /** * Map the 1st part ( T1 ) of this { @ link Tuple8 } 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 Tuple8 } with a different T1 value */
public < R > Tuple8 < R , T2 , T3 , T4 , T5 , T6 , T7 , T8 > mapT1 ( Function < T1 , R > mapper ) { } }

implementation: return new Tuple8 < > ( mapper . apply ( t1 ) , t2 , t3 , t4 , t5 , t6 , t7 , t8 ) ;


signature:
public class ParallelFlux { /** * Take a Publisher and prepare to consume it on multiple ' rails ' ( one per CPU core )
 * in a round - robin fashion .
 * @ param < T > the value type
 * @ param source the source Publisher
 * @ return the { @ link ParallelFlux } instance */
public static < T > ParallelFlux < T > from ( Publisher < ? extends T > source ) { } }

implementation: return from ( source , Runtime . getRuntime ( ) . availableProcessors ( ) , Queues . SMALL_BUFFER_SIZE , Queues . small ( ) ) ;


signature:
public class ParallelFlux { /** * Take a Publisher and prepare to consume it on { @ code parallelism } number of ' rails ' ,
 * possibly ordered and in a round - robin fashion .
 * @ param < T > the value type
 * @ param source the source Publisher
 * @ param parallelism the number of parallel rails
 * @ return the new { @ link ParallelFlux } instance */
public static < T > ParallelFlux < T > from ( Publisher < ? extends T > source , int parallelism ) { } }

implementation: return from ( source , parallelism , Queues . SMALL_BUFFER_SIZE , Queues . small ( ) ) ;


signature:
public class ParallelFlux { /** * Take a Publisher and prepare to consume it on { @ code parallelism } number of ' rails '
 * and in a round - robin fashion and use custom prefetch amount and queue
 * for dealing with the source Publisher ' s values .
 * @ param < T > the value type
 * @ param source the source Publisher
 * @ param parallelism the number of parallel rails
 * @ param prefetch the number of values to prefetch from the source
 * @ param queueSupplier the queue structure supplier to hold the prefetched values
 * from the source until there is a rail ready to process it .
 * @ return the new { @ link ParallelFlux } instance */
public static < T > ParallelFlux < T > from ( Publisher < ? extends T > source , int parallelism , int prefetch , Supplier < Queue < T > > queueSupplier ) { } }

implementation: Objects . requireNonNull ( queueSupplier , "queueSupplier" ) ; Objects . requireNonNull ( source , "source" ) ; return onAssembly ( new ParallelSource < > ( source , parallelism , prefetch , queueSupplier ) ) ;


signature:
public class ParallelFlux { /** * Wraps multiple Publishers into a { @ link ParallelFlux } which runs them in parallel and
 * unordered .
 * @ param < T > the value type
 * @ param publishers the array of publishers
 * @ return the new { @ link ParallelFlux } instance */
@ SafeVarargs public static < T > ParallelFlux < T > from ( Publisher < T > ... publishers ) { } }

implementation: return onAssembly ( new ParallelArraySource < > ( publishers ) ) ;


signature:
public class ParallelFlux { /** * Perform a fluent transformation to a value via a converter function which receives
 * this ParallelFlux .
 * @ param < U > the output value type
 * @ param converter the converter function from { @ link ParallelFlux } to some type
 * @ return the value returned by the converter function */
public final < U > U as ( Function < ? super ParallelFlux < T > , U > converter ) { } }

implementation: return converter . apply ( this ) ;


signature:
public class ParallelFlux { /** * Activate assembly tracing for this particular { @ link ParallelFlux } , in case of an
 * error upstream of the checkpoint . Tracing incurs the cost of an exception stack trace
 * creation .
 * It should be placed towards the end of the reactive chain , as errors
 * triggered downstream of it cannot be observed and augmented with assembly trace .
 * @ return the assembly tracing { @ link ParallelFlux } */
public final ParallelFlux < T > checkpoint ( ) { } }

implementation: AssemblySnapshot stacktrace = new AssemblySnapshot ( null , Traces . callSiteSupplierFactory . get ( ) ) ; return new ParallelFluxOnAssembly < > ( this , stacktrace ) ;


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

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


signature:
public class ParallelFlux { /** * Collect the elements in each rail into a collection supplied via a
 * collectionSupplier and collected into with a collector action , emitting the
 * collection at the end .
 * @ param < C > the collection type
 * @ param collectionSupplier the supplier of the collection in each rail
 * @ param collector the collector , taking the per - rail collection and the current
 * item
 * @ return the new { @ link ParallelFlux } instance */
public final < C > ParallelFlux < C > collect ( Supplier < ? extends C > collectionSupplier , BiConsumer < ? super C , ? super T > collector ) { } }

implementation: return onAssembly ( new ParallelCollect < > ( this , collectionSupplier , collector ) ) ;


signature:
public class ParallelFlux { /** * Sorts the ' rails ' according to the comparator and returns a full sorted list as a
 * Publisher .
 * This operator requires a finite source ParallelFlux .
 * @ param comparator the comparator to compare elements
 * @ param capacityHint the expected number of total elements
 * @ return the new Mono instance */
public final Mono < List < T > > collectSortedList ( Comparator < ? super T > comparator , int capacityHint ) { } }

implementation: int ch = capacityHint / parallelism ( ) + 1 ; ParallelFlux < List < T > > railReduced = reduce ( ( ) -> new ArrayList < > ( ch ) , ( a , b ) -> { a . add ( b ) ; return a ; } ) ; ParallelFlux < List < T > > railSorted = railReduced . map ( list -> { list . sort ( comparator ) ; return list ; } ) ; Mono < List < T > > merged = railSorted . reduce ( ( a , b ) -> sortedMerger ( a , b , comparator ) ) ; return merged ;


signature:
public class ParallelFlux { /** * Allows composing operators off the ' rails ' , as individual { @ link GroupedFlux } instances keyed by
 * the zero based rail ' s index . The transformed groups are { @ link Flux # parallel parallelized } back
 * once the transformation has been applied .
 * Note that like in { @ link # groups ( ) } , requests and cancellation compose through , and
 * cancelling only one rail may result in undefined behavior .
 * @ param composer the composition function to apply on each { @ link GroupedFlux rail }
 * @ param < U > the type of the resulting parallelized flux
 * @ return a { @ link ParallelFlux } of the composed groups */
public final < U > ParallelFlux < U > composeGroup ( Function < ? super GroupedFlux < Integer , T > , ? extends Publisher < ? extends U > > composer ) { } }

implementation: if ( getPrefetch ( ) > - 1 ) { return from ( groups ( ) . flatMap ( composer :: apply ) , parallelism ( ) , getPrefetch ( ) , Queues . small ( ) ) ; } else { return from ( groups ( ) . flatMap ( composer :: apply ) , parallelism ( ) ) ; }


signature:
public class ParallelFlux { /** * Generates and concatenates Publishers on each ' rail ' , signalling errors immediately
 * and generating 2 publishers upfront .
 * @ param < R > the result type
 * @ param mapper the function to map each rail ' s value into a Publisher source and the
 * inner Publishers ( immediate , boundary , end )
 * @ return the new { @ link ParallelFlux } instance */
public final < R > ParallelFlux < R > concatMap ( Function < ? super T , ? extends Publisher < ? extends R > > mapper ) { } }

implementation: return concatMap ( mapper , 2 , ErrorMode . IMMEDIATE ) ;


signature:
public class ParallelFlux { /** * Run the specified runnable when a ' rail ' receives a cancellation .
 * @ param onCancel the callback
 * @ return the new { @ link ParallelFlux } instance */
public final ParallelFlux < T > doOnCancel ( Runnable onCancel ) { } }

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


signature:
public class ParallelFlux { /** * Run the specified runnable when a ' rail ' completes .
 * @ param onComplete the callback
 * @ return the new { @ link ParallelFlux } instance */
public final ParallelFlux < T > doOnComplete ( Runnable onComplete ) { } }

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


signature:
public class ParallelFlux { /** * Triggers side - effects when the { @ link ParallelFlux } emits an item , fails with an error
 * or completes successfully . All these events are represented as a { @ link Signal }
 * that is passed to the side - effect callback . Note that with { @ link ParallelFlux } and
 * the { @ link # subscribe ( Consumer ) lambda - based subscribes } or the
 * { @ link # subscribe ( CoreSubscriber [ ] ) array - based one } , onError and onComplete will be
 * invoked as many times as there are rails , resulting in as many corresponding
 * { @ link Signal } being seen in the callback .
 * Use of { @ link # subscribe ( Subscriber ) } , which calls { @ link # sequential ( ) } , might
 * cancel some rails , resulting in less signals being observed . This is an advanced
 * operator , typically used for monitoring of a ParallelFlux .
 * @ param signalConsumer the mandatory callback to call on
 * { @ link Subscriber # onNext ( Object ) } , { @ link Subscriber # onError ( Throwable ) } and
 * { @ link Subscriber # onComplete ( ) }
 * @ return an observed { @ link ParallelFlux }
 * @ see # doOnNext ( Consumer )
 * @ see # doOnError ( Consumer )
 * @ see # doOnComplete ( Runnable )
 * @ see # subscribe ( CoreSubscriber [ ] )
 * @ see Signal */
public final ParallelFlux < T > doOnEach ( Consumer < ? super Signal < T > > signalConsumer ) { } }

implementation: Objects . requireNonNull ( signalConsumer , "signalConsumer" ) ; return onAssembly ( new ParallelDoOnEach < > ( this , ( ctx , v ) -> signalConsumer . accept ( Signal . next ( v , ctx ) ) , ( ctx , e ) -> signalConsumer . accept ( Signal . error ( e , ctx ) ) , ctx -> signalConsumer . accept ( Signal . complete ( ctx ) ) ) ) ;


signature:
public class ParallelFlux { /** * Call the specified consumer with the exception passing through any ' rail ' .
 * @ param onError the callback
 * @ return the new { @ link ParallelFlux } instance */
public final ParallelFlux < T > doOnError ( Consumer < ? super Throwable > onError ) { } }

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


signature:
public class ParallelFlux { /** * Call the specified consumer with the request amount if any rail receives a
 * request .
 * @ param onRequest the callback
 * @ return the new { @ link ParallelFlux } instance */
public final ParallelFlux < T > doOnRequest ( LongConsumer onRequest ) { } }

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


signature:
public class ParallelFlux { /** * Filters the source values on each ' rail ' .
 * Note that the same predicate may be called from multiple threads concurrently .
 * @ param predicate the function returning true to keep a value or false to drop a
 * value
 * @ return the new { @ link ParallelFlux } instance */
public final ParallelFlux < T > filter ( Predicate < ? super T > predicate ) { } }

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


signature:
public class ParallelFlux { /** * Generates and flattens Publishers on each ' rail ' .
 * Errors are not delayed and uses unbounded concurrency along with default inner
 * prefetch .
 * @ param < R > the result type
 * @ param mapper the function to map each rail ' s value into a Publisher
 * @ return the new { @ link ParallelFlux } instance */
public final < R > ParallelFlux < R > flatMap ( Function < ? super T , ? extends Publisher < ? extends R > > mapper ) { } }

implementation: return flatMap ( mapper , false , Integer . MAX_VALUE , Queues . SMALL_BUFFER_SIZE ) ;


signature:
public class ParallelFlux { /** * Generates and flattens Publishers on each ' rail ' , optionally delaying errors and
 * having a total number of simultaneous subscriptions to the inner Publishers .
 * It uses a default inner prefetch .
 * @ param < R > the result type
 * @ param mapper the function to map each rail ' s value into a Publisher
 * @ param delayError should the errors from the main and the inner sources delayed
 * till everybody terminates ?
 * @ param maxConcurrency the maximum number of simultaneous subscriptions to the
 * generated inner Publishers
 * @ return the new { @ link ParallelFlux } instance */
public final < R > ParallelFlux < R > flatMap ( Function < ? super T , ? extends Publisher < ? extends R > > mapper , boolean delayError , int maxConcurrency ) { } }

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


signature:
public class ParallelFlux { /** * Generates and flattens Publishers on each ' rail ' , optionally delaying errors ,
 * having a total number of simultaneous subscriptions to the inner Publishers and
 * using the given prefetch amount for the inner Publishers .
 * @ param < R > the result type
 * @ param mapper the function to map each rail ' s value into a Publisher
 * @ param delayError should the errors from the main and the inner sources delayed
 * till everybody terminates ?
 * @ param maxConcurrency the maximum number of simultaneous subscriptions to the
 * generated inner Publishers
 * @ param prefetch the number of items to prefetch from each inner Publisher
 * @ return the new { @ link ParallelFlux } instance */
public final < R > ParallelFlux < R > flatMap ( Function < ? super T , ? extends Publisher < ? extends R > > mapper , boolean delayError , int maxConcurrency , int prefetch ) { } }

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


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

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


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

implementation: return log ( category , level , false , options ) ;


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

implementation: return onAssembly ( new ParallelLog < > ( this , new SignalLogger < > ( this , category , level , showOperatorLine , options ) ) ) ;


signature:
public class ParallelFlux { /** * Maps the source values on each ' rail ' to another value .
 * Note that the same mapper function may be called from multiple threads
 * concurrently .
 * @ param < U > the output value type
 * @ param mapper the mapper function turning Ts into Us .
 * @ return the new { @ link ParallelFlux } instance */
public final < U > ParallelFlux < U > map ( Function < ? super T , ? extends U > mapper ) { } }

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


signature:
public class ParallelFlux { /** * Merges the values from each ' rail ' , but choose which one to merge by way of a
 * provided { @ link Comparator } , picking the smallest of all rails . The result is
 * exposed back as a { @ link Flux } .
 * This version uses a default prefetch of { @ link Queues # SMALL _ BUFFER _ SIZE } .
 * @ param comparator the comparator to choose the smallest value available from all rails
 * @ return the new Flux instance
 * @ see ParallelFlux # ordered ( Comparator , int ) */
public final Flux < T > ordered ( Comparator < ? super T > comparator ) { } }

implementation: return ordered ( comparator , Queues . SMALL_BUFFER_SIZE ) ;


signature:
public class ParallelFlux { /** * Merges the values from each ' rail ' , but choose which one to merge by way of a
 * provided { @ link Comparator } , picking the smallest of all rails . The result is
 * exposed back as a { @ link Flux } .
 * @ param comparator the comparator to choose the smallest value available from all rails
 * @ param prefetch the prefetch to use
 * @ return the new Flux instance
 * @ see ParallelFlux # ordered ( Comparator ) */
public final Flux < T > ordered ( Comparator < ? super T > comparator , int prefetch ) { } }

implementation: return new ParallelMergeOrdered < > ( this , prefetch , Queues . get ( prefetch ) , comparator ) ;


signature:
public class ParallelFlux { /** * Reduces all values within a ' rail ' and across ' rails ' with a reducer function into
 * a single sequential value .
 * Note that the same reducer function may be called from multiple threads
 * concurrently .
 * @ param reducer the function to reduce two values into one .
 * @ return the new Mono instance emitting the reduced value or empty if the
 * { @ link ParallelFlux } was empty */
public final Mono < T > reduce ( BiFunction < T , T , T > reducer ) { } }

implementation: Objects . requireNonNull ( reducer , "reducer" ) ; return Mono . onAssembly ( new ParallelMergeReduce < > ( this , reducer ) ) ;


signature:
public class ParallelFlux { /** * Reduces all values within a ' rail ' to a single value ( with a possibly different
 * type ) via a reducer function that is initialized on each rail from an
 * initialSupplier value .
 * Note that the same mapper function may be called from multiple threads
 * concurrently .
 * @ param < R > the reduced output type
 * @ param initialSupplier the supplier for the initial value
 * @ param reducer the function to reduce a previous output of reduce ( or the initial
 * value supplied ) with a current source value .
 * @ return the new { @ link ParallelFlux } instance */
public final < R > ParallelFlux < R > reduce ( Supplier < R > initialSupplier , BiFunction < R , ? super T , R > reducer ) { } }

implementation: Objects . requireNonNull ( initialSupplier , "initialSupplier" ) ; Objects . requireNonNull ( reducer , "reducer" ) ; return onAssembly ( new ParallelReduceSeed < > ( this , initialSupplier , reducer ) ) ;


signature:
public class ParallelFlux { /** * Specifies where each ' rail ' will observe its incoming values with possibly
 * work - stealing and a given prefetch amount .
 * This operator uses the default prefetch size returned by { @ code
 * Queues . SMALL _ BUFFER _ SIZE } .
 * The operator will call { @ code Scheduler . createWorker ( ) } as many times as this
 * ParallelFlux ' s parallelism level is .
 * No assumptions are made about the Scheduler ' s parallelism level , if the Scheduler ' s
 * parallelism level is lower than the ParallelFlux ' s , some rails may end up on
 * the same thread / worker .
 * This operator doesn ' t require the Scheduler to be trampolining as it does its own
 * built - in trampolining logic .
 * @ param scheduler the scheduler to use that rail ' s worker has run out of work .
 * @ param prefetch the number of values to request on each ' rail ' from the source
 * @ return the new { @ link ParallelFlux } instance */
public final ParallelFlux < T > runOn ( Scheduler scheduler , int prefetch ) { } }

implementation: Objects . requireNonNull ( scheduler , "scheduler" ) ; return onAssembly ( new ParallelRunOn < > ( this , scheduler , prefetch , Queues . get ( prefetch ) ) ) ;


signature:
public class ParallelFlux { /** * Merges the values from each ' rail ' in a round - robin or same - order fashion and
 * exposes it as a regular Publisher sequence , running with a give prefetch value for
 * the rails .
 * @ param prefetch the prefetch amount to use for each rail
 * @ return the new Flux instance */
public final Flux < T > sequential ( int prefetch ) { } }

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


signature:
public class ParallelFlux { /** * Sorts the ' rails ' of this { @ link ParallelFlux } and returns a Publisher that
 * sequentially picks the smallest next value from the rails .
 * This operator requires a finite source ParallelFlux .
 * @ param comparator the comparator to use
 * @ param capacityHint the expected number of total elements
 * @ return the new Flux instance */
public final Flux < T > sorted ( Comparator < ? super T > comparator , int capacityHint ) { } }

implementation: int ch = capacityHint / parallelism ( ) + 1 ; ParallelFlux < List < T > > railReduced = reduce ( ( ) -> new ArrayList < > ( ch ) , ( a , b ) -> { a . add ( b ) ; return a ; } ) ; ParallelFlux < List < T > > railSorted = railReduced . map ( list -> { list . sort ( comparator ) ; return list ; } ) ; return Flux . onAssembly ( new ParallelMergeSort < > ( railSorted , comparator ) ) ;


signature:
public class ParallelFlux { /** * Subscribes to this { @ link ParallelFlux } by providing an onNext and onError callback
 * and triggers the execution chain for all ' rails ' .
 * @ param onNext consumer of onNext signals
 * @ param onError consumer of error signal */
public final Disposable subscribe ( @ Nullable Consumer < ? super T > onNext , Consumer < ? super Throwable > onError ) { } }

implementation: return subscribe ( onNext , onError , null ) ;


signature:
public class ParallelFlux { /** * Subscribes to this { @ link ParallelFlux } by providing an onNext , onError ,
 * onComplete and onSubscribe callback and triggers the execution chain for all
 * ' rails ' .
 * @ param onNext consumer of onNext signals
 * @ param onError consumer of error signal
 * @ param onComplete callback on completion signal
 * @ param onSubscribe consumer of the subscription signal */
public final Disposable subscribe ( @ Nullable Consumer < ? super T > onNext , @ Nullable Consumer < ? super Throwable > onError , @ Nullable Runnable onComplete , @ Nullable Consumer < ? super Subscription > onSubscribe ) { } }

implementation: CorePublisher < T > publisher = Operators . onLastAssembly ( this ) ; if ( publisher instanceof ParallelFlux ) { @ SuppressWarnings ( "unchecked" ) LambdaSubscriber < ? super T > [ ] subscribers = new LambdaSubscriber [ parallelism ( ) ] ; int i = 0 ; while ( i < subscribers . length ) { subscribers [ i ++ ] = new LambdaSubscriber < > ( onNext , onError , onComplete , onSubscribe ) ; } ( ( ParallelFlux < T > ) publisher ) . subscribe ( subscribers ) ; return Disposables . composite ( subscribers ) ; } else { LambdaSubscriber < ? super T > subscriber = new LambdaSubscriber < > ( onNext , onError , onComplete , onSubscribe ) ; publisher . subscribe ( Operators . toCoreSubscriber ( new FluxHide . SuppressFuseableSubscriber < > ( subscriber ) ) ) ; return subscriber ; }


signature:
public class ParallelFlux { /** * Merge the rails into a { @ link # sequential ( ) } Flux and
 * { @ link Flux # subscribe ( Subscriber ) subscribe } to said Flux .
 * @ param s the subscriber to use on { @ link # sequential ( ) } Flux */
@ Override @ SuppressWarnings ( "unchecked" ) public final void subscribe ( Subscriber < ? super T > s ) { } }

implementation: FluxHide . SuppressFuseableSubscriber < T > subscriber = new FluxHide . SuppressFuseableSubscriber < > ( Operators . toCoreSubscriber ( s ) ) ; Operators . onLastAssembly ( sequential ( ) ) . subscribe ( Operators . toCoreSubscriber ( subscriber ) ) ;


signature:
public class ParallelFlux { /** * Tag this ParallelFlux 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 ParallelFlux < T > tag ( String key , String value ) { } }

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


signature:
public class ParallelFlux { /** * Allows composing operators , in assembly time , on top of this { @ link ParallelFlux }
 * and returns another { @ link ParallelFlux } with composed features .
 * @ param < U > the output value type
 * @ param composer the composer function from { @ link ParallelFlux } ( this ) to another
 * ParallelFlux
 * @ return the { @ link ParallelFlux } returned by the function */
public final < U > ParallelFlux < U > transform ( Function < ? super ParallelFlux < T > , ParallelFlux < U > > composer ) { } }

implementation: return onAssembly ( as ( composer ) ) ;


signature:
public class ParallelFlux { /** * Validates the number of subscribers and returns true if their number matches the
 * parallelism level of this ParallelFlux .
 * @ param subscribers the array of Subscribers
 * @ return true if the number of subscribers equals to the parallelism level */
protected final boolean validate ( Subscriber < ? > [ ] subscribers ) { } }

implementation: int p = parallelism ( ) ; if ( subscribers . length != p ) { IllegalArgumentException iae = new IllegalArgumentException ( "parallelism = " + "" + p + ", subscribers = " + subscribers . length ) ; for ( Subscriber < ? > s : subscribers ) { Operators . error ( s , iae ) ; } return false ; } return true ;


signature:
public class ParallelFlux { /** * Generates and concatenates Publishers on each ' rail ' , optionally delaying errors
 * and using the given prefetch amount for generating Publishers upfront .
 * @ param < R > the result type
 * @ param mapper the function to map each rail ' s value into a Publisher
 * @ param prefetch the number of items to prefetch from each inner Publisher
 * @ param errorMode the error handling , i . e . , when to report errors from the main
 * source and the inner Publishers ( immediate , boundary , end )
 * @ return the new { @ link ParallelFlux } instance */
final < R > ParallelFlux < R > concatMap ( Function < ? super T , ? extends Publisher < ? extends R > > mapper , int prefetch , ErrorMode errorMode ) { } }

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


signature:
public class ParallelFlux { /** * Generates and concatenates Publishers on each ' rail ' , delaying errors
 * and using the given prefetch amount for generating Publishers upfront .
 * @ param < R > the result type
 * @ param mapper the function to map each rail ' s value into a Publisher
 * @ param delayUntilEnd true if delayed until all sources are concatenated
 * @ param prefetch the number of items to prefetch from each inner Publisher
 * source and the inner Publishers ( immediate , boundary , end )
 * @ return the new { @ link ParallelFlux } instance */
final < R > ParallelFlux < R > concatMapDelayError ( Function < ? super T , ? extends Publisher < ? extends R > > mapper , boolean delayUntilEnd , int prefetch ) { } }

implementation: return concatMap ( mapper , prefetch , delayUntilEnd ? ErrorMode . END : ErrorMode . BOUNDARY ) ;


signature:
public class Traces { /** * Return true for strings ( usually from a stack trace element ) that should be
 * sanitized out by { @ link Traces # callSiteSupplierFactory } .
 * @ param stackTraceRow the row to check
 * @ return true if it should be sanitized out , false if it should be kept */
static boolean shouldSanitize ( String stackTraceRow ) { } }

implementation: return stackTraceRow . startsWith ( "java.util.function" ) || stackTraceRow . startsWith ( "reactor.core.publisher.Mono.onAssembly" ) || stackTraceRow . equals ( "reactor.core.publisher.Mono.onAssembly" ) || stackTraceRow . equals ( "reactor.core.publisher.Flux.onAssembly" ) || stackTraceRow . equals ( "reactor.core.publisher.ParallelFlux.onAssembly" ) || stackTraceRow . startsWith ( "reactor.core.publisher.SignalLogger" ) || stackTraceRow . startsWith ( "reactor.core.publisher.FluxOnAssembly" ) || stackTraceRow . startsWith ( "reactor.core.publisher.MonoOnAssembly." ) || stackTraceRow . startsWith ( "reactor.core.publisher.MonoCallableOnAssembly." ) || stackTraceRow . startsWith ( "reactor.core.publisher.FluxCallableOnAssembly." ) || stackTraceRow . startsWith ( "reactor.core.publisher.Hooks" ) || stackTraceRow . startsWith ( "sun.reflect" ) || stackTraceRow . startsWith ( "java.util.concurrent.ThreadPoolExecutor" ) || stackTraceRow . startsWith ( "java.lang.reflect" ) ;


signature:
public class Traces { /** * Extract operator information out of an assembly stack trace in { @ link String } form
 * ( see { @ link Traces # callSiteSupplierFactory } ) which potentially
 * has a header line that one can skip by setting { @ code skipFirst } to { @ code true } .
 * Most operators will result in a line of the form { @ code " Flux . map ⇢ user . code . Class . method ( Class . java : 123 ) " } ,
 * that is :
 * < ol >
 * < li > The top of the stack is inspected for Reactor API references , and the deepest
 * one is kept , since multiple API references generally denote an alias operator .
 * ( eg . { @ code " Flux . map " } ) < / li >
 * < li > The next stacktrace element is considered user code and is appended to the
 * result with a { @ code ⇢ } separator . ( eg . { @ code " ⇢ user . code . Class . method ( Class . java : 123 ) " } ) < / li >
 * < li > If no user code is found in the sanitized stack , then the API reference is outputed in the later format only . < / li >
 * < li > If the sanitized stack is empty , returns { @ code " [ no operator assembly information ] " } < / li >
 * < / ol >
 * @ param source the sanitized assembly stacktrace in String format .
 * @ return a { @ link String } representing operator and operator assembly site extracted
 * from the assembly stack trace . */
static String extractOperatorAssemblyInformation ( String source , boolean skipFirst ) { } }

implementation:
String [ ] uncleanTraces = source . split ( "\n" ) ; final List < String > traces = Stream . of ( uncleanTraces ) . map ( String :: trim ) . filter ( s -> ! s . isEmpty ( ) ) . skip ( skipFirst ? 1 : 0 ) . collect ( Collectors . toList ( ) ) ; if ( traces . isEmpty ( ) ) { return "[no operator assembly information]" ; } int i = 0 ; while ( i < traces . size ( ) && ! isUserCode ( traces . get ( i ) ) ) { i ++ ; } String apiLine ; String userCodeLine ; if ( i == 0 ) { // no line was a reactor API line
apiLine = "" ; userCodeLine = traces . get ( 0 ) ; } else if ( i == traces . size ( ) ) { // we skipped ALL lines , meaning they ' re all reactor API lines . We ' ll fully display the last one
apiLine = "" ; userCodeLine = traces . get ( i - 1 ) . replaceFirst ( "reactor.core.publisher." , "" ) ; } else { // currently on user code line , previous one is API
apiLine = traces . get ( i - 1 ) ; userCodeLine = traces . get ( i ) ; } // now we want something in the form " Flux . map ⇢ user . code . Class . method ( Class . java : 123 ) "
if ( apiLine . isEmpty ( ) ) return userCodeLine ; int linePartIndex = apiLine . indexOf ( '(' ) ; if ( linePartIndex > 0 ) { apiLine = apiLine . substring ( 0 , linePartIndex ) ; } apiLine = apiLine . replaceFirst ( "reactor.core.publisher." , "" ) ; return apiLine + " ⇢ " + userCodeLine ;


signature:
public class FluxMetrics { /** * Extract the name and tags from the upstream , and detect if there was an actual name
 * ( ie . distinct from { @ link Scannable # stepName ( ) } ) set by the user .
 * @ param source the upstream
 * @ return a { @ link Tuple2 } of name and list of { @ link Tag } */
static Tuple2 < String , List < Tag > > resolveNameAndTags ( Publisher < ? > source ) { } }

implementation:
// resolve the tags and names at instantiation
String name ; List < Tag > tags ; Scannable scannable = Scannable . from ( source ) ; if ( scannable . isScanAvailable ( ) ) { String nameOrDefault = scannable . name ( ) ; if ( scannable . stepName ( ) . equals ( nameOrDefault ) ) { name = REACTOR_DEFAULT_NAME ; } else { name = nameOrDefault ; } tags = scannable . tags ( ) . map ( tuple -> Tag . of ( tuple . getT1 ( ) , tuple . getT2 ( ) ) ) . collect ( Collectors . toList ( ) ) ; } else { LOGGER . warn ( "Attempting to activate metrics but the upstream is not Scannable. " + "You might want to use `name()` (and optionally `tags()`) right before `metrics()`" ) ; name = REACTOR_DEFAULT_NAME ; tags = Collections . emptyList ( ) ; } return Tuples . of ( name , tags ) ;


signature:
public class Tuple6 { /** * Map the 1st part ( T1 ) of this { @ link Tuple6 } 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 Tuple6 } with a different T1 value */
public < R > Tuple6 < R , T2 , T3 , T4 , T5 , T6 > mapT1 ( Function < T1 , R > mapper ) { } }

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


signature:
public class Exceptions { /** * Update an empty atomic reference with the given exception , or combine further added
 * exceptions together as suppressed exceptions under a root Throwable with
 * the { @ code " Multiple exceptions " } message , if the atomic reference already holds
 * one . This is short - circuited if the reference contains { @ link # TERMINATED } .
 * @ param < T > the parent instance type
 * @ param field the target field updater
 * @ param instance the parent instance for the field
 * @ param exception the Throwable to add .
 * @ return true if added , false if the field contained the { @ link # TERMINATED }
 * instance .
 * @ see # unwrapMultiple ( Throwable ) */
public static < T > boolean addThrowable ( AtomicReferenceFieldUpdater < T , Throwable > field , T instance , Throwable exception ) { } }

implementation:
for ( ; ; ) { Throwable current = field . get ( instance ) ; if ( current == TERMINATED ) { return false ; } if ( current instanceof CompositeException ) { // this is ok , composite exceptions are never singletons
current . addSuppressed ( exception ) ; return true ; } Throwable update ; if ( current == null ) { update = exception ; } else { update = multiple ( current , exception ) ; } if ( field . compareAndSet ( instance , current , update ) ) { return true ; } }


signature:
public class Exceptions { /** * Create a composite exception that wraps the given { @ link Throwable Throwable ( s ) } ,
 * as suppressed exceptions . Instances create by this method can be detected using the
 * { @ link # isMultiple ( Throwable ) } check . The { @ link # unwrapMultiple ( Throwable ) } method
 * will correctly unwrap these to a { @ link List } of the suppressed exceptions . Note
 * that is will also be consistent in producing a List for other types of exceptions
 * by putting the input inside a single - element List .
 * @ param throwables the exceptions to wrap into a composite
 * @ return a composite exception with a standard message , and the given throwables as
 * suppressed exceptions
 * @ see # addThrowable ( AtomicReferenceFieldUpdater , Object , Throwable ) */
public static RuntimeException multiple ( Throwable ... throwables ) { } }

implementation:
CompositeException multiple = new CompositeException ( ) ; // noinspection ConstantConditions
if ( throwables != null ) { for ( Throwable t : throwables ) { // this is ok , multiple is always a new non - singleton instance
multiple . addSuppressed ( t ) ; } } return multiple ;


signature:
public class Exceptions { /** * Create a composite exception that wraps the given { @ link Throwable Throwable ( s ) } ,
 * as suppressed exceptions . Instances create by this method can be detected using the
 * { @ link # isMultiple ( Throwable ) } check . The { @ link # unwrapMultiple ( Throwable ) } method
 * will correctly unwrap these to a { @ link List } of the suppressed exceptions . Note
 * that is will also be consistent in producing a List for other types of exceptions
 * by putting the input inside a single - element List .
 * @ param throwables the exceptions to wrap into a composite
 * @ return a composite exception with a standard message , and the given throwables as
 * suppressed exceptions
 * @ see # addThrowable ( AtomicReferenceFieldUpdater , Object , Throwable ) */
public static RuntimeException multiple ( Iterable < Throwable > throwables ) { } }

implementation:
RuntimeException multiple = new RuntimeException ( "Multiple exceptions" ) ; // noinspection ConstantConditions
if ( throwables != null ) { for ( Throwable t : throwables ) { // this is ok , multiple is always a new non - singleton instance
multiple . addSuppressed ( t ) ; } } return multiple ;


signature:
public class Exceptions { /** * Return a new { @ link RejectedExecutionException } with standard message and cause ,
 * unless the { @ code cause } is already a { @ link RejectedExecutionException } created
 * via { @ link # failWithRejected ( Throwable ) } ( not the singleton - producing variants ) .
 * @ param cause the original exception that caused the rejection
 * @ return a new { @ link RejectedExecutionException } with standard message and cause */
public static RejectedExecutionException failWithRejected ( Throwable cause ) { } }

implementation: if ( cause instanceof ReactorRejectedExecutionException ) { return ( RejectedExecutionException ) cause ; } return new ReactorRejectedExecutionException ( "Scheduler unavailable" , cause ) ;


signature:
public class Exceptions { /** * Check if the given error is a { @ link # errorCallbackNotImplemented ( Throwable ) callback not implemented }
 * exception , in which case its { @ link Throwable # getCause ( ) cause } will be the propagated
 * error that couldn ' t be processed .
 * @ param t the { @ link Throwable } error to check
 * @ return true if given { @ link Throwable } is a callback not implemented exception . */
public static boolean isErrorCallbackNotImplemented ( @ Nullable Throwable t ) { } }

implementation: return t != null && t . getClass ( ) . equals ( ErrorCallbackNotImplemented . class ) ;


signature:
public class Exceptions { /** * Prepare an unchecked { @ link RuntimeException } that should be propagated
 * downstream through { @ link org . reactivestreams . Subscriber # onError ( Throwable ) } .
 * < p > This method invokes { @ link # throwIfFatal ( Throwable ) } .
 * @ param t the root cause
 * @ return an unchecked exception to propagate through onError signals . */
public static RuntimeException propagate ( Throwable t ) { } }

implementation: throwIfFatal ( t ) ; if ( t instanceof RuntimeException ) { return ( RuntimeException ) t ; } return new ReactiveException ( t ) ;


signature:
public class Exceptions { /** * Atomic utility to safely mark a volatile throwable reference with a terminal
 * marker .
 * @ param field the atomic container
 * @ param instance the reference instance
 * @ param < T > the instance type
 * @ return the previously masked throwable */
@ Nullable public static < T > Throwable terminate ( AtomicReferenceFieldUpdater < T , Throwable > field , T instance ) { } }

implementation: Throwable current = field . get ( instance ) ; if ( current != TERMINATED ) { current = field . getAndSet ( instance , TERMINATED ) ; } return current ;


signature:
public class Exceptions { /** * Throws a particular { @ code Throwable } only if it belongs to a set of " fatal " error
 * varieties . These varieties are as follows : < ul >
 * < li > { @ code BubblingException } ( as detectable by { @ link # isBubbling ( Throwable ) } ) < / li >
 * < li > { @ code ErrorCallbackNotImplemented } ( as detectable by { @ link # isErrorCallbackNotImplemented ( Throwable ) } ) < / li >
 * < li > { @ link VirtualMachineError } < / li > < li > { @ link ThreadDeath } < / li > < li > { @ link LinkageError } < / li > < / ul >
 * @ param t the exception to evaluate */
public static void throwIfFatal ( @ Nullable Throwable t ) { } }

implementation: if ( t instanceof BubblingException ) { throw ( BubblingException ) t ; } if ( t instanceof ErrorCallbackNotImplemented ) { throw ( ErrorCallbackNotImplemented ) t ; } throwIfJvmFatal ( t ) ;


signature:
public class Exceptions { /** * Throws a particular { @ code Throwable } only if it belongs to a set of " fatal " error
 * varieties native to the JVM . These varieties are as follows :
 * < ul > < li > { @ link VirtualMachineError } < / li > < li > { @ link ThreadDeath } < / li >
 * < li > { @ link LinkageError } < / li > < / ul >
 * @ param t the exception to evaluate */
public static void throwIfJvmFatal ( @ Nullable Throwable t ) { } }

implementation: if ( t instanceof VirtualMachineError ) { throw ( VirtualMachineError ) t ; } if ( t instanceof ThreadDeath ) { throw ( ThreadDeath ) t ; } if ( t instanceof LinkageError ) { throw ( LinkageError ) t ; }


signature:
public class Exceptions { /** * Unwrap a particular { @ code Throwable } only if it is was wrapped via
 * { @ link # bubble ( Throwable ) bubble } or { @ link # propagate ( Throwable ) propagate } .
 * @ param t the exception to unwrap
 * @ return the unwrapped exception or current one if null */
public static Throwable unwrap ( Throwable t ) { } }

implementation: Throwable _t = t ; while ( _t instanceof ReactiveException ) { _t = _t . getCause ( ) ; } return _t == null ? t : _t ;


signature:
public class Exceptions { /** * Attempt to unwrap a { @ link Throwable } into a { @ link List } of Throwables . This is
 * only done on the condition that said Throwable is a composite exception built by
 * { @ link # multiple ( Throwable . . . ) } , in which case the list contains the exceptions
 * wrapped as suppressed exceptions in the composite . In any other case , the list
 * only contains the input Throwable ( or is empty in case of null input ) .
 * @ param potentialMultiple the { @ link Throwable } to unwrap if multiple
 * @ return a { @ link List } of the exceptions suppressed by the { @ link Throwable } if
 * multiple , or a List containing the Throwable otherwise . Null input results in an
 * empty List . */
public static List < Throwable > unwrapMultiple ( @ Nullable Throwable potentialMultiple ) { } }

implementation: if ( potentialMultiple == null ) { return Collections . emptyList ( ) ; } if ( isMultiple ( potentialMultiple ) ) { return Arrays . asList ( potentialMultiple . getSuppressed ( ) ) ; } return Collections . singletonList ( potentialMultiple ) ;


signature:
public class Exceptions { /** * Safely suppress a { @ link Throwable } on a { @ link RuntimeException } . The returned
 * { @ link RuntimeException } , bearing the suppressed exception , is most often the same
 * as the original exception unless :
 * < ul >
 * < li > original and tentatively suppressed exceptions are one and the same : do
 * nothing but return the original . < / li >
 * < li > original is one of the singleton { @ link RejectedExecutionException } created
 * by Reactor : make a copy the { @ link RejectedExecutionException } , add the
 * suppressed exception to it and return that copy . < / li >
 * < / ul >
 * @ param original the original { @ link RuntimeException } to bear a suppressed exception
 * @ param suppressed the { @ link Throwable } to suppress
 * @ return the ( most of the time original ) { @ link RuntimeException } bearing the
 * suppressed { @ link Throwable } */
public static final RuntimeException addSuppressed ( RuntimeException original , Throwable suppressed ) { } }

implementation: if ( original == suppressed ) { return original ; } if ( original == REJECTED_EXECUTION || original == NOT_TIME_CAPABLE_REJECTED_EXECUTION ) { RejectedExecutionException ree = new RejectedExecutionException ( original . getMessage ( ) ) ; ree . addSuppressed ( suppressed ) ; return ree ; } else { original . addSuppressed ( suppressed ) ; return original ; }


signature:
public class SignalLogger { /** * Structured logging with level adaptation and operator ascii graph if required .
 * @ param signalType the type of signal being logged
 * @ param signalValue the value for the signal ( use empty string if not required ) */
void log ( SignalType signalType , Object signalValue ) { } }

implementation: String line = fuseable ? LOG_TEMPLATE_FUSEABLE : LOG_TEMPLATE ; if ( operatorLine != null ) { line = line + " " + operatorLine ; } if ( level == Level . FINEST ) { log . trace ( line , signalType , signalValue ) ; } else if ( level == Level . FINE ) { log . debug ( line , signalType , signalValue ) ; } else if ( level == Level . INFO ) { log . info ( line , signalType , signalValue ) ; } else if ( level == Level . WARNING ) { log . warn ( line , signalType , signalValue ) ; } else if ( level == Level . SEVERE ) { log . error ( line , signalType , signalValue ) ; }


signature:
public class SignalLogger { /** * Structured logging with level adaptation and operator ascii graph if required +
 * protection against loggers that detect objects like { @ link Fuseable . QueueSubscription }
 * as { @ link java . util . Collection } and attempt to use their iterator for logging .
 * @ see # log */
void safeLog ( SignalType signalType , Object signalValue ) { } }

implementation: if ( signalValue instanceof Fuseable . QueueSubscription ) { signalValue = String . valueOf ( signalValue ) ; if ( log . isDebugEnabled ( ) ) { log . debug ( "A Fuseable Subscription has been passed to the logging framework, this is generally a sign of a misplaced log(), " + "eg. 'window(2).log()' instead of 'window(2).flatMap(w -> w.log())'" ) ; } } try { log ( signalType , signalValue ) ; } catch ( UnsupportedOperationException uoe ) { log ( signalType , String . valueOf ( signalValue ) ) ; if ( log . isDebugEnabled ( ) ) { log . debug ( "UnsupportedOperationException has been raised by the logging framework, does your log() placement make sense? " + "eg. 'window(2).log()' instead of 'window(2).flatMap(w -> w.log())'" , uoe ) ; } }


signature: public class WebDriverManager { protected String preDownload ( String target , String version ) { } }

implementation: log . trace ( "Pre-download. target={}, version={}" , target , version ) ; return target ;


signature:
public class WebDriverManager { /** * This method works also for http : / / npm . taobao . org / and
 * https : / / bitbucket . org / mirrors . */
protected List < URL > getDriversFromMirror ( URL driverUrl ) throws IOException { } }

implementation: if ( mirrorLog ) { log . info ( "Crawling driver list from mirror {}" , driverUrl ) ; mirrorLog = true ; } else { log . trace ( "[Recursive call] Crawling driver list from mirror {}" , driverUrl ) ; } String driverStr = driverUrl . toString ( ) ; String driverUrlContent = driverUrl . getPath ( ) ; HttpResponse response = httpClient . execute ( httpClient . createHttpGet ( driverUrl ) ) ; try ( InputStream in = response . getEntity ( ) . getContent ( ) ) { org . jsoup . nodes . Document doc = Jsoup . parse ( in , null , "" ) ; Iterator < org . jsoup . nodes . Element > iterator = doc . select ( "a" ) . iterator ( ) ; List < URL > urlList = new ArrayList < > ( ) ; while ( iterator . hasNext ( ) ) { String link = iterator . next ( ) . attr ( "href" ) ; if ( link . contains ( "mirror" ) && link . endsWith ( SLASH ) ) { urlList . addAll ( getDriversFromMirror ( new URL ( driverStr + link . replace ( driverUrlContent , "" ) ) ) ) ; } else if ( link . startsWith ( driverUrlContent ) && ! link . contains ( "icons" ) ) { urlList . add ( new URL ( driverStr + link . replace ( driverUrlContent , "" ) ) ) ; } } return urlList ; }


signature:
public class FileDownloadNotificationListener { /** * The notification item with the { @ code task } is told to destroy .
 * @ param task The task used to identify the will be destroyed notification item . */
public void destroyNotification ( BaseDownloadTask task ) { } }

implementation: if ( disableNotification ( task ) ) { return ; } this . helper . showIndeterminate ( task . getId ( ) , task . getStatus ( ) ) ; final BaseNotificationItem n = this . helper . remove ( task . getId ( ) ) ; if ( ! interceptCancel ( task , n ) && n != null ) { n . cancel ( ) ; }


signature:
public class DownloadTask { /** * - - - - - Get - - - - - */
@ Override public int getId ( ) { } }

implementation: if ( mId != 0 ) { return mId ; } if ( ! TextUtils . isEmpty ( mPath ) && ! TextUtils . isEmpty ( mUrl ) ) { return mId = FileDownloadUtils . generateId ( mUrl , mPath , mPathAsDirectory ) ; } return 0 ;


signature:
public class DownloadLaunchRunnable { /** * the trial connection is for : 1 . etag verify ; 2 . partial support verify . */
private void trialConnect ( ) throws IOException , RetryDirectly , IllegalAccessException , FileDownloadSecurityException { } }

implementation: FileDownloadConnection trialConnection = null ; try { final ConnectionProfile trialConnectionProfile ; if ( isNeedForceDiscardRange ) { trialConnectionProfile = ConnectionProfile . ConnectionProfileBuild . buildTrialConnectionProfileNoRange ( ) ; } else { trialConnectionProfile = ConnectionProfile . ConnectionProfileBuild . buildTrialConnectionProfile ( ) ; } final ConnectTask trialConnectTask = new ConnectTask . Builder ( ) . setDownloadId ( model . getId ( ) ) . setUrl ( model . getUrl ( ) ) . setEtag ( model . getETag ( ) ) . setHeader ( userRequestHeader ) . setConnectionProfile ( trialConnectionProfile ) . build ( ) ; trialConnection = trialConnectTask . connect ( ) ; handleTrialConnectResult ( trialConnectTask . getRequestHeader ( ) , trialConnectTask , trialConnection ) ; } finally { if ( trialConnection != null ) trialConnection . ending ( ) ; }


signature:
public class FileDownloadBroadcastHandler { /** * Parse the { @ code intent } from the filedownloader broadcast .
 * @ param intent the intent from the broadcast .
 * @ return the file download model . */
public static FileDownloadModel parseIntent ( Intent intent ) { } }

implementation: if ( ! ACTION_COMPLETED . equals ( intent . getAction ( ) ) ) { throw new IllegalArgumentException ( FileDownloadUtils . formatString ( "can't recognize the intent with action %s, on the current" + " version we only support action [%s]" , intent . getAction ( ) , ACTION_COMPLETED ) ) ; } return intent . getParcelableExtra ( KEY_MODEL ) ;


signature:
public class FileDownloadLineAsync { /** * The { @ link FileDownloader # startForeground ( int , Notification ) } request . */
public boolean startForeground ( final int id , final Notification notification ) { } }

implementation: if ( FileDownloader . getImpl ( ) . isServiceConnected ( ) ) { FileDownloader . getImpl ( ) . startForeground ( id , notification ) ; return true ; } else { FileDownloader . getImpl ( ) . bindService ( new Runnable ( ) { @ Override public void run ( ) { FileDownloader . getImpl ( ) . startForeground ( id , notification ) ; } } ) ; return false ; }


signature:
public class DefaultConnectionCountAdapter { /** * 100MB */
@ Override public int determineConnectionCount ( int downloadId , String url , String path , long totalLength ) { } }

implementation: if ( totalLength < ONE_CONNECTION_UPPER_LIMIT ) { return 1 ; } if ( totalLength < TWO_CONNECTION_UPPER_LIMIT ) { return 2 ; } if ( totalLength < THREE_CONNECTION_UPPER_LIMIT ) { return 3 ; } if ( totalLength < FOUR_CONNECTION_UPPER_LIMIT ) { return 4 ; } return 5 ;


signature:
public class FileDownloadLine { /** * The { @ link FileDownloader # startForeground ( int , Notification ) } request . */
public void startForeground ( final int id , final Notification notification ) { } }

implementation: if ( FileDownloader . getImpl ( ) . isServiceConnected ( ) ) { FileDownloader . getImpl ( ) . startForeground ( id , notification ) ; return ; } final ConnectSubscriber subscriber = new ConnectSubscriber ( ) { @ Override public void connected ( ) { FileDownloader . getImpl ( ) . startForeground ( id , notification ) ; } @ Override public Object getValue ( ) { return null ; } } ; wait ( subscriber ) ;


signature:
public class FileDownloadLine { /** * The { @ link FileDownloader # getSoFar ( int ) } request . */
public long getSoFar ( final int id ) { } }

implementation: if ( FileDownloader . getImpl ( ) . isServiceConnected ( ) ) { return FileDownloader . getImpl ( ) . getSoFar ( id ) ; } final ConnectSubscriber subscriber = new ConnectSubscriber ( ) { private long mValue ; @ Override public void connected ( ) { mValue = FileDownloader . getImpl ( ) . getSoFar ( id ) ; } @ Override public Object getValue ( ) { return mValue ; } } ; wait ( subscriber ) ; return ( long ) subscriber . getValue ( ) ;


signature:
public class FileDownloadLine { /** * The { @ link FileDownloader # getStatus ( int , String ) } request . */
public byte getStatus ( final int id , final String path ) { } }

implementation: if ( FileDownloader . getImpl ( ) . isServiceConnected ( ) ) { return FileDownloader . getImpl ( ) . getStatus ( id , path ) ; } if ( path != null && new File ( path ) . exists ( ) ) { return FileDownloadStatus . completed ; } final ConnectSubscriber subscriber = new ConnectSubscriber ( ) { private byte mValue ; @ Override public void connected ( ) { mValue = FileDownloader . getImpl ( ) . getStatus ( id , path ) ; } @ Override public Object getValue ( ) { return mValue ; } } ; wait ( subscriber ) ; return ( byte ) subscriber . getValue ( ) ;


signature:
public class FileDownloadHeader { /** * We have already handled etag , and will add ' If - Match ' & ' Range ' value if it works .
 * @ see com . liulishuo . filedownloader . download . ConnectTask # addUserRequiredHeader */
public void add ( String name , String value ) { } }

implementation: if ( name == null ) throw new NullPointerException ( "name == null" ) ; if ( name . isEmpty ( ) ) throw new IllegalArgumentException ( "name is empty" ) ; if ( value == null ) throw new NullPointerException ( "value == null" ) ; if ( mHeaderMap == null ) { mHeaderMap = new HashMap < > ( ) ; } List < String > values = mHeaderMap . get ( name ) ; if ( values == null ) { values = new ArrayList < > ( ) ; mHeaderMap . put ( name , values ) ; } if ( ! values . contains ( value ) ) { values . add ( value ) ; }


signature:
public class FileDownloadHeader { /** * We have already handled etag , and will add ' If - Match ' & ' Range ' value if it works .
 * @ see com . liulishuo . filedownloader . download . ConnectTask # addUserRequiredHeader */
public void add ( String line ) { } }

implementation: String [ ] parsed = line . split ( ":" ) ; final String name = parsed [ 0 ] . trim ( ) ; final String value = parsed [ 1 ] . trim ( ) ; add ( name , value ) ;


signature:
public class FileDownloader { /** * Using this method to setup the FileDownloader only you want to register your own customize
 * components for Filedownloader , otherwise just using { @ link # setup ( Context ) } instead .
 * Please invoke this method on the { @ link Application # onCreate ( ) } because of the customize
 * components must be assigned before FileDownloader is running .
 * Such as :
 * class MyApplication extends Application {
 * public void onCreate ( ) {
 * FileDownloader . setupOnApplicationOnCreate ( this )
 * . idGenerator ( new MyIdGenerator ( ) )
 * . database ( new MyDatabase ( ) )
 * . commit ( ) ;
 * @ param application the application .
 * @ return the customize components maker . */
public static DownloadMgrInitialParams . InitCustomMaker setupOnApplicationOnCreate ( Application application ) { } }

implementation: final Context context = application . getApplicationContext ( ) ; FileDownloadHelper . holdContext ( context ) ; DownloadMgrInitialParams . InitCustomMaker customMaker = new DownloadMgrInitialParams . InitCustomMaker ( ) ; CustomComponentHolder . getImpl ( ) . setInitCustomMaker ( customMaker ) ; return customMaker ;


signature:
public class FileDownloader { /** * Start the download queue by the same listener .
 * @ param listener Used to assemble tasks which is bound by the same { @ code listener }
 * @ param isSerial Whether start tasks one by one rather than parallel .
 * @ return { @ code true } if start tasks successfully . */
public boolean start ( final FileDownloadListener listener , final boolean isSerial ) { } }

implementation: if ( listener == null ) { FileDownloadLog . w ( this , "Tasks with the listener can't start, because the listener " + "provided is null: [null, %B]" , isSerial ) ; return false ; } return isSerial ? getQueuesHandler ( ) . startQueueSerial ( listener ) : getQueuesHandler ( ) . startQueueParallel ( listener ) ;


signature:
public class FileDownloader { /** * Pause the download queue by the same { @ code listener } .
 * @ param listener the listener .
 * @ see # pause ( int ) */
public void pause ( final FileDownloadListener listener ) { } }

implementation: FileDownloadTaskLauncher . getImpl ( ) . expire ( listener ) ; final List < BaseDownloadTask . IRunningTask > taskList = FileDownloadList . getImpl ( ) . copy ( listener ) ; for ( BaseDownloadTask . IRunningTask task : taskList ) { task . getOrigin ( ) . pause ( ) ; }


signature:
public class FileDownloader { /** * Pause all tasks running in FileDownloader . */
public void pauseAll ( ) { } }

implementation:
FileDownloadTaskLauncher . getImpl ( ) . expireAll ( ) ; final BaseDownloadTask . IRunningTask [ ] downloadList = FileDownloadList . getImpl ( ) . copy ( ) ; for ( BaseDownloadTask . IRunningTask task : downloadList ) { task . getOrigin ( ) . pause ( ) ; } // double check , for case : File Download progress alive but ui progress has died and relived
// so FileDownloadList not always contain all running task exactly .
if ( FileDownloadServiceProxy . getImpl ( ) . isConnected ( ) ) { FileDownloadServiceProxy . getImpl ( ) . pauseAllTasks ( ) ; } else { PauseAllMarker . createMarker ( ) ; }


signature:
public class FileDownloader { /** * Pause downloading tasks with the { @ code id } .
 * @ param id the { @ code id } .
 * @ return The size of tasks has been paused .
 * @ see # pause ( FileDownloadListener ) */
public int pause ( final int id ) { } }

implementation: List < BaseDownloadTask . IRunningTask > taskList = FileDownloadList . getImpl ( ) . getDownloadingList ( id ) ; if ( null == taskList || taskList . isEmpty ( ) ) { FileDownloadLog . w ( this , "request pause but not exist %d" , id ) ; return 0 ; } for ( BaseDownloadTask . IRunningTask task : taskList ) { task . getOrigin ( ) . pause ( ) ; } return taskList . size ( ) ;


signature:
public class FileDownloader { /** * Clear the data with the provided { @ code id } .
 * Normally used to deleting the data in filedownloader database , when it is paused or in
 * downloading status . If you want to re - download it clearly .
 * < strong > Note : < / strong > YOU NO NEED to clear the data when it is already completed downloading
 * because the data would be deleted when it completed downloading automatically by
 * FileDownloader .
 * If there are tasks with the { @ code id } in downloading , will be paused first ;
 * If delete the data with the { @ code id } in the filedownloader database successfully , will try
 * to delete its intermediate downloading file and downloaded file .
 * @ param id the download { @ code id } .
 * @ param targetFilePath the target path .
 * @ return { @ code true } if the data with the { @ code id } in filedownloader database was deleted ,
 * and tasks with the { @ code id } was paused ; { @ code false } otherwise . */
public boolean clear ( final int id , final String targetFilePath ) { } }

implementation:
pause ( id ) ; if ( FileDownloadServiceProxy . getImpl ( ) . clearTaskData ( id ) ) { // delete the task data in the filedownloader database successfully or no data with the
// id in filedownloader database .
final File intermediateFile = new File ( FileDownloadUtils . getTempPath ( targetFilePath ) ) ; if ( intermediateFile . exists ( ) ) { // noinspection ResultOfMethodCallIgnored
intermediateFile . delete ( ) ; } final File targetFile = new File ( targetFilePath ) ; if ( targetFile . exists ( ) ) { // noinspection ResultOfMethodCallIgnored
targetFile . delete ( ) ; } return true ; } return false ;


signature:
public class FileDownloader { /** * Get downloaded bytes so far by the downloadId . */
public long getSoFar ( final int downloadId ) { } }

implementation: BaseDownloadTask . IRunningTask task = FileDownloadList . getImpl ( ) . get ( downloadId ) ; if ( task == null ) { return FileDownloadServiceProxy . getImpl ( ) . getSofar ( downloadId ) ; } return task . getOrigin ( ) . getLargeFileSoFarBytes ( ) ;


signature:
public class FileDownloader { /** * Get the total bytes of the target file for the task with the { code id } . */
public long getTotal ( final int id ) { } }

implementation: BaseDownloadTask . IRunningTask task = FileDownloadList . getImpl ( ) . get ( id ) ; if ( task == null ) { return FileDownloadServiceProxy . getImpl ( ) . getTotal ( id ) ; } return task . getOrigin ( ) . getLargeFileTotalBytes ( ) ;


signature:
public class FileDownloader { /** * Find the running task by { @ code url } and default path , and replace its listener with
 * the new one { @ code listener } .
 * @ return The target task ' s DownloadId , if not exist target task , and replace failed , will be 0
 * @ see # replaceListener ( int , FileDownloadListener )
 * @ see # replaceListener ( String , String , FileDownloadListener ) */
public int replaceListener ( String url , FileDownloadListener listener ) { } }

implementation: return replaceListener ( url , FileDownloadUtils . getDefaultSaveFilePath ( url ) , listener ) ;


signature:
public class FileDownloader { /** * Find the running task by { @ code url } and { @ code path } , and replace its listener with
 * the new one { @ code listener } .
 * @ return The target task ' s DownloadId , if not exist target task , and replace failed , will be 0
 * @ see # replaceListener ( String , FileDownloadListener )
 * @ see # replaceListener ( int , FileDownloadListener ) */
public int replaceListener ( String url , String path , FileDownloadListener listener ) { } }

implementation: return replaceListener ( FileDownloadUtils . generateId ( url , path ) , listener ) ;


signature:
public class FileDownloader { /** * Find the running task by { @ code id } , and replace its listener width the new one
 * { @ code listener } .
 * @ return The target task ' s DownloadId , if not exist target task , and replace failed , will be 0
 * @ see # replaceListener ( String , FileDownloadListener )
 * @ see # replaceListener ( String , String , FileDownloadListener ) */
public int replaceListener ( int id , FileDownloadListener listener ) { } }

implementation: final BaseDownloadTask . IRunningTask task = FileDownloadList . getImpl ( ) . get ( id ) ; if ( task == null ) { return 0 ; } task . getOrigin ( ) . setListener ( listener ) ; return task . getOrigin ( ) . getId ( ) ;


signature:
public class FileDownloader { /** * Start and bind the FileDownloader service and run { @ code runnable } as soon as the binding is
 * successful .
 * < strong > Tips : < / strong > The FileDownloader service will start and bind automatically when any
 * task is request to start .
 * @ param runnable the command will be executed as soon as the FileDownloader Service is
 * successfully bound .
 * @ see # isServiceConnected ( )
 * @ see # bindService ( )
 * @ see # addServiceConnectListener ( FileDownloadConnectListener ) */
public void bindService ( final Runnable runnable ) { } }

implementation: if ( isServiceConnected ( ) ) { runnable . run ( ) ; } else { FileDownloadServiceProxy . getImpl ( ) . bindStartByContext ( FileDownloadHelper . getAppContext ( ) , runnable ) ; }


signature:
public class FileDownloader { /** * Unbind and stop the downloader service when there is no task running in the FileDownloader .
 * @ return { @ code true } if unbind and stop the downloader service successfully , { @ code false }
 * there are some tasks running in the FileDownloader . */
public boolean unBindServiceIfIdle ( ) { } }

implementation:
// check idle
if ( ! isServiceConnected ( ) ) { return false ; } if ( FileDownloadList . getImpl ( ) . isEmpty ( ) && FileDownloadServiceProxy . getImpl ( ) . isIdle ( ) ) { unBindService ( ) ; return true ; } return false ;


signature:
public class FileDownloader { /** * Set the maximum count of the network thread , what is the number of simultaneous downloads in
 * FileDownloader .
 * @ param count the number of simultaneous downloads , scope : [ 1 , 12 ] .
 * @ return whether is successful to set the max network thread count .
 * If there are any actively executing tasks in FileDownloader , you will receive a warn
 * priority log int the logcat and this operation would be failed . */
public boolean setMaxNetworkThreadCount ( final int count ) { } }

implementation: if ( ! FileDownloadList . getImpl ( ) . isEmpty ( ) ) { FileDownloadLog . w ( this , "Can't change the max network thread count, because there " + "are actively executing tasks in FileDownloader, please try again after all" + " actively executing tasks are completed or invoking FileDownloader#pauseAll" + " directly." ) ; return false ; } return FileDownloadServiceProxy . getImpl ( ) . setMaxNetworkThreadCount ( count ) ;


signature:
public class FileDownloadManager { /** * synchronize for safe : check downloading , check resume , update data , execute runnable */
public synchronized void start ( final String url , final String path , final boolean pathAsDirectory , final int callbackProgressTimes , final int callbackProgressMinIntervalMillis , final int autoRetryTimes , final boolean forceReDownload , final FileDownloadHeader header , final boolean isWifiRequired ) { } }

implementation:
if ( FileDownloadLog . NEED_LOG ) { FileDownloadLog . d ( this , "request start the task with url(%s) path(%s) isDirectory(%B)" , url , path , pathAsDirectory ) ; } // clear pause all marker so that the delay pause all check doesn ' t obstruct normal start of
// new task
PauseAllMarker . clearMarker ( ) ; final int id = FileDownloadUtils . generateId ( url , path , pathAsDirectory ) ; FileDownloadModel model = mDatabase . find ( id ) ; List < ConnectionModel > dirConnectionModelList = null ; if ( ! pathAsDirectory && model == null ) { // try dir data .
final int dirCaseId = FileDownloadUtils . generateId ( url , FileDownloadUtils . getParent ( path ) , true ) ; model = mDatabase . find ( dirCaseId ) ; if ( model != null && path . equals ( model . getTargetFilePath ( ) ) ) { if ( FileDownloadLog . NEED_LOG ) { FileDownloadLog . d ( this , "task[%d] find model by dirCaseId[%d]" , id , dirCaseId ) ; } dirConnectionModelList = mDatabase . findConnectionModel ( dirCaseId ) ; } } if ( FileDownloadHelper . inspectAndInflowDownloading ( id , model , this , true ) ) { if ( FileDownloadLog . NEED_LOG ) { FileDownloadLog . d ( this , "has already started download %d" , id ) ; } return ; } final String targetFilePath = model != null ? model . getTargetFilePath ( ) : FileDownloadUtils . getTargetFilePath ( path , pathAsDirectory , null ) ; if ( FileDownloadHelper . inspectAndInflowDownloaded ( id , targetFilePath , forceReDownload , true ) ) { if ( FileDownloadLog . NEED_LOG ) { FileDownloadLog . d ( this , "has already completed downloading %d" , id ) ; } return ; } final long sofar = model != null ? model . getSoFar ( ) : 0 ; final String tempFilePath = model != null ? model . getTempFilePath ( ) : FileDownloadUtils . getTempPath ( targetFilePath ) ; if ( FileDownloadHelper . inspectAndInflowConflictPath ( id , sofar , tempFilePath , targetFilePath , this ) ) { if ( FileDownloadLog . NEED_LOG ) { FileDownloadLog . d ( this , "there is an another task with the same target-file-path %d %s" , id , targetFilePath ) ; } // because of the file is dirty for this task .
if ( model != null ) { mDatabase . remove ( id ) ; mDatabase . removeConnections ( id ) ; } return ; } // real start
// - create model
boolean needUpdate2DB ; if ( model != null && ( model . getStatus ( ) == FileDownloadStatus . paused || model . getStatus ( ) == FileDownloadStatus . error || model . getStatus ( ) == FileDownloadStatus . pending || model . getStatus ( ) == FileDownloadStatus . started || model . getStatus ( ) == FileDownloadStatus . connected ) // FileDownloadRunnable
// invoke # isBreakpointAvailable to determine whether it is really invalid .
) { if ( model . getId ( ) != id ) { // in try dir case .
mDatabase . remove ( model . getId ( ) ) ; mDatabase . removeConnections ( model . getId ( ) ) ; model . setId ( id ) ; model . setPath ( path , pathAsDirectory ) ; if ( dirConnectionModelList != null ) { for ( ConnectionModel connectionModel : dirConnectionModelList ) { connectionModel . setId ( id ) ; mDatabase . insertConnectionModel ( connectionModel ) ; } } needUpdate2DB = true ; } else { if ( ! TextUtils . equals ( url , model . getUrl ( ) ) ) { // for cover the case of reusing the downloaded processing with the different
// url ( using with idGenerator ) .
model . setUrl ( url ) ; needUpdate2DB = true ; } else { needUpdate2DB = false ; } } } else { if ( model == null ) { model = new FileDownloadModel ( ) ; } model . setUrl ( url ) ; model . setPath ( path , pathAsDirectory ) ; model . setId ( id ) ; model . setSoFar ( 0 ) ; model . setTotal ( 0 ) ; model . setStatus ( FileDownloadStatus . pending ) ; model . setConnectionCount ( 1 ) ; needUpdate2DB = true ; } // - update model to db
if ( needUpdate2DB ) { mDatabase . update ( model ) ; } final DownloadLaunchRunnable . Builder builder = new DownloadLaunchRunnable . Builder ( ) ; final DownloadLaunchRunnable runnable = builder . setModel ( model ) . setHeader ( header ) . setThreadPoolMonitor ( this ) . setMinIntervalMillis ( callbackProgressMinIntervalMillis ) . setCallbackProgressMaxCount ( callbackProgressTimes ) . setForceReDownload ( forceReDownload ) . setWifiRequired ( isWifiRequired ) . setMaxRetryTimes ( autoRetryTimes ) . build ( ) ; // - execute
mThreadPool . execute ( runnable ) ;


signature:
public class FileDownloadManager { /** * Pause all running task */
public void pauseAll ( ) { } }

implementation: List < Integer > list = mThreadPool . getAllExactRunningDownloadIds ( ) ; if ( FileDownloadLog . NEED_LOG ) { FileDownloadLog . d ( this , "pause all tasks %d" , list . size ( ) ) ; } for ( Integer id : list ) { pause ( id ) ; }


signature:
public class FileDownloadMessenger { /** * sync */
@ Override public void notifyBlockComplete ( MessageSnapshot snapshot ) { } }

implementation: if ( FileDownloadLog . NEED_LOG ) { FileDownloadLog . d ( this , "notify block completed %s %s" , mTask , Thread . currentThread ( ) . getName ( ) ) ; } mLifeCycleCallback . onIng ( ) ; process ( snapshot ) ;