@ SuppressWarnings ( "unchecked" ) static < T , R > boolean trySubscribeScalarMap ( Publisher < ? extends T > source , CoreSubscriber < ? super R > s , Function < ? super T , ? extends Publisher < ? extends R > > mapper , boolean fuseableExpected ) { if ( source instanceof Callable ) { T t ; try { t = ( ( Callable < ? extends T > ) source ) . call ( ) ; } catch ( Throwable e ) { Operators . error ( s , Operators . onOperatorError ( e , s . currentContext ( ) ) ) ; return true ; } if ( t == null ) { Operators . complete ( s ) ; return true ; } Publisher < ? extends R > p ; try { p = Objects . requireNonNull ( mapper . apply ( t ) , "The mapper returned a null Publisher" ) ; } catch ( Throwable e ) { Operators . error ( s , Operators . onOperatorError ( null , e , t , s . currentContext ( ) ) ) ; return true ; } if ( p instanceof Callable ) { R v ; try { v = ( ( Callable < R > ) p ) . call ( ) ; } catch ( Throwable e ) { Operators . error ( s , Operators . onOperatorError ( null , e , t , s . currentContext ( ) ) ) ; return true ; } if ( v != null ) { s . onSubscribe ( Operators . scalarSubscription ( s , v ) ) ; } else { Operators . complete ( s ) ; } } else { if ( ! fuseableExpected || p instanceof Fuseable ) { p . subscribe ( s ) ; } else { p . subscribe ( new FluxHide . SuppressFuseableSubscriber <> ( s ) ) ; } } return true ; } return false ; }
Checks if the source is a Supplier and if the mapper s publisher output is also a supplier thus avoiding subscribing to any of them .