@ SafeVarargs public static < U > SimpleReactStream < U > firstOf ( final SimpleReactStream < U > ... futureStreams ) { final List < Tuple2 < SimpleReactStream < U > , QueueReader > > racers = Stream . of ( futureStreams ) . map ( s -> Tuple . tuple ( s , new Queue . QueueReader ( s . toQueue ( ) , null ) ) ) . collect ( Collectors . toList ( ) ) ; while ( true ) { for ( final Tuple2 < SimpleReactStream < U > , Queue . QueueReader > q : racers ) { if ( q . _2 ( ) . notEmpty ( ) ) { EagerFutureStreamFunctions . closeOthers ( q . _2 ( ) . getQueue ( ) , racers . stream ( ) . map ( t -> t . _2 ( ) . getQueue ( ) ) . collect ( Collectors . toList ( ) ) ) ; closeOthers ( q . _1 ( ) , racers . stream ( ) . map ( t -> t . _1 ( ) ) . collect ( Collectors . toList ( ) ) ) ; return q . _1 ( ) . fromStream ( q . _2 ( ) . getQueue ( ) . stream ( q . _1 ( ) . getSubscription ( ) ) ) ; } } LockSupport . parkNanos ( 1l ) ; } }
Return first Stream out of provided Streams that starts emitted results