signature:
public class ApplicationInformation { /** * 拆分网络地址为host and port
 * @ param hostAndPort 主机和端口
 * @ return 主机端口数组 */
public static String [ ] splitAddress ( String hostAndPort ) { } }

implementation: if ( hostAndPort . indexOf ( ':' ) == - 1 ) { throw new IllegalStateException ( "non exists port" ) ; } String [ ] result = hostAndPort . split ( ":" ) ; if ( StringUtils . isEmpty ( result [ 0 ] ) ) { result [ 0 ] = "0.0.0.0" ; return result ; } if ( result [ 0 ] . charAt ( 0 ) == '/' ) { result [ 0 ] = result [ 0 ] . substring ( 1 ) ; return result ; } return result ;


signature:
public class ApplicationInformation { /** * 模块HTTP端口号
 * @ param serverProperties serverProperties
 * @ return int */
public static int serverPort ( ServerProperties serverProperties ) { } }

implementation: return Objects . isNull ( serverProperties ) ? 0 : ( Objects . isNull ( serverProperties . getPort ( ) ) ? 8080 : serverProperties . getPort ( ) ) ;


signature:
public class DefaultGlobalContext { /** * 在用户业务前生成 , 业务后销毁
 * @ param groupId groupId */
@ Override public void destroyTx ( String groupId ) { } }

implementation: attachmentCache . remove ( groupId + ".dtx" ) ; log . debug ( "Destroy TxContext[{}]" , groupId ) ;


signature:
public class DTXServiceExecutor { /** * 事务业务执行
 * @ param info info
 * @ return Object
 * @ throws Throwable Throwable */
public Object transactionRunning ( TxTransactionInfo info ) throws Throwable { } }

implementation:
// 1 . 获取事务类型
String transactionType = info . getTransactionType ( ) ; // 2 . 获取事务传播状态
DTXPropagationState propagationState = propagationResolver . resolvePropagationState ( info ) ; // 2.1 如果不参与分布式事务立即终止
if ( propagationState . isIgnored ( ) ) { return info . getBusinessCallback ( ) . call ( ) ; } // 3 . 获取本地分布式事务控制器
DTXLocalControl dtxLocalControl = txLcnBeanHelper . loadDTXLocalControl ( transactionType , propagationState ) ; // 4 . 织入事务操作
try { // 4.1 记录事务类型到事务上下文
Set < String > transactionTypeSet = globalContext . txContext ( info . getGroupId ( ) ) . getTransactionTypes ( ) ; transactionTypeSet . add ( transactionType ) ; dtxLocalControl . preBusinessCode ( info ) ; // 4.2 业务执行前
txLogger . txTrace ( info . getGroupId ( ) , info . getUnitId ( ) , "pre business code, unit type: {}" , transactionType ) ; // 4.3 执行业务
Object result = dtxLocalControl . doBusinessCode ( info ) ; // 4.4 业务执行成功
txLogger . txTrace ( info . getGroupId ( ) , info . getUnitId ( ) , "business success" ) ; dtxLocalControl . onBusinessCodeSuccess ( info , result ) ; return result ; } catch ( TransactionException e ) { txLogger . error ( info . getGroupId ( ) , info . getUnitId ( ) , "before business code error" ) ; throw e ; } catch ( Throwable e ) { // 4.5 业务执行失败
txLogger . error ( info . getGroupId ( ) , info . getUnitId ( ) , Transactions . TAG_TRANSACTION , "business code error" ) ; dtxLocalControl . onBusinessCodeError ( info , e ) ; throw e ; } finally { // 4.6 业务执行完毕
dtxLocalControl . postBusinessCode ( info ) ; }


signature:
public class TableStruct { /** * 重写Getter
 * @ return getFullyQualifiedPrimaryKeys */
public List < String > getFullyQualifiedPrimaryKeys ( ) { } }

implementation: if ( this . fullyQualifiedPrimaryKeys != null ) { return this . fullyQualifiedPrimaryKeys ; } List < String > pks = new ArrayList < > ( ) ; this . getPrimaryKeys ( ) . forEach ( key -> pks . add ( tableName + '.' + key ) ) ; this . fullyQualifiedPrimaryKeys = pks ; return this . fullyQualifiedPrimaryKeys ;


signature:
public class MysqlLoggerHelper { /** * 按筛选条件获取记录数
 * @ param where where条件部分
 * @ param params 参数
 * @ return 总共记录数 */
private long total ( String where , Object ... params ) { } }

implementation: if ( logDbProperties . isEnabled ( ) ) { return dbHelper . query ( "select count(*) from t_logger where " + where , new ScalarHandler < > ( ) , params ) ; } else { throw new NotEnableLogException ( "not enable logger" ) ; }


signature:
public class TableStructAnalyser { /** * 存在数据表判断
 * @ param tableName tableName
 * @ return exists
 * @ throws SQLException SQLException */
public boolean existsTable ( String tableName ) throws SQLException { } }

implementation: Connection connection = null ; try { DTXLocalContext . makeUnProxy ( ) ; connection = dataSource . getConnection ( ) ; connection . setAutoCommit ( true ) ; return existsTable ( connection , tableName ) ; } finally { DbUtils . close ( connection ) ; DTXLocalContext . undoProxyStatus ( ) ; }


signature:
public class SocketManager { /** * 获取模块的远程标识keys
 * @ param moduleName 模块名称
 * @ return remoteKeys */
public List < String > removeKeys ( String moduleName ) { } }

implementation: List < String > allKeys = new ArrayList < > ( ) ; for ( Channel channel : channels ) { if ( moduleName . equals ( getModuleName ( channel ) ) ) { allKeys . add ( channel . remoteAddress ( ) . toString ( ) ) ; } } return allKeys ;


signature:
public class SocketManager { /** * 绑定连接数据
 * @ param remoteKey 远程标识
 * @ param appName 模块名称
 * @ param labelName TC标识名称 */
public void bindModuleName ( String remoteKey , String appName , String labelName ) throws RpcException { } }

implementation: AppInfo appInfo = new AppInfo ( ) ; appInfo . setAppName ( appName ) ; appInfo . setLabelName ( labelName ) ; appInfo . setCreateTime ( new Date ( ) ) ; if ( containsLabelName ( labelName ) ) { throw new RpcException ( "labelName:" + labelName + " has exist." ) ; } appNames . put ( remoteKey , appInfo ) ;


signature:
public class SocketManager { /** * 获取模块名称
 * @ param channel 管道信息
 * @ return 模块名称 */
public String getModuleName ( Channel channel ) { } }

implementation: String key = channel . remoteAddress ( ) . toString ( ) ; return getModuleName ( key ) ;


signature:
public class SocketManager { /** * 获取模块名称
 * @ param remoteKey 远程唯一标识
 * @ return 模块名称 */
public String getModuleName ( String remoteKey ) { } }

implementation: AppInfo appInfo = appNames . get ( remoteKey ) ; return appInfo == null ? null : appInfo . getAppName ( ) ;


signature:
public class TracingContext { /** * 判断是否有事务组
 * @ return result */
public boolean hasGroup ( ) { } }

implementation: return Objects . nonNull ( fields ) && fields . containsKey ( TracingConstants . GROUP_ID ) && StringUtils . hasText ( fields . get ( TracingConstants . GROUP_ID ) ) ;


signature:
public class TracingContext { /** * 添加App
 * @ param serviceId serviceId
 * @ param address address */
public void addApp ( String serviceId , String address ) { } }

implementation: if ( hasGroup ( ) ) { JSONObject map = JSON . parseObject ( this . fields . get ( TracingConstants . APP_MAP ) ) ; if ( map . containsKey ( serviceId ) ) { return ; } map . put ( serviceId , address ) ; this . fields . put ( TracingConstants . APP_MAP , JSON . toJSONString ( map ) ) ; return ; } raiseNonGroupException ( ) ;


signature:
public class TracingContext { /** * String Type App map .
 * @ return appMap */
public String appMapString ( ) { } }

implementation: if ( hasGroup ( ) ) { String appMap = Optional . ofNullable ( this . fields . get ( TracingConstants . APP_MAP ) ) . orElse ( "" ) ; log . debug ( "App map: {}" , appMap ) ; return appMap ; } raiseNonGroupException ( ) ; return "{}" ;


signature:
public class P6Util { /** * Locates a file on the file system or on the classpath .
 * Search order :
 * < ol >
 * < li > current working directory ( for relative filePath ) or any directory ( for absolute filePath ) < / li >
 * < li > class filePath < / li >
 * < / ol >
 * @ param file the relative filePath of the file to locate
 * @ return A URL to the file or null if not found */
public static URL locateFile ( String file ) { } }

implementation:
File fp ; URL result = null ; try { // try to find relative to current working directory first
fp = new File ( file ) ; if ( fp . exists ( ) ) { result = fp . toURI ( ) . toURL ( ) ; } // next try to load from context class loader
if ( result == null ) { result = locateOnClassPath ( file ) ; } } catch ( Exception e ) { } return result ;


signature:
public class P6Util { /** * Locates a file on the classpath .
 * @ param filename the relative filePath of the file to load
 * @ return the URL of the file or null if not found */
public static URL locateOnClassPath ( String filename ) { } }

implementation:
URL result ; // first try to load from context class loader
result = Thread . currentThread ( ) . getContextClassLoader ( ) . getResource ( filename ) ; // next try the current class loader which loaded p6spy
if ( result == null ) { result = P6Util . class . getClassLoader ( ) . getResource ( filename ) ; } // finally try the system class loader
if ( result == null ) { result = ClassLoader . getSystemResource ( filename ) ; } return result ;


signature:
public class P6Util { /** * A utility for using the current class loader ( rather than the
 * system class loader ) when instantiating a new class .
 * The idea is that the thread ' s current loader might have an
 * obscure notion of what your class filePath is ( e . g . an app server ) that
 * will not be captured properly by the system class loader .
 * taken from http : / / sourceforge . net / forum / message . php ? msg _ id = 1720229
 * @ param name class name to load
 * @ return the newly loaded class
 * @ throws ClassNotFoundException ClassNotFoundException */
public static Class < ? > forName ( String name ) throws ClassNotFoundException { } }

implementation:
ClassLoader ctxLoader = null ; try { ctxLoader = Thread . currentThread ( ) . getContextClassLoader ( ) ; return Class . forName ( name , true , ctxLoader ) ; } catch ( ClassNotFoundException ex ) { // try to fall through and use the default
// Class . forName
// if ( ctxLoader = = null ) { throw ex ; }
} catch ( SecurityException ex ) { } return Class . forName ( name ) ;


signature:
public class ConnectionInformation { /** * This method should only be used in test scenarios
 * @ param connection the underlying connection ( possibly a mock )
 * @ return a new { @ link ConnectionInformation } instance */
public static ConnectionInformation fromConnection ( Connection connection ) { } }

implementation: final ConnectionInformation connectionInformation = new ConnectionInformation ( ) ; connectionInformation . connection = connection ; return connectionInformation ;


signature:
public class DTXLocalContext { /** * 获取或新建一个线程变量 。
 * @ return 当前线程变量 */
public static DTXLocalContext getOrNew ( ) { } }

implementation: if ( currentLocal . get ( ) == null ) { currentLocal . set ( new DTXLocalContext ( ) ) ; } return currentLocal . get ( ) ;


signature:
public class DTXLocalContext { /** * 设置代理资源 */
public static void makeProxy ( ) { } }

implementation: if ( currentLocal . get ( ) != null ) { cur ( ) . proxyTmp = cur ( ) . proxy ; cur ( ) . proxy = true ; }


signature:
public class DTXLocalContext { /** * 设置不代理资源 */
public static void makeUnProxy ( ) { } }

implementation: if ( currentLocal . get ( ) != null ) { cur ( ) . proxyTmp = cur ( ) . proxy ; cur ( ) . proxy = false ; }


signature:
public class DTXLocalContext { /** * 清理线程变量 */
public static void makeNeverAppeared ( ) { } }

implementation: if ( currentLocal . get ( ) != null ) { log . debug ( "clean thread local[{}]: {}" , DTXLocalContext . class . getSimpleName ( ) , cur ( ) ) ; currentLocal . set ( null ) ; }


signature:
public class DTXLocalContext { /** * 事务状态
 * @ param userDtxState state
 * @ return 1 commit 0 rollback */
public static int transactionState ( int userDtxState ) { } }

implementation: DTXLocalContext dtxLocalContext = Objects . requireNonNull ( currentLocal . get ( ) , "DTX can't be null." ) ; return userDtxState == 1 ? dtxLocalContext . sysTransactionState : userDtxState ;


signature:
public class ExceptionToResourceMapping { /** * Looks at the exception and its causes trying to find an ID . */
public Integer mapThrowable ( final Throwable throwable ) { } }

implementation:
Throwable throwableToCheck = throwable ; int depthToGo = 20 ; while ( true ) { Integer resId = mapThrowableFlat ( throwableToCheck ) ; if ( resId != null ) { return resId ; } else { throwableToCheck = throwableToCheck . getCause ( ) ; depthToGo -- ; if ( depthToGo <= 0 || throwableToCheck == throwable || throwableToCheck == null ) { Log . d ( "EventBus" , "No specific message resource ID found for " + throwable ) ; // return config . defaultErrorMsgId ;
return null ; } } }


signature:
public class ExceptionToResourceMapping { /** * Mapping without checking the cause ( done in mapThrowable ) . */
protected Integer mapThrowableFlat ( Throwable throwable ) { } }

implementation: Class < ? extends Throwable > throwableClass = throwable . getClass ( ) ; Integer resId = throwableToMsgIdMap . get ( throwableClass ) ; if ( resId == null ) { Class < ? extends Throwable > closestClass = null ; Set < Entry < Class < ? extends Throwable > , Integer > > mappings = throwableToMsgIdMap . entrySet ( ) ; for ( Entry < Class < ? extends Throwable > , Integer > mapping : mappings ) { Class < ? extends Throwable > candidate = mapping . getKey ( ) ; if ( candidate . isAssignableFrom ( throwableClass ) ) { if ( closestClass == null || closestClass . isAssignableFrom ( candidate ) ) { closestClass = candidate ; resId = mapping . getValue ( ) ; } } } } return resId ;


signature:
public class EventBusAnnotationProcessor { /** * Subscriber classes should be skipped if their class or any involved event class are not visible to the index . */
private void checkForSubscribersToSkip ( Messager messager , String myPackage ) { } }

implementation: for ( TypeElement skipCandidate : methodsByClass . keySet ( ) ) { TypeElement subscriberClass = skipCandidate ; while ( subscriberClass != null ) { if ( ! isVisible ( myPackage , subscriberClass ) ) { boolean added = classesToSkip . add ( skipCandidate ) ; if ( added ) { String msg ; if ( subscriberClass . equals ( skipCandidate ) ) { msg = "Falling back to reflection because class is not public" ; } else { msg = "Falling back to reflection because " + skipCandidate + " has a non-public super class" ; } messager . printMessage ( Diagnostic . Kind . NOTE , msg , subscriberClass ) ; } break ; } List < ExecutableElement > methods = methodsByClass . get ( subscriberClass ) ; if ( methods != null ) { for ( ExecutableElement method : methods ) { String skipReason = null ; VariableElement param = method . getParameters ( ) . get ( 0 ) ; TypeMirror typeMirror = getParamTypeMirror ( param , messager ) ; if ( ! ( typeMirror instanceof DeclaredType ) || ! ( ( ( DeclaredType ) typeMirror ) . asElement ( ) instanceof TypeElement ) ) { skipReason = "event type cannot be processed" ; } if ( skipReason == null ) { TypeElement eventTypeElement = ( TypeElement ) ( ( DeclaredType ) typeMirror ) . asElement ( ) ; if ( ! isVisible ( myPackage , eventTypeElement ) ) { skipReason = "event type is not public" ; } } if ( skipReason != null ) { boolean added = classesToSkip . add ( skipCandidate ) ; if ( added ) { String msg = "Falling back to reflection because " + skipReason ; if ( ! subscriberClass . equals ( skipCandidate ) ) { msg += " (found in super class for " + skipCandidate + ")" ; } messager . printMessage ( Diagnostic . Kind . NOTE , msg , param ) ; } break ; } } } subscriberClass = getSuperclass ( subscriberClass ) ; } }


signature:
public class EventBus { /** * Convenience singleton for apps using a process - wide EventBus instance . */
public static EventBus getDefault ( ) { } }

implementation: EventBus instance = defaultInstance ; if ( instance == null ) { synchronized ( EventBus . class ) { instance = EventBus . defaultInstance ; if ( instance == null ) { instance = EventBus . defaultInstance = new EventBus ( ) ; } } } return instance ;


signature:
public class EventBus { /** * Registers the given subscriber to receive events . Subscribers must call { @ link # unregister ( Object ) } once they
 * are no longer interested in receiving events .
 * Subscribers have event handling methods that must be annotated by { @ link Subscribe } .
 * The { @ link Subscribe } annotation also allows configuration like { @ link
 * ThreadMode } and priority . */
public void register ( Object subscriber ) { } }

implementation: Class < ? > subscriberClass = subscriber . getClass ( ) ; List < SubscriberMethod > subscriberMethods = subscriberMethodFinder . findSubscriberMethods ( subscriberClass ) ; synchronized ( this ) { for ( SubscriberMethod subscriberMethod : subscriberMethods ) { subscribe ( subscriber , subscriberMethod ) ; } }


signature:
public class EventBus { /** * Must be called in synchronized block */
private void subscribe ( Object subscriber , SubscriberMethod subscriberMethod ) { } }

implementation:
Class < ? > eventType = subscriberMethod . eventType ; Subscription newSubscription = new Subscription ( subscriber , subscriberMethod ) ; CopyOnWriteArrayList < Subscription > subscriptions = subscriptionsByEventType . get ( eventType ) ; if ( subscriptions == null ) { subscriptions = new CopyOnWriteArrayList < > ( ) ; subscriptionsByEventType . put ( eventType , subscriptions ) ; } else { if ( subscriptions . contains ( newSubscription ) ) { throw new EventBusException ( "Subscriber " + subscriber . getClass ( ) + " already registered to event " + eventType ) ; } } int size = subscriptions . size ( ) ; for ( int i = 0 ; i <= size ; i ++ ) { if ( i == size || subscriberMethod . priority > subscriptions . get ( i ) . subscriberMethod . priority ) { subscriptions . add ( i , newSubscription ) ; break ; } } List < Class < ? > > subscribedEvents = typesBySubscriber . get ( subscriber ) ; if ( subscribedEvents == null ) { subscribedEvents = new ArrayList < > ( ) ; typesBySubscriber . put ( subscriber , subscribedEvents ) ; } subscribedEvents . add ( eventType ) ; if ( subscriberMethod . sticky ) { if ( eventInheritance ) { // Existing sticky events of all subclasses of eventType have to be considered .
// Note : Iterating over all events may be inefficient with lots of sticky events ,
// thus data structure should be changed to allow a more efficient lookup
// ( e . g . an additional map storing sub classes of super classes : Class - > List < Class > ) .
Set < Map . Entry < Class < ? > , Object > > entries = stickyEvents . entrySet ( ) ; for ( Map . Entry < Class < ? > , Object > entry : entries ) { Class < ? > candidateEventType = entry . getKey ( ) ; if ( eventType . isAssignableFrom ( candidateEventType ) ) { Object stickyEvent = entry . getValue ( ) ; checkPostStickyEventToSubscription ( newSubscription , stickyEvent ) ; } } } else { Object stickyEvent = stickyEvents . get ( eventType ) ; checkPostStickyEventToSubscription ( newSubscription , stickyEvent ) ; } }


signature:
public class EventBus { /** * Only updates subscriptionsByEventType , not typesBySubscriber ! Caller must update typesBySubscriber . */
private void unsubscribeByEventType ( Object subscriber , Class < ? > eventType ) { } }

implementation: List < Subscription > subscriptions = subscriptionsByEventType . get ( eventType ) ; if ( subscriptions != null ) { int size = subscriptions . size ( ) ; for ( int i = 0 ; i < size ; i ++ ) { Subscription subscription = subscriptions . get ( i ) ; if ( subscription . subscriber == subscriber ) { subscription . active = false ; subscriptions . remove ( i ) ; i -- ; size -- ; } } }


signature:
public class EventBus { /** * Unregisters the given subscriber from all event classes . */
public synchronized void unregister ( Object subscriber ) { } }

implementation: List < Class < ? > > subscribedTypes = typesBySubscriber . get ( subscriber ) ; if ( subscribedTypes != null ) { for ( Class < ? > eventType : subscribedTypes ) { unsubscribeByEventType ( subscriber , eventType ) ; } typesBySubscriber . remove ( subscriber ) ; } else { logger . log ( Level . WARNING , "Subscriber to unregister was not registered before: " + subscriber . getClass ( ) ) ; }


signature:
public class EventBus { /** * Posts the given event to the event bus . */
public void post ( Object event ) { } }

implementation: PostingThreadState postingState = currentPostingThreadState . get ( ) ; List < Object > eventQueue = postingState . eventQueue ; eventQueue . add ( event ) ; if ( ! postingState . isPosting ) { postingState . isMainThread = isMainThread ( ) ; postingState . isPosting = true ; if ( postingState . canceled ) { throw new EventBusException ( "Internal error. Abort state was not reset" ) ; } try { while ( ! eventQueue . isEmpty ( ) ) { postSingleEvent ( eventQueue . remove ( 0 ) , postingState ) ; } } finally { postingState . isPosting = false ; postingState . isMainThread = false ; } }


signature:
public class EventBus { /** * Called from a subscriber ' s event handling method , further event delivery will be canceled . Subsequent
 * subscribers
 * won ' t receive the event . Events are usually canceled by higher priority subscribers ( see
 * { @ link Subscribe # priority ( ) } ) . Canceling is restricted to event handling methods running in posting thread
 * { @ link ThreadMode # POSTING } . */
public void cancelEventDelivery ( Object event ) { } }

implementation: PostingThreadState postingState = currentPostingThreadState . get ( ) ; if ( ! postingState . isPosting ) { throw new EventBusException ( "This method may only be called from inside event handling methods on the posting thread" ) ; } else if ( event == null ) { throw new EventBusException ( "Event may not be null" ) ; } else if ( postingState . event != event ) { throw new EventBusException ( "Only the currently handled event may be aborted" ) ; } else if ( postingState . subscription . subscriberMethod . threadMode != ThreadMode . POSTING ) { throw new EventBusException ( " event handlers may only abort the incoming event" ) ; } postingState . canceled = true ;


signature:
public class EventBus { /** * Posts the given event to the event bus and holds on to the event ( because it is sticky ) . The most recent sticky
 * event of an event ' s type is kept in memory for future access by subscribers using { @ link Subscribe # sticky ( ) } . */
public void postSticky ( Object event ) { } }

implementation:
synchronized ( stickyEvents ) { stickyEvents . put ( event . getClass ( ) , event ) ; } // Should be posted after it is putted , in case the subscriber wants to remove immediately
post ( event ) ;


signature:
public class EventBus { /** * Gets the most recent sticky event for the given type .
 * @ see # postSticky ( Object ) */
public < T > T getStickyEvent ( Class < T > eventType ) { } }

implementation: synchronized ( stickyEvents ) { return eventType . cast ( stickyEvents . get ( eventType ) ) ; }


signature:
public class EventBus { /** * Remove and gets the recent sticky event for the given event type .
 * @ see # postSticky ( Object ) */
public < T > T removeStickyEvent ( Class < T > eventType ) { } }

implementation: synchronized ( stickyEvents ) { return eventType . cast ( stickyEvents . remove ( eventType ) ) ; }


signature:
public class EventBus { /** * Removes the sticky event if it equals to the given event .
 * @ return true if the events matched and the sticky event was removed . */
public boolean removeStickyEvent ( Object event ) { } }

implementation: synchronized ( stickyEvents ) { Class < ? > eventType = event . getClass ( ) ; Object existingEvent = stickyEvents . get ( eventType ) ; if ( event . equals ( existingEvent ) ) { stickyEvents . remove ( eventType ) ; return true ; } else { return false ; } }


signature:
public class EventBus { /** * Looks up all Class objects including super classes and interfaces . Should also work for interfaces . */
private static List < Class < ? > > lookupAllEventTypes ( Class < ? > eventClass ) { } }

implementation: synchronized ( eventTypesCache ) { List < Class < ? > > eventTypes = eventTypesCache . get ( eventClass ) ; if ( eventTypes == null ) { eventTypes = new ArrayList < > ( ) ; Class < ? > clazz = eventClass ; while ( clazz != null ) { eventTypes . add ( clazz ) ; addInterfaces ( eventTypes , clazz . getInterfaces ( ) ) ; clazz = clazz . getSuperclass ( ) ; } eventTypesCache . put ( eventClass , eventTypes ) ; } return eventTypes ; }


signature:
public class EventBus { /** * Recurses through super interfaces . */
static void addInterfaces ( List < Class < ? > > eventTypes , Class < ? > [ ] interfaces ) { } }

implementation: for ( Class < ? > interfaceClass : interfaces ) { if ( ! eventTypes . contains ( interfaceClass ) ) { eventTypes . add ( interfaceClass ) ; addInterfaces ( eventTypes , interfaceClass . getInterfaces ( ) ) ; } }


signature:
public class EventBus { /** * Invokes the subscriber if the subscriptions is still active . Skipping subscriptions prevents race conditions
 * between { @ link # unregister ( Object ) } and event delivery . Otherwise the event might be delivered after the
 * subscriber unregistered . This is particularly important for main thread delivery and registrations bound to the
 * live cycle of an Activity or Fragment . */
void invokeSubscriber ( PendingPost pendingPost ) { } }

implementation: Object event = pendingPost . event ; Subscription subscription = pendingPost . subscription ; PendingPost . releasePendingPost ( pendingPost ) ; if ( subscription . active ) { invokeSubscriber ( subscription , event ) ; }


signature:
public class AsyncExecutor { /** * Posts an failure event if the given { @ link RunnableEx } throws an Exception . */
public void execute ( final RunnableEx runnable ) { } }

implementation: threadPool . execute ( new Runnable ( ) { @ Override public void run ( ) { try { runnable . run ( ) ; } catch ( Exception e ) { Object event ; try { event = failureEventConstructor . newInstance ( e ) ; } catch ( Exception e1 ) { eventBus . getLogger ( ) . log ( Level . SEVERE , "Original exception:" , e ) ; throw new RuntimeException ( "Could not create failure event" , e1 ) ; } if ( event instanceof HasExecutionScope ) { ( ( HasExecutionScope ) event ) . setExecutionScope ( scope ) ; } eventBus . post ( event ) ; } } } ) ;


signature:
public class EventBusBuilder { /** * Method name verification is done for methods starting with onEvent to avoid typos ; using this method you can
 * exclude subscriber classes from this check . Also disables checks for method modifiers ( public , not static nor
 * abstract ) . */
public EventBusBuilder skipMethodVerificationFor ( Class < ? > clazz ) { } }

implementation: if ( skipMethodVerificationForClasses == null ) { skipMethodVerificationForClasses = new ArrayList < > ( ) ; } skipMethodVerificationForClasses . add ( clazz ) ; return this ;


signature:
public class EventBusBuilder { /** * Adds an index generated by EventBus ' annotation preprocessor . */
public EventBusBuilder addIndex ( SubscriberInfoIndex index ) { } }

implementation: if ( subscriberInfoIndexes == null ) { subscriberInfoIndexes = new ArrayList < > ( ) ; } subscriberInfoIndexes . add ( index ) ; return this ;


signature:
public class EventBusBuilder { /** * Installs the default EventBus returned by { @ link EventBus # getDefault ( ) } using this builders ' values . Must be
 * done only once before the first usage of the default EventBus .
 * @ throws EventBusException if there ' s already a default EventBus instance in place */
public EventBus installDefaultEventBus ( ) { } }

implementation: synchronized ( EventBus . class ) { if ( EventBus . defaultInstance != null ) { throw new EventBusException ( "Default instance already exists." + " It may be only set once before it's used the first time to ensure consistent behavior." ) ; } EventBus . defaultInstance = build ( ) ; return EventBus . defaultInstance ; }


signature:
public class ErrorDialogManager { /** * Scope is limited to the activity ' s class . */
public static void attachTo ( Activity activity , boolean finishAfterDialog , Bundle argumentsForErrorDialog ) { } }

implementation: Object executionScope = activity . getClass ( ) ; attachTo ( activity , executionScope , finishAfterDialog , argumentsForErrorDialog ) ;


signature:
public class ErrorDialogFragmentFactory { /** * Prepares the fragment ' s arguments and creates the fragment . May be overridden to provide custom error fragments . */
protected T prepareErrorFragment ( ThrowableFailureEvent event , boolean finishAfterDialog , Bundle argumentsForErrorDialog ) { } }

implementation:
if ( event . isSuppressErrorUi ( ) ) { // Show nothing by default
return null ; } Bundle bundle ; if ( argumentsForErrorDialog != null ) { bundle = ( Bundle ) argumentsForErrorDialog . clone ( ) ; } else { bundle = new Bundle ( ) ; } if ( ! bundle . containsKey ( ErrorDialogManager . KEY_TITLE ) ) { String title = getTitleFor ( event , bundle ) ; bundle . putString ( ErrorDialogManager . KEY_TITLE , title ) ; } if ( ! bundle . containsKey ( ErrorDialogManager . KEY_MESSAGE ) ) { String message = getMessageFor ( event , bundle ) ; bundle . putString ( ErrorDialogManager . KEY_MESSAGE , message ) ; } if ( ! bundle . containsKey ( ErrorDialogManager . KEY_FINISH_AFTER_DIALOG ) ) { bundle . putBoolean ( ErrorDialogManager . KEY_FINISH_AFTER_DIALOG , finishAfterDialog ) ; } if ( ! bundle . containsKey ( ErrorDialogManager . KEY_EVENT_TYPE_ON_CLOSE ) && config . defaultEventTypeOnDialogClosed != null ) { bundle . putSerializable ( ErrorDialogManager . KEY_EVENT_TYPE_ON_CLOSE , config . defaultEventTypeOnDialogClosed ) ; } if ( ! bundle . containsKey ( ErrorDialogManager . KEY_ICON_ID ) && config . defaultDialogIconId != 0 ) { bundle . putInt ( ErrorDialogManager . KEY_ICON_ID , config . defaultDialogIconId ) ; } return createErrorFragment ( event , bundle ) ;


signature:
public class ErrorDialogFragmentFactory { /** * May be overridden to provide custom error title . */
protected String getTitleFor ( ThrowableFailureEvent event , Bundle arguments ) { } }

implementation: return config . resources . getString ( config . defaultTitleId ) ;


signature:
public class ErrorDialogFragmentFactory { /** * May be overridden to provide custom error messages . */
protected String getMessageFor ( ThrowableFailureEvent event , Bundle arguments ) { } }

implementation: int msgResId = config . getMessageIdForThrowable ( event . throwable ) ; return config . resources . getString ( msgResId ) ;


signature:
public class FluxProcessor { /** * Build a { @ link FluxProcessor } whose data are emitted by the most recent emitted { @ link Publisher } .
 * The { @ link Flux } will complete once both the publishers source and the last switched to { @ link Publisher } have
 * completed .
 * < img class = " marble " src = " https : / / raw . githubusercontent . com / reactor / reactor - core / v3.1.3 . RELEASE / src / docs / marble / switchonnext . png " alt = " " >
 * @ param < T > the produced type
 * @ return a { @ link FluxProcessor } accepting publishers and producing T */
public static < T > FluxProcessor < Publisher < ? extends T > , T > switchOnNext ( ) { } }

implementation: UnicastProcessor < Publisher < ? extends T > > emitter = UnicastProcessor . create ( ) ; FluxProcessor < Publisher < ? extends T > , T > p = FluxProcessor . wrap ( emitter , switchOnNext ( emitter ) ) ; return p ;


signature:
public class FluxProcessor { /** * Transform a receiving { @ link Subscriber } and a producing { @ link Publisher } in a logical { @ link FluxProcessor } .
 * The link between the passed upstream and returned downstream will not be created automatically , e . g . not
 * subscribed together . A { @ link Processor } might choose to have orthogonal sequence input and output .
 * @ param < IN > the receiving type
 * @ param < OUT > the producing type
 * @ param upstream the upstream subscriber
 * @ param downstream the downstream publisher
 * @ return a new blackboxed { @ link FluxProcessor } */
public static < IN , OUT > FluxProcessor < IN , OUT > wrap ( final Subscriber < IN > upstream , final Publisher < OUT > downstream ) { } }

implementation: return new DelegateProcessor < > ( downstream , upstream ) ;


signature:
public class FluxProcessor { /** * Create a { @ link FluxSink } that safely gates multi - threaded producer
 * { @ link Subscriber # onNext ( Object ) } . This processor will be subscribed to
 * that { @ link FluxSink } , and any previous subscribers will be unsubscribed .
 * < p > The returned { @ link FluxSink } will not apply any
 * { @ link FluxSink . OverflowStrategy } and overflowing { @ link FluxSink # next ( Object ) }
 * will behave in two possible ways depending on the Processor :
 * < ul >
 * < li > an unbounded processor will handle the overflow itself by dropping or
 * buffering < / li >
 * < li > a bounded processor will block / spin on IGNORE strategy , or apply the
 * strategy behavior < / li >
 * < / ul >
 * @ param strategy the overflow strategy , see { @ link FluxSink . OverflowStrategy }
 * for the
 * available strategies
 * @ return a serializing { @ link FluxSink } */
public final FluxSink < IN > sink ( FluxSink . OverflowStrategy strategy ) { } }

implementation: Objects . requireNonNull ( strategy , "strategy" ) ; if ( getBufferSize ( ) == Integer . MAX_VALUE ) { strategy = FluxSink . OverflowStrategy . IGNORE ; } FluxCreate . BaseSink < IN > s = FluxCreate . createSink ( this , strategy ) ; onSubscribe ( s ) ; if ( s . isCancelled ( ) || ( isSerialized ( ) && getBufferSize ( ) == Integer . MAX_VALUE ) ) { return s ; } if ( serializeAlways ( ) ) return new FluxCreate . SerializedSink < > ( s ) ; else return new FluxCreate . SerializeOnRequestSink < > ( s ) ;


signature:
public class OperatorDisposables { /** * Atomically set the field to the given non - null { @ link Disposable } and return true ,
 * or return false if the field is non - null .
 * If the target field contains the common { @ link # DISPOSED } instance , the supplied disposable
 * is disposed . If the field contains other non - null { @ link Disposable } , an { @ link IllegalStateException }
 * is signalled to the { @ code errorCallback } .
 * @ param updater the target field updater
 * @ param holder the target instance holding the field
 * @ param newValue the new Disposable to set , not null
 * @ return true if the operation succeeded , false */
public static < T > boolean setOnce ( AtomicReferenceFieldUpdater < T , Disposable > updater , T holder , Disposable newValue , Consumer < RuntimeException > errorCallback ) { } }

implementation: Objects . requireNonNull ( newValue , "newValue is null" ) ; if ( ! updater . compareAndSet ( holder , null , newValue ) ) { newValue . dispose ( ) ; if ( updater . get ( holder ) != DISPOSED ) { errorCallback . accept ( new IllegalStateException ( "Disposable already pushed" ) ) ; } return false ; } return true ;


signature:
public class OperatorDisposables { /** * Verify that current is null and next is not null , otherwise signal a
 * { @ link NullPointerException } to the { @ code errorCallback } and return false .
 * @ param current the current { @ link Disposable } , expected to be null
 * @ param next the next { @ link Disposable } , expected to be non - null
 * @ return true if the validation succeeded */
public static boolean validate ( @ Nullable Disposable current , Disposable next , Consumer < RuntimeException > errorCallback ) { } }

implementation:
// noinspection ConstantConditions
if ( next == null ) { errorCallback . accept ( new NullPointerException ( "next is null" ) ) ; return false ; } if ( current != null ) { next . dispose ( ) ; errorCallback . accept ( new IllegalStateException ( "Disposable already pushed" ) ) ; return false ; } return true ;


signature:
public class OperatorDisposables { /** * Atomically try to set the given { @ link Disposable } on the field if it is null or
 * disposes it if the field contains { @ link # DISPOSED } .
 * @ param updater the target field updater
 * @ param holder the target instance holding the field
 * @ param newValue the disposable to set
 * @ return true if successful , false otherwise */
public static < T > boolean trySet ( AtomicReferenceFieldUpdater < T , Disposable > updater , T holder , Disposable newValue ) { } }

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


signature:
public class FluxIterable { /** * Common method to take an Iterator as a source of values .
 * @ param s the subscriber to feed this iterator to
 * @ param it the iterator to use as a source of values */
@ SuppressWarnings ( "unchecked" ) static < T > void subscribe ( CoreSubscriber < ? super T > s , Iterator < ? extends T > it ) { } }

implementation: subscribe ( s , it , null ) ;


signature:
public class FluxIterable { /** * Common method to take an Iterator as a source of values .
 * @ param s the subscriber to feed this iterator to
 * @ param it the iterator to use as a source of values
 * @ param onClose close handler to call once we ' re done with the iterator ( provided it
 * is not null , this includes when the iteration errors or complete or the subscriber
 * is cancelled ) . Null to ignore . */
@ SuppressWarnings ( "unchecked" ) static < T > void subscribe ( CoreSubscriber < ? super T > s , Iterator < ? extends T > it , @ Nullable Runnable onClose ) { } }

implementation:
// noinspection ConstantConditions
if ( it == null ) { Operators . error ( s , new NullPointerException ( "The iterator is null" ) ) ; return ; } boolean b ; try { b = it . hasNext ( ) ; } catch ( Throwable e ) { Operators . error ( s , Operators . onOperatorError ( e , s . currentContext ( ) ) ) ; if ( onClose != null ) { try { onClose . run ( ) ; } catch ( Throwable t ) { Operators . onErrorDropped ( t , s . currentContext ( ) ) ; } } return ; } if ( ! b ) { Operators . complete ( s ) ; if ( onClose != null ) { try { onClose . run ( ) ; } catch ( Throwable t ) { Operators . onErrorDropped ( t , s . currentContext ( ) ) ; } } return ; } if ( s instanceof ConditionalSubscriber ) { s . onSubscribe ( new IterableSubscriptionConditional < > ( ( ConditionalSubscriber < ? super T > ) s , it , onClose ) ) ; } else { s . onSubscribe ( new IterableSubscription < > ( s , it , onClose ) ) ; }


signature:
public class TopicProcessor { /** * Create a new TopicProcessor using the provided backlog size , with a blockingWait Strategy
 * and auto - cancellation . < p > A new Cached ThreadExecutorPool will be implicitly created and will use the passed name to
 * qualify the created threads .
 * @ param name Use a new Cached ExecutorService and assign this name to the created
 * threads
 * @ param bufferSize A Backlog Size to mitigate slow subscribers
 * @ param < E > Type of processed signals
 * @ return the fresh TopicProcessor instance */
public static < E > TopicProcessor < E > create ( String name , int bufferSize ) { } }

implementation: return TopicProcessor . < E > builder ( ) . name ( name ) . bufferSize ( bufferSize ) . build ( ) ;


signature:
public class TopicProcessor { /** * Create a new shared TopicProcessor using the passed backlog size , with a blockingWait
 * Strategy and auto - cancellation .
 * A Shared Processor authorizes concurrent onNext calls and is suited for multi - threaded
 * publisher that will fan - in data .
 * A new Cached ThreadExecutorPool will be implicitly created and will use the passed
 * name to qualify the created threads .
 * @ param name Use a new Cached ExecutorService and assign this name to the created
 * threads
 * @ param bufferSize A Backlog Size to mitigate slow subscribers
 * @ param < E > Type of processed signals
 * @ return a fresh processor */
public static < E > TopicProcessor < E > share ( String name , int bufferSize ) { } }

implementation: return TopicProcessor . < E > builder ( ) . share ( true ) . name ( name ) . bufferSize ( bufferSize ) . build ( ) ;


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

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


signature:
public class FluxMergeOrdered { /** * Returns a new instance which has the additional source to be merged together with
 * the current array of sources . The provided { @ link Comparator } is tested for equality
 * with the current comparator , and if different is combined by { @ link Comparator # thenComparing ( Comparator ) } .
 * This operation doesn ' t change the current { @ link FluxMergeOrdered } instance .
 * @ param source the new source to merge with the others
 * @ return the new { @ link FluxMergeOrdered } instance */
FluxMergeOrdered < T > mergeAdditionalSource ( Publisher < ? extends T > source , Comparator < ? super T > otherComparator ) { } }

implementation: int n = sources . length ; @ SuppressWarnings ( "unchecked" ) Publisher < ? extends T > [ ] newArray = new Publisher [ n + 1 ] ; System . arraycopy ( sources , 0 , newArray , 0 , n ) ; newArray [ n ] = source ; if ( ! valueComparator . equals ( otherComparator ) ) { @ SuppressWarnings ( "unchecked" ) Comparator < T > currentComparator = ( Comparator < T > ) this . valueComparator ; final Comparator < T > newComparator = currentComparator . thenComparing ( otherComparator ) ; return new FluxMergeOrdered < > ( prefetch , queueSupplier , newComparator , newArray ) ; } return new FluxMergeOrdered < > ( prefetch , queueSupplier , valueComparator , newArray ) ;


signature:
public class MpscLinkedQueue { /** * { @ inheritDoc } < br >
 * IMPLEMENTATION NOTES : < br >
 * Offer is allowed from multiple threads . < br >
 * Offer allocates a new node and :
 * < ol >
 * < li > Swaps it atomically with current producer node ( only one producer ' wins ' )
 * < li > Sets the new node as the node following from the swapped producer node
 * < / ol >
 * This works because each producer is guaranteed to ' plant ' a new node and link the old node . No 2
 * producers can get the same producer node as part of XCHG guarantee .
 * @ see java . util . Queue # offer ( java . lang . Object ) */
@ Override @ SuppressWarnings ( "unchecked" ) public final boolean offer ( final E e ) { } }

implementation:
Objects . requireNonNull ( e , "The offered value 'e' must be non-null" ) ; final LinkedQueueNode < E > nextNode = new LinkedQueueNode < > ( e ) ; final LinkedQueueNode < E > prevProducerNode = PRODUCER_NODE_UPDATER . getAndSet ( this , nextNode ) ; // Should a producer thread get interrupted here the chain WILL be broken until that thread is resumed
// and completes the store in prev . next .
prevProducerNode . soNext ( nextNode ) ; // StoreStore
return true ;


signature:
public class MpscLinkedQueue { /** * { @ inheritDoc } < br >
 * IMPLEMENTATION NOTES : < br >
 * Poll is allowed from a SINGLE thread . < br >
 * Poll reads the next node from the consumerNode and :
 * < ol >
 * < li > If it is null , the queue is assumed empty ( though it might not be ) .
 * < li > If it is not null set it as the consumer node and return it ' s now evacuated value .
 * < / ol >
 * This means the consumerNode . value is always null , which is also the starting point for the queue .
 * Because null values are not allowed to be offered this is the only node with it ' s value set to null at
 * any one time .
 * @ see java . util . Queue # poll ( ) */
@ Nullable @ Override public E poll ( ) { } }

implementation:
LinkedQueueNode < E > currConsumerNode = consumerNode ; // don ' t load twice , it ' s alright
LinkedQueueNode < E > nextNode = currConsumerNode . lvNext ( ) ; if ( nextNode != null ) { // we have to null out the value because we are going to hang on to the node
final E nextValue = nextNode . getAndNullValue ( ) ; // Fix up the next ref of currConsumerNode to prevent promoted nodes from keeping new ones alive .
// We use a reference to self instead of null because null is already a meaningful value ( the next of
// producer node is null ) .
currConsumerNode . soNext ( currConsumerNode ) ; CONSUMER_NODE_UPDATER . lazySet ( this , nextNode ) ; // currConsumerNode is now no longer referenced and can be collected
return nextValue ; } else if ( currConsumerNode != producerNode ) { while ( ( nextNode = currConsumerNode . lvNext ( ) ) == null ) { } // got the next node . . .
// we have to null out the value because we are going to hang on to the node
final E nextValue = nextNode . getAndNullValue ( ) ; // Fix up the next ref of currConsumerNode to prevent promoted nodes from keeping new ones alive .
// We use a reference to self instead of null because null is already a meaningful value ( the next of
// producer node is null ) .
currConsumerNode . soNext ( currConsumerNode ) ; CONSUMER_NODE_UPDATER . lazySet ( this , nextNode ) ; // currConsumerNode is now no longer referenced and can be collected
return nextValue ; } return null ;


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

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


signature:
public class WorkQueueProcessor { /** * Create a new WorkQueueProcessor using the passed buffer size , blockingWait
 * Strategy and auto - cancel . < p > A new Cached ThreadExecutorPool will be implicitly
 * created and will use the passed name to qualify the created threads .
 * @ param name Use a new Cached ExecutorService and assign this name to the created
 * threads
 * @ param bufferSize A Backlog Size to mitigate slow subscribers
 * @ param < E > Type of processed signals
 * @ return a fresh processor */
public static < E > WorkQueueProcessor < E > create ( String name , int bufferSize ) { } }

implementation: return WorkQueueProcessor . < E > builder ( ) . name ( name ) . bufferSize ( bufferSize ) . build ( ) ;


signature:
public class WorkQueueProcessor { /** * This method will attempt to compute the maximum amount of subscribers a
 * { @ link WorkQueueProcessor } can accommodate based on a given { @ link ExecutorService } .
 * It can only accurately detect this for { @ link ThreadPoolExecutor } and
 * { @ link ForkJoinPool } instances , and will return { @ link Integer # MIN _ VALUE } for other
 * executor implementations .
 * @ param executor the executor to attempt to introspect .
 * @ return the maximum number of subscribers the executor can accommodate if it can
 * be computed , or { @ link Integer # MIN _ VALUE } if it cannot be determined . */
static int bestEffortMaxSubscribers ( ExecutorService executor ) { } }

implementation: int maxSubscribers = Integer . MIN_VALUE ; if ( executor instanceof ThreadPoolExecutor ) { maxSubscribers = ( ( ThreadPoolExecutor ) executor ) . getMaximumPoolSize ( ) ; } else if ( executor instanceof ForkJoinPool ) { maxSubscribers = ( ( ForkJoinPool ) executor ) . getParallelism ( ) ; } return maxSubscribers ;


signature:
public class DrainUtils { /** * Perform a potential post - completion request accounting .
 * @ param < T > the output value type
 * @ param < F > the field type holding the requested amount
 * @ param n the requested amount
 * @ param actual the consumer of values
 * @ param queue the queue holding the available values
 * @ param field the field updater for the requested amount
 * @ param instance the parent instance for the requested field
 * @ param isCancelled callback to detect cancellation
 * @ return true if the state indicates a completion state . */
static < T , F > boolean postCompleteRequest ( long n , Subscriber < ? super T > actual , Queue < T > queue , AtomicLongFieldUpdater < F > field , F instance , BooleanSupplier isCancelled ) { } }

implementation:
for ( ; ; ) { long r = field . get ( instance ) ; // extract the current request amount
long r0 = r & REQUESTED_MASK ; // preserve COMPLETED _ MASK and calculate new requested amount
long u = ( r & COMPLETED_MASK ) | Operators . addCap ( r0 , n ) ; if ( field . compareAndSet ( instance , r , u ) ) { // ( complete , 0 ) - > ( complete , n ) transition then replay
if ( r == COMPLETED_MASK ) { postCompleteDrain ( n | COMPLETED_MASK , actual , queue , field , instance , isCancelled ) ; return true ; } // ( active , r ) - > ( active , r + n ) transition then continue with requesting from upstream
return false ; } }


signature:
public class DrainUtils { /** * Drains the queue either in a pre - or post - complete state .
 * @ param n the requested amount
 * @ param actual the consumer of values
 * @ param queue the queue holding available values
 * @ param field the field updater holding the requested amount
 * @ param instance the parent instance of the requested field
 * @ param isCancelled callback to detect cancellation
 * @ return true if the queue was completely drained or the drain process was cancelled */
static < T , F > boolean postCompleteDrain ( long n , Subscriber < ? super T > actual , Queue < T > queue , AtomicLongFieldUpdater < F > field , F instance , BooleanSupplier isCancelled ) { } }

implementation:
// TODO enable fast - path
// if ( n = = - 1 | | n = = Long . MAX _ VALUE ) {
// for ( ; ; ) {
// if ( isDisposed . getAsBoolean ( ) ) {
// break ;
// T v = queue . poll ( ) ;
// if ( v = = null ) {
// actual . onComplete ( ) ;
// break ;
// actual . onNext ( v ) ;
// return true ;
long e = n & COMPLETED_MASK ; for ( ; ; ) { while ( e != n ) { if ( isCancelled . getAsBoolean ( ) ) { return true ; } T t = queue . poll ( ) ; if ( t == null ) { actual . onComplete ( ) ; return true ; } actual . onNext ( t ) ; e ++ ; } if ( isCancelled . getAsBoolean ( ) ) { return true ; } if ( queue . isEmpty ( ) ) { actual . onComplete ( ) ; return true ; } n = field . get ( instance ) ; if ( n == e ) { n = field . addAndGet ( instance , - ( e & REQUESTED_MASK ) ) ; if ( ( n & REQUESTED_MASK ) == 0L ) { return false ; } e = n & COMPLETED_MASK ; } }


signature:
public class DrainUtils { /** * Tries draining the queue if the source just completed .
 * @ param < T > the output value type
 * @ param < F > the field type holding the requested amount
 * @ param actual the consumer of values
 * @ param queue the queue holding available values
 * @ param field the field updater holding the requested amount
 * @ param instance the parent instance of the requested field
 * @ param isCancelled callback to detect cancellation */
public static < T , F > void postComplete ( CoreSubscriber < ? super T > actual , Queue < T > queue , AtomicLongFieldUpdater < F > field , F instance , BooleanSupplier isCancelled ) { } }

implementation:
if ( queue . isEmpty ( ) ) { actual . onComplete ( ) ; return ; } if ( postCompleteDrain ( field . get ( instance ) , actual , queue , field , instance , isCancelled ) ) { return ; } for ( ; ; ) { long r = field . get ( instance ) ; if ( ( r & COMPLETED_MASK ) != 0L ) { return ; } long u = r | COMPLETED_MASK ; // ( active , r ) - > ( complete , r ) transition
if ( field . compareAndSet ( instance , r , u ) ) { // if the requested amount was non - zero , drain the queue
if ( r != 0L ) { postCompleteDrain ( u , actual , queue , field , instance , isCancelled ) ; } return ; } }


signature:
public class DrainUtils { /** * Perform a potential post - completion request accounting .
 * @ param < T > the output value type
 * @ param < F > the field type holding the requested amount
 * @ param n the request amount
 * @ param actual the consumer of values
 * @ param queue the queue of available values
 * @ param field the field updater for the requested amount
 * @ param instance the parent instance of the requested field
 * @ param isCancelled callback to detect cancellation
 * @ param error if not null , the error to signal after the queue has been drained
 * @ return true if the state indicates a completion state . */
public static < T , F > boolean postCompleteRequestDelayError ( long n , Subscriber < ? super T > actual , Queue < T > queue , AtomicLongFieldUpdater < F > field , F instance , BooleanSupplier isCancelled , Throwable error ) { } }

implementation:
for ( ; ; ) { long r = field . get ( instance ) ; // extract the current request amount
long r0 = r & REQUESTED_MASK ; // preserve COMPLETED _ MASK and calculate new requested amount
long u = ( r & COMPLETED_MASK ) | Operators . addCap ( r0 , n ) ; if ( field . compareAndSet ( instance , r , u ) ) { // ( complete , 0 ) - > ( complete , n ) transition then replay
if ( r == COMPLETED_MASK ) { postCompleteDrainDelayError ( n | COMPLETED_MASK , actual , queue , field , instance , isCancelled , error ) ; return true ; } // ( active , r ) - > ( active , r + n ) transition then continue with requesting from upstream
return false ; } }


signature:
public class DrainUtils { /** * Tries draining the queue if the source just completed .
 * @ param < T > the output value type
 * @ param < F > the field type holding the requested amount
 * @ param actual the consumer of values
 * @ param queue the queue of available values
 * @ param field the field updater for the requested amount
 * @ param instance the parent instance of the requested field
 * @ param isCancelled callback to detect cancellation
 * @ param error if not null , the error to signal after the queue has been drained */
public static < T , F > void postCompleteDelayError ( CoreSubscriber < ? super T > actual , Queue < T > queue , AtomicLongFieldUpdater < F > field , F instance , BooleanSupplier isCancelled , @ Nullable Throwable error ) { } }

implementation:
if ( queue . isEmpty ( ) ) { if ( error == null ) { actual . onComplete ( ) ; } else { actual . onError ( error ) ; } return ; } if ( postCompleteDrainDelayError ( field . get ( instance ) , actual , queue , field , instance , isCancelled , error ) ) { return ; } for ( ; ; ) { long r = field . get ( instance ) ; if ( ( r & COMPLETED_MASK ) != 0L ) { return ; } long u = r | COMPLETED_MASK ; // ( active , r ) - > ( complete , r ) transition
if ( field . compareAndSet ( instance , r , u ) ) { // if the requested amount was non - zero , drain the queue
if ( r != 0L ) { postCompleteDrainDelayError ( u , actual , queue , field , instance , isCancelled , error ) ; } return ; } }


signature:
public class BlockingSingleSubscriber { /** * Block until the first value arrives and return it , otherwise
 * return null for an empty source and rethrow any exception .
 * @ param timeout the timeout to wait
 * @ param unit the time unit
 * @ return the first value or null if the source is empty */
@ Nullable final T blockingGet ( long timeout , TimeUnit unit ) { } }

implementation:
if ( Schedulers . isInNonBlockingThread ( ) ) { throw new IllegalStateException ( "block()/blockFirst()/blockLast() are blocking, which is not supported in thread " + Thread . currentThread ( ) . getName ( ) ) ; } if ( getCount ( ) != 0 ) { try { if ( ! await ( timeout , unit ) ) { dispose ( ) ; throw new IllegalStateException ( "Timeout on blocking read for " + timeout + " " + unit ) ; } } catch ( InterruptedException ex ) { dispose ( ) ; RuntimeException re = Exceptions . propagate ( ex ) ; // this is ok , as re is always a new non - singleton instance
re . addSuppressed ( new Exception ( "#block has been interrupted" ) ) ; throw re ; } } Throwable e = error ; if ( e != null ) { RuntimeException re = Exceptions . propagate ( e ) ; // this is ok , as re is always a new non - singleton instance
re . addSuppressed ( new Exception ( "#block terminated with an error" ) ) ; throw re ; } return value ;


signature:
public class Flux { /** * Concatenate all sources provided in an { @ link Iterable } , forwarding elements
 * emitted by the sources downstream .
 * Concatenation is achieved by sequentially subscribing to the first source then
 * waiting for it to complete before subscribing to the next , and so on until the
 * last source completes . Any error interrupts the sequence immediately and is
 * forwarded downstream .
 * < img class = " marble " src = " doc - files / marbles / concatVarSources . svg " alt = " " >
 * @ param sources The { @ link Iterable } of { @ link Publisher } to concatenate
 * @ param < T > The type of values in both source and output sequences
 * @ return a new { @ link Flux } concatenating all source sequences */
public static < T > Flux < T > concat ( Iterable < ? extends Publisher < ? extends T > > sources ) { } }

implementation: return onAssembly ( new FluxConcatIterable < > ( sources ) ) ;


signature:
public class Flux { /** * Concatenates the values to the end of the { @ link Flux }
 * < img class = " marble " src = " doc - files / marbles / concatWithValues . svg " alt = " " >
 * @ param values The values to concatenate
 * @ return a new { @ link Flux } concatenating all source sequences */
@ SafeVarargs public final Flux < T > concatWithValues ( T ... values ) { } }

implementation: return concatWith ( Flux . fromArray ( values ) ) ;


signature:
public class Flux { /** * Concatenate all sources emitted as an onNext signal from a parent { @ link Publisher } ,
 * forwarding elements emitted by the sources downstream .
 * Concatenation is achieved by sequentially subscribing to the first source then
 * waiting for it to complete before subscribing to the next , and so on until the
 * last source completes . Any error interrupts the sequence immediately and is
 * forwarded downstream .
 * < img class = " marble " src = " doc - files / marbles / concatAsyncSources . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure upon cancellation .
 * @ param sources The { @ link Publisher } of { @ link Publisher } to concatenate
 * @ param < T > The type of values in both source and output sequences
 * @ return a new { @ link Flux } concatenating all inner sources sequences */
public static < T > Flux < T > concat ( Publisher < ? extends Publisher < ? extends T > > sources ) { } }

implementation: return concat ( sources , Queues . XS_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Concatenate all sources emitted as an onNext signal from a parent { @ link Publisher } ,
 * forwarding elements emitted by the sources downstream .
 * Concatenation is achieved by sequentially subscribing to the first source then
 * waiting for it to complete before subscribing to the next , and so on until the
 * last source completes . Any error interrupts the sequence immediately and is
 * forwarded downstream .
 * < img class = " marble " src = " doc - files / marbles / concatAsyncSources . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure upon cancellation .
 * @ param sources The { @ link Publisher } of { @ link Publisher } to concatenate
 * @ param prefetch the inner source request size
 * @ param < T > The type of values in both source and output sequences
 * @ return a new { @ link Flux } concatenating all inner sources sequences */
public static < T > Flux < T > concat ( Publisher < ? extends Publisher < ? extends T > > sources , int prefetch ) { } }

implementation: return onAssembly ( new FluxConcatMap < > ( from ( sources ) , identityFunction ( ) , Queues . get ( prefetch ) , prefetch , FluxConcatMap . ErrorMode . IMMEDIATE ) ) ;


signature:
public class Flux { /** * Concatenate all sources provided as a vararg , forwarding elements emitted by the
 * sources downstream .
 * Concatenation is achieved by sequentially subscribing to the first source then
 * waiting for it to complete before subscribing to the next , and so on until the
 * last source completes . Any error interrupts the sequence immediately and is
 * forwarded downstream .
 * < img class = " marble " src = " doc - files / marbles / concatVarSources . svg " alt = " " >
 * @ param sources The { @ link Publisher } of { @ link Publisher } to concat
 * @ param < T > The type of values in both source and output sequences
 * @ return a new { @ link Flux } concatenating all source sequences */
@ SafeVarargs public static < T > Flux < T > concat ( Publisher < ? extends T > ... sources ) { } }

implementation: return onAssembly ( new FluxConcatArray < > ( false , sources ) ) ;


signature:
public class Flux { /** * Concatenate all sources emitted as an onNext signal from a parent { @ link Publisher } ,
 * forwarding elements emitted by the sources downstream .
 * Concatenation is achieved by sequentially subscribing to the first source then
 * waiting for it to complete before subscribing to the next , and so on until the
 * last source completes . Errors do not interrupt the main sequence but are propagated
 * after the rest of the sources have had a chance to be concatenated .
 * < img class = " marble " src = " doc - files / marbles / concatAsyncSources . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure upon cancellation .
 * @ param sources The { @ link Publisher } of { @ link Publisher } to concatenate
 * @ param < T > The type of values in both source and output sequences
 * @ return a new { @ link Flux } concatenating all inner sources sequences , delaying errors */
public static < T > Flux < T > concatDelayError ( Publisher < ? extends Publisher < ? extends T > > sources ) { } }

implementation: return concatDelayError ( sources , Queues . XS_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Concatenate all sources emitted as an onNext signal from a parent { @ link Publisher } ,
 * forwarding elements emitted by the sources downstream .
 * Concatenation is achieved by sequentially subscribing to the first source then
 * waiting for it to complete before subscribing to the next , and so on until the
 * last source completes . Errors do not interrupt the main sequence but are propagated
 * after the rest of the sources have had a chance to be concatenated .
 * < img class = " marble " src = " doc - files / marbles / concatAsyncSources . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure upon cancellation .
 * @ param sources The { @ link Publisher } of { @ link Publisher } to concatenate
 * @ param prefetch the inner source request size
 * @ param < T > The type of values in both source and output sequences
 * @ return a new { @ link Flux } concatenating all inner sources sequences until complete or error */
public static < T > Flux < T > concatDelayError ( Publisher < ? extends Publisher < ? extends T > > sources , int prefetch ) { } }

implementation: return onAssembly ( new FluxConcatMap < > ( from ( sources ) , identityFunction ( ) , Queues . get ( prefetch ) , prefetch , FluxConcatMap . ErrorMode . END ) ) ;


signature:
public class Flux { /** * Concatenate all sources provided as a vararg , forwarding elements emitted by the
 * sources downstream .
 * Concatenation is achieved by sequentially subscribing to the first source then
 * waiting for it to complete before subscribing to the next , and so on until the
 * last source completes . Errors do not interrupt the main sequence but are propagated
 * after the rest of the sources have had a chance to be concatenated .
 * < img class = " marble " src = " doc - files / marbles / concatVarSources . svg " alt = " " >
 * @ reactor . discard This operator discards elements it internally queued for backpressure upon cancellation .
 * @ param sources The { @ link Publisher } of { @ link Publisher } to concat
 * @ param < T > The type of values in both source and output sequences
 * @ return a new { @ link Flux } concatenating all source sequences */
@ SafeVarargs public static < T > Flux < T > concatDelayError ( Publisher < ? extends T > ... sources ) { } }

implementation: return onAssembly ( new FluxConcatArray < > ( true , sources ) ) ;


signature:
public class Flux { /** * Programmatically create a { @ link Flux } with the capability of emitting multiple
 * elements in a synchronous or asynchronous manner through the { @ link FluxSink } API .
 * This includes emitting elements from multiple threads .
 * < img class = " marble " src = " doc - files / marbles / createForFlux . svg " alt = " " >
 * This Flux factory is useful if one wants to adapt some other multi - valued async API
 * and not worry about cancellation and backpressure ( which is handled by buffering
 * all signals if the downstream can ' t keep up ) .
 * For example :
 * < pre > < code >
 * Flux . & lt ; String & gt ; create ( emitter - & gt ; {
 * ActionListener al = e - & gt ; {
 * emitter . next ( textField . getText ( ) ) ;
 * / / without cleanup support :
 * button . addActionListener ( al ) ;
 * / / with cleanup support :
 * button . addActionListener ( al ) ;
 * emitter . onDispose ( ( ) - > {
 * button . removeListener ( al ) ;
 * < / code > < / pre >
 * @ reactor . discard The { @ link FluxSink } exposed by this operator buffers in case of
 * overflow . The buffer is discarded when the main sequence is cancelled .
 * @ param < T > The type of values in the sequence
 * @ param emitter Consume the { @ link FluxSink } provided per - subscriber by Reactor to generate signals .
 * @ return a { @ link Flux }
 * @ see # push ( Consumer ) */
public static < T > Flux < T > create ( Consumer < ? super FluxSink < T > > emitter ) { } }

implementation: return create ( emitter , OverflowStrategy . BUFFER ) ;


signature:
public class Flux { /** * Programmatically create a { @ link Flux } with the capability of emitting multiple
 * elements in a synchronous or asynchronous manner through the { @ link FluxSink } API .
 * This includes emitting elements from multiple threads .
 * < img class = " marble " src = " doc - files / marbles / createWithOverflowStrategy . svg " alt = " " >
 * This Flux factory is useful if one wants to adapt some other multi - valued async API
 * and not worry about cancellation and backpressure ( which is handled by buffering
 * all signals if the downstream can ' t keep up ) .
 * For example :
 * < pre > < code >
 * Flux . & lt ; String & gt ; create ( emitter - & gt ; {
 * ActionListener al = e - & gt ; {
 * emitter . next ( textField . getText ( ) ) ;
 * / / without cleanup support :
 * button . addActionListener ( al ) ;
 * / / with cleanup support :
 * button . addActionListener ( al ) ;
 * emitter . onDispose ( ( ) - > {
 * button . removeListener ( al ) ;
 * } , FluxSink . OverflowStrategy . LATEST ) ;
 * < / code > < / pre >
 * @ reactor . discard The { @ link FluxSink } exposed by this operator discards elements
 * as relevant to the chosen { @ link OverflowStrategy } . For example , the { @ link OverflowStrategy # DROP }
 * discards each items as they are being dropped , while { @ link OverflowStrategy # BUFFER }
 * will discard the buffer upon cancellation .
 * @ param < T > The type of values in the sequence
 * @ param backpressure the backpressure mode , see { @ link OverflowStrategy } for the
 * available backpressure modes
 * @ param emitter Consume the { @ link FluxSink } provided per - subscriber by Reactor to generate signals .
 * @ return a { @ link Flux }
 * @ see # push ( Consumer , reactor . core . publisher . FluxSink . OverflowStrategy ) */
public static < T > Flux < T > create ( Consumer < ? super FluxSink < T > > emitter , OverflowStrategy backpressure ) { } }

implementation: return onAssembly ( new FluxCreate < > ( emitter , backpressure , FluxCreate . CreateMode . PUSH_PULL ) ) ;


signature:
public class Flux { /** * Programmatically create a { @ link Flux } with the capability of emitting multiple
 * elements from a single - threaded producer through the { @ link FluxSink } API . For
 * a multi - threaded capable alternative , see { @ link # create ( Consumer ) } .
 * < img class = " marble " src = " doc - files / marbles / push . svg " alt = " " >
 * This Flux factory is useful if one wants to adapt some other single - threaded
 * multi - valued async API and not worry about cancellation and backpressure ( which is
 * handled by buffering all signals if the downstream can ' t keep up ) .
 * For example :
 * < pre > < code >
 * Flux . & lt ; String & gt ; push ( emitter - & gt ; {
 * ActionListener al = e - & gt ; {
 * emitter . next ( textField . getText ( ) ) ;
 * / / without cleanup support :
 * button . addActionListener ( al ) ;
 * / / with cleanup support :
 * button . addActionListener ( al ) ;
 * emitter . onDispose ( ( ) - > {
 * button . removeListener ( al ) ;
 * < / code > < / pre >
 * @ reactor . discard The { @ link FluxSink } exposed by this operator buffers in case of
 * overflow . The buffer is discarded when the main sequence is cancelled .
 * @ param < T > The type of values in the sequence
 * @ param emitter Consume the { @ link FluxSink } provided per - subscriber by Reactor to generate signals .
 * @ return a { @ link Flux }
 * @ see # create ( Consumer ) */
public static < T > Flux < T > push ( Consumer < ? super FluxSink < T > > emitter ) { } }

implementation: return onAssembly ( new FluxCreate < > ( emitter , OverflowStrategy . BUFFER , FluxCreate . CreateMode . PUSH_ONLY ) ) ;


signature:
public class Flux { /** * Lazily supply a { @ link Publisher } every time a { @ link Subscription } is made on the
 * resulting { @ link Flux } , so the actual source instantiation is deferred until each
 * subscribe and the { @ link Supplier } can create a subscriber - specific instance .
 * If the supplier doesn ' t generate a new instance however , this operator will
 * effectively behave like { @ link # from ( Publisher ) } .
 * < img class = " marble " src = " doc - files / marbles / deferForFlux . svg " alt = " " >
 * @ param supplier the { @ link Publisher } { @ link Supplier } to call on subscribe
 * @ param < T > the type of values passing through the { @ link Flux }
 * @ return a deferred { @ link Flux } */
public static < T > Flux < T > defer ( Supplier < ? extends Publisher < T > > supplier ) { } }

implementation: return onAssembly ( new FluxDefer < > ( supplier ) ) ;


signature:
public class Flux { /** * Create a { @ link Flux } that terminates with an error immediately after being
 * subscribed to . The { @ link Throwable } is generated by a { @ link Supplier } , invoked
 * each time there is a subscription and allowing for lazy instantiation .
 * < img class = " marble " src = " doc - files / marbles / errorWithSupplier . svg " alt = " " >
 * @ param errorSupplier the error signal { @ link Supplier } to invoke for each { @ link Subscriber }
 * @ param < T > the reified type of the target { @ link Subscriber }
 * @ return a new failing { @ link Flux } */
public static < T > Flux < T > error ( Supplier < Throwable > errorSupplier ) { } }

implementation: return onAssembly ( new FluxErrorSupplied < > ( errorSupplier ) ) ;


signature:
public class Flux { /** * Create a { @ link Flux } that terminates with the specified error , either immediately
 * after being subscribed to or after being first requested .
 * < img class = " marble " src = " doc - files / marbles / errorWhenRequested . svg " alt = " " >
 * @ param throwable the error to signal to each { @ link Subscriber }
 * @ param whenRequested if true , will onError on the first request instead of subscribe ( ) .
 * @ param < O > the reified type of the target { @ link Subscriber }
 * @ return a new failing { @ link Flux } */
public static < O > Flux < O > error ( Throwable throwable , boolean whenRequested ) { } }

implementation: if ( whenRequested ) { return onAssembly ( new FluxErrorOnRequest < > ( throwable ) ) ; } else { return onAssembly ( new FluxError < > ( throwable ) ) ; }


signature:
public class Flux { /** * Pick the first { @ link Publisher } to emit any signal ( onNext / onError / onComplete ) and
 * replay all signals from that { @ link Publisher } , effectively behaving like the
 * fastest of these competing sources .
 * < img class = " marble " src = " doc - files / marbles / firstForFlux . svg " alt = " " >
 * @ param sources The competing source publishers
 * @ param < I > The type of values in both source and output sequences
 * @ return a new { @ link Flux } behaving like the fastest of its sources */
@ SafeVarargs public static < I > Flux < I > first ( Publisher < ? extends I > ... sources ) { } }

implementation: return onAssembly ( new FluxFirstEmitting < > ( sources ) ) ;


signature:
public class Flux { /** * Pick the first { @ link Publisher } to emit any signal ( onNext / onError / onComplete ) and
 * replay all signals from that { @ link Publisher } , effectively behaving like the
 * fastest of these competing sources .
 * < img class = " marble " src = " doc - files / marbles / firstForFlux . svg " alt = " " >
 * @ param sources The competing source publishers
 * @ param < I > The type of values in both source and output sequences
 * @ return a new { @ link Flux } behaving like the fastest of its sources */
public static < I > Flux < I > first ( Iterable < ? extends Publisher < ? extends I > > sources ) { } }


signature:
public class Flux { /** * Decorate the specified { @ link Publisher } with the { @ link Flux } API .
 * < img class = " marble " src = " doc - files / marbles / fromForFlux . svg " alt = " " >
 * @ param source the source to decorate
 * @ param < T > The type of values in both source and output sequences
 * @ return a new { @ link Flux } */
public static < T > Flux < T > from ( Publisher < ? extends T > source ) { } }

implementation: if ( source instanceof Flux ) { @ SuppressWarnings ( "unchecked" ) Flux < T > casted = ( Flux < T > ) source ; return casted ; } if ( source instanceof Fuseable . ScalarCallable ) { try { @ SuppressWarnings ( "unchecked" ) T t = ( ( Fuseable . ScalarCallable < T > ) source ) . call ( ) ; if ( t != null ) { return just ( t ) ; } return empty ( ) ; } catch ( Exception e ) { return error ( e ) ; } } return wrap ( source ) ;


signature:
public class Flux { /** * Create a { @ link Flux } that emits the items contained in the provided array .
 * < img class = " marble " src = " doc - files / marbles / fromArray . svg " alt = " " >
 * @ param array the array to read data from
 * @ param < T > The type of values in the source array and resulting Flux
 * @ return a new { @ link Flux } */
public static < T > Flux < T > fromArray ( T [ ] array ) { } }

implementation: if ( array . length == 0 ) { return empty ( ) ; } if ( array . length == 1 ) { return just ( array [ 0 ] ) ; } return onAssembly ( new FluxArray < > ( array ) ) ;


signature:
public class Flux { /** * Create a { @ link Flux } that emits the items contained in the provided { @ link Iterable } .
 * A new iterator will be created for each subscriber .
 * < img class = " marble " src = " doc - files / marbles / fromIterable . svg " alt = " " >
 * @ param it the { @ link Iterable } to read data from
 * @ param < T > The type of values in the source { @ link Iterable } and resulting Flux
 * @ return a new { @ link Flux } */
public static < T > Flux < T > fromIterable ( Iterable < ? extends T > it ) { } }

implementation: return onAssembly ( new FluxIterable < > ( it ) ) ;


signature:
public class Flux { /** * Create a { @ link Flux } that emits the items contained in the provided { @ link Stream } .
 * Keep in mind that a { @ link Stream } cannot be re - used , which can be problematic in
 * case of multiple subscriptions or re - subscription ( like with { @ link # repeat ( ) } or
 * { @ link # retry ( ) } ) . The { @ link Stream } is { @ link Stream # close ( ) closed } automatically
 * by the operator on cancellation , error or completion .
 * < img class = " marble " src = " doc - files / marbles / fromStream . svg " alt = " " >
 * @ param s the { @ link Stream } to read data from
 * @ param < T > The type of values in the source { @ link Stream } and resulting Flux
 * @ return a new { @ link Flux } */
public static < T > Flux < T > fromStream ( Stream < ? extends T > s ) { } }

implementation: Objects . requireNonNull ( s , "Stream s must be provided" ) ; return onAssembly ( new FluxStream < > ( ( ) -> s ) ) ;


signature:
public class Flux { /** * Create a { @ link Flux } that emits the items contained in a { @ link Stream } created by
 * the provided { @ link Supplier } for each subscription . The { @ link Stream } is
 * { @ link Stream # close ( ) closed } automatically by the operator on cancellation , error
 * or completion .
 * < img class = " marble " src = " doc - files / marbles / fromStream . svg " alt = " " >
 * @ param streamSupplier the { @ link Supplier } that generates the { @ link Stream } from
 * which to read data
 * @ param < T > The type of values in the source { @ link Stream } and resulting Flux
 * @ return a new { @ link Flux } */
public static < T > Flux < T > fromStream ( Supplier < Stream < ? extends T > > streamSupplier ) { } }

implementation: return onAssembly ( new FluxStream < > ( streamSupplier ) ) ;


signature:
public class Flux { /** * Programmatically create a { @ link Flux } by generating signals one - by - one via a
 * consumer callback .
 * < img class = " marble " src = " doc - files / marbles / generateStateless . svg " alt = " " >
 * @ param < T > the value type emitted
 * @ param generator Consume the { @ link SynchronousSink } provided per - subscriber by Reactor
 * to generate a < strong > single < / strong > signal on each pass .
 * @ return a { @ link Flux } */
public static < T > Flux < T > generate ( Consumer < SynchronousSink < T > > generator ) { } }

implementation: Objects . requireNonNull ( generator , "generator" ) ; return onAssembly ( new FluxGenerate < > ( generator ) ) ;


signature:
public class Flux { /** * Programmatically create a { @ link Flux } by generating signals one - by - one via a
 * consumer callback and some state . The { @ code stateSupplier } may return { @ literal null } .
 * < img class = " marble " src = " doc - files / marbles / generate . svg " alt = " " >
 * @ param < T > the value type emitted
 * @ param < S > the per - subscriber custom state type
 * @ param stateSupplier called for each incoming Subscriber to provide the initial state for the generator bifunction
 * @ param generator Consume the { @ link SynchronousSink } provided per - subscriber by Reactor
 * as well as the current state to generate a < strong > single < / strong > signal on each pass
 * and return a ( new ) state .
 * @ return a { @ link Flux } */
public static < T , S > Flux < T > generate ( Callable < S > stateSupplier , BiFunction < S , SynchronousSink < T > , S > generator ) { } }

implementation: return onAssembly ( new FluxGenerate < > ( stateSupplier , generator ) ) ;


signature:
public class Flux { /** * Create a { @ link Flux } that emits long values starting with 0 and incrementing at
 * specified time intervals , after an initial delay , on the global timer . If demand is
 * not produced in time , an onError will be signalled with an
 * { @ link Exceptions # isOverflow ( Throwable ) overflow } { @ code IllegalStateException }
 * detailing the tick that couldn ' t be emitted . In normal conditions , the { @ link Flux }
 * will never complete .
 * Runs on the { @ link Schedulers # parallel ( ) } Scheduler .
 * < img class = " marble " src = " doc - files / marbles / intervalWithDelay . svg " alt = " " >
 * @ param delay the { @ link Duration } to wait before emitting 0l
 * @ param period the period { @ link Duration } before each following increment
 * @ return a new { @ link Flux } emitting increasing numbers at regular intervals */
public static Flux < Long > interval ( Duration delay , Duration period ) { } }

implementation: return interval ( delay , period , Schedulers . parallel ( ) ) ;


signature:
public class Flux { /** * Create a { @ link Flux } that emits long values starting with 0 and incrementing at
 * specified time intervals , on the specified { @ link Scheduler } . If demand is not
 * produced in time , an onError will be signalled with an { @ link Exceptions # isOverflow ( Throwable ) overflow }
 * { @ code IllegalStateException } detailing the tick that couldn ' t be emitted .
 * In normal conditions , the { @ link Flux } will never complete .
 * < img class = " marble " src = " doc - files / marbles / interval . svg " alt = " " >
 * @ param period the period { @ link Duration } between each increment
 * @ param timer a time - capable { @ link Scheduler } instance to run on
 * @ return a new { @ link Flux } emitting increasing numbers at regular intervals */
public static Flux < Long > interval ( Duration period , Scheduler timer ) { } }

implementation: return onAssembly ( new FluxInterval ( period . toMillis ( ) , period . toMillis ( ) , TimeUnit . MILLISECONDS , timer ) ) ;


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

implementation: return merge ( source , Queues . SMALL_BUFFER_SIZE , Queues . XS_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Merge data from { @ link Publisher } sequences emitted by the passed { @ link Publisher }
 * into an interleaved merged sequence . Unlike { @ link # concat ( Publisher ) concat } , inner
 * sources are subscribed to eagerly ( but at most { @ code concurrency } sources are
 * subscribed to at the same time ) .
 * < img class = " marble " src = " doc - files / marbles / mergeAsyncSources . svg " alt = " " >
 * Note that merge is tailored to work with asynchronous sources or finite sources . When dealing with
 * an infinite source that doesn ' t already publish on a dedicated Scheduler , you must isolate that source
 * in its own Scheduler , as merge would otherwise attempt to drain it before subscribing to
 * another source .
 * @ param source a { @ link Publisher } of { @ link Publisher } sources to merge
 * @ param concurrency the request produced to the main source thus limiting concurrent merge backlog
 * @ param < T > the merged type
 * @ return a merged { @ link Flux } */
public static < T > Flux < T > merge ( Publisher < ? extends Publisher < ? extends T > > source , int concurrency ) { } }

implementation: return merge ( source , concurrency , Queues . XS_BUFFER_SIZE ) ;


signature:
public class Flux { /** * Merge data from { @ link Publisher } sequences emitted by the passed { @ link Publisher }
 * into an interleaved merged sequence . Unlike { @ link # concat ( Publisher ) concat } , inner
 * sources are subscribed to eagerly ( but at most { @ code concurrency } sources are
 * subscribed to at the same time ) .
 * < img class = " marble " src = " doc - files / marbles / mergeAsyncSources . svg " alt = " " >
 * Note that merge is tailored to work with asynchronous sources or finite sources . When dealing with
 * an infinite source that doesn ' t already publish on a dedicated Scheduler , you must isolate that source
 * in its own Scheduler , as merge would otherwise attempt to drain it before subscribing to
 * another source .
 * @ param source a { @ link Publisher } of { @ link Publisher } sources to merge
 * @ param concurrency the request produced to the main source thus limiting concurrent merge backlog
 * @ param prefetch the inner source request size
 * @ param < T > the merged type
 * @ return a merged { @ link Flux } */
public static < T > Flux < T > merge ( Publisher < ? extends Publisher < ? extends T > > source , int concurrency , int prefetch ) { } }

implementation: return onAssembly ( new FluxFlatMap < > ( from ( source ) , identityFunction ( ) , false , concurrency , Queues . get ( concurrency ) , prefetch , Queues . get ( prefetch ) ) ) ;


signature:
public class Flux { /** * Merge data from { @ link Publisher } sequences contained in an { @ link Iterable }
 * into an interleaved merged sequence . Unlike { @ link # concat ( Publisher ) concat } , inner
 * sources are subscribed to eagerly .
 * A new { @ link Iterator } will be created for each subscriber .
 * < img class = " marble " src = " doc - files / marbles / mergeFixedSources . svg " alt = " " >
 * Note that merge is tailored to work with asynchronous sources or finite sources . When dealing with
 * an infinite source that doesn ' t already publish on a dedicated Scheduler , you must isolate that source
 * in its own Scheduler , as merge would otherwise attempt to drain it before subscribing to
 * another source .
 * @ param sources the { @ link Iterable } of sources to merge ( will be lazily iterated on subscribe )
 * @ param < I > The source type of the data sequence
 * @ return a merged { @ link Flux } */
public static < I > Flux < I > merge ( Iterable < ? extends Publisher < ? extends I > > sources ) { } }

implementation: return merge ( fromIterable ( sources ) ) ;


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

implementation: return merge ( Queues . XS_BUFFER_SIZE , sources ) ;


signature:
public class Flux { /** * Merge data from { @ link Publisher } sequences contained in an array / vararg
 * into an interleaved merged sequence . Unlike { @ link # concat ( Publisher ) concat } ,
 * sources are subscribed to eagerly .
 * < img class = " marble " src = " doc - files / marbles / mergeFixedSources . svg " alt = " " >
 * Note that merge is tailored to work with asynchronous sources or finite sources . When dealing with
 * an infinite source that doesn ' t already publish on a dedicated Scheduler , you must isolate that source
 * in its own Scheduler , as merge would otherwise attempt to drain it before subscribing to
 * another source .
 * @ param sources the array of { @ link Publisher } sources to merge
 * @ param prefetch the inner source request size
 * @ param < I > The source type of the data sequence
 * @ return a fresh Reactive { @ link Flux } publisher ready to be subscribed */
@ SafeVarargs public static < I > Flux < I > merge ( int prefetch , Publisher < ? extends I > ... sources ) { } }

implementation: return merge ( prefetch , false , sources ) ;


signature:
public class Flux { /** * Merge data from { @ link Publisher } sequences contained in an array / vararg
 * into an interleaved merged sequence . Unlike { @ link # concat ( Publisher ) concat } ,
 * sources are subscribed to eagerly .
 * This variant will delay any error until after the rest of the merge backlog has been processed .
 * < img class = " marble " src = " doc - files / marbles / mergeFixedSources . svg " alt = " " >
 * Note that merge is tailored to work with asynchronous sources or finite sources . When dealing with
 * an infinite source that doesn ' t already publish on a dedicated Scheduler , you must isolate that source
 * in its own Scheduler , as merge would otherwise attempt to drain it before subscribing to
 * another source .
 * @ param sources the array of { @ link Publisher } sources to merge
 * @ param prefetch the inner source request size
 * @ param < I > The source type of the data sequence
 * @ return a fresh Reactive { @ link Flux } publisher ready to be subscribed */
@ SafeVarargs public static < I > Flux < I > mergeDelayError ( int prefetch , Publisher < ? extends I > ... sources ) { } }

implementation: return merge ( prefetch , true , sources ) ;


signature:
public class Flux { /** * Merge data from provided { @ link Publisher } sequences into an ordered merged sequence ,
 * by picking the smallest values from each source ( as defined by their natural order ) .
 * This is not a { @ link # sort ( ) } , as it doesn ' t consider the whole of each sequences .
 * Instead , this operator considers only one value from each source and picks the
 * smallest of all these values , then replenishes the slot for that picked source .
 * < img class = " marble " src = " doc - files / marbles / mergeOrderedNaturalOrder . svg " alt = " " >
 * @ param sources { @ link Publisher } sources of { @ link Comparable } to merge
 * @ param < I > a { @ link Comparable } merged type that has a { @ link Comparator # naturalOrder ( ) natural order }
 * @ return a merged { @ link Flux } that , subscribing early but keeping the original ordering */
@ SafeVarargs public static < I extends Comparable < ? super I > > Flux < I > mergeOrdered ( Publisher < ? extends I > ... sources ) { } }

implementation: return mergeOrdered ( Queues . SMALL_BUFFER_SIZE , Comparator . naturalOrder ( ) , sources ) ;


signature:
public class Flux { /** * Merge data from provided { @ link Publisher } sequences into an ordered merged sequence ,
 * by picking the smallest values from each source ( as defined by the provided
 * { @ link Comparator } ) . This is not a { @ link # sort ( Comparator ) } , as it doesn ' t consider
 * the whole of each sequences .
 * Instead , this operator considers only one value from each source and picks the
 * smallest of all these values , then replenishes the slot for that picked source .
 * < img class = " marble " src = " doc - files / marbles / mergeOrdered . svg " alt = " " >
 * @ param comparator the { @ link Comparator } to use to find the smallest value
 * @ param sources { @ link Publisher } sources to merge
 * @ param < T > the merged type
 * @ return a merged { @ link Flux } that , subscribing early but keeping the original ordering */
@ SafeVarargs public static < T > Flux < T > mergeOrdered ( Comparator < ? super T > comparator , Publisher < ? extends T > ... sources ) { } }

implementation: return mergeOrdered ( Queues . SMALL_BUFFER_SIZE , comparator , sources ) ;


signature:
public class Flux { /** * Merge data from provided { @ link Publisher } sequences into an ordered merged sequence ,
 * by picking the smallest values from each source ( as defined by the provided
 * { @ link Comparator } ) . This is not a { @ link # sort ( Comparator ) } , as it doesn ' t consider
 * the whole of each sequences .
 * Instead , this operator considers only one value from each source and picks the
 * smallest of all these values , then replenishes the slot for that picked source .
 * < img class = " marble " src = " doc - files / marbles / mergeOrdered . svg " alt = " " >
 * @ param prefetch the number of elements to prefetch from each source ( avoiding too
 * many small requests to the source when picking )
 * @ param comparator the { @ link Comparator } to use to find the smallest value
 * @ param sources { @ link Publisher } sources to merge
 * @ param < T > the merged type
 * @ return a merged { @ link Flux } that , subscribing early but keeping the original ordering */
@ SafeVarargs public static < T > Flux < T > mergeOrdered ( int prefetch , Comparator < ? super T > comparator , Publisher < ? extends T > ... sources ) { } }

implementation: if ( sources . length == 0 ) { return empty ( ) ; } if ( sources . length == 1 ) { return from ( sources [ 0 ] ) ; } return onAssembly ( new FluxMergeOrdered < > ( prefetch , Queues . get ( prefetch ) , comparator , sources ) ) ;


signature:
public class Flux { /** * Merge data from { @ link Publisher } sequences emitted by the passed { @ link Publisher }
 * into an ordered merged sequence . Unlike concat , the inner publishers are subscribed to
 * eagerly . Unlike merge , their emitted values are merged into the final sequence in
 * subscription order .
 * < img class = " marble " src = " doc - files / marbles / mergeSequentialAsyncSources . svg " alt = " " >
 * @ param sources a { @ link Publisher } of { @ link Publisher } sources to merge
 * @ param < T > the merged type
 * @ return a merged { @ link Flux } , subscribing early but keeping the original ordering */
public static < T > Flux < T > mergeSequential ( Publisher < ? extends Publisher < ? extends T > > sources ) { } }

implementation: return mergeSequential ( sources , false , Queues . SMALL_BUFFER_SIZE , Queues . XS_BUFFER_SIZE ) ;


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

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


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

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


signature:
public class Flux { /** * Merge data from { @ link Publisher } sequences provided in an array / vararg
 * into an ordered merged sequence . Unlike concat , sources are subscribed to
 * eagerly . Unlike merge , their emitted values are merged into the final sequence in subscription order .
 * < img class = " marble " src = " doc - files / marbles / mergeSequentialVarSources . svg " alt = " " >
 * @ param sources a number of { @ link Publisher } sequences to merge
 * @ param < I > the merged type
 * @ return a merged { @ link Flux } , subscribing early but keeping the original ordering */
@ SafeVarargs public static < I > Flux < I > mergeSequential ( Publisher < ? extends I > ... sources ) { } }

implementation: return mergeSequential ( Queues . XS_BUFFER_SIZE , false , sources ) ;


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

implementation: return mergeSequential ( prefetch , false , sources ) ;


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

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


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