public void initFromIterator ( Iterator < IN > iterator ) { this . pending = new ArrayDeque <> ( ) ; isIteratorBased = true ; //this._streams = kxReactiveStreams; needs sync init !! // in case iterator is blocking, need a dedicated thread .. Executor iteratorThread = new ThreadPoolExecutor ( 0 , 1 , 10L , TimeUnit . MILLISECONDS , new LinkedBlockingQueue < Runnable > ( ) ) ; producer = new Subscription ( ) { boolean complete = false ; @ Override public void request ( long outern ) { iteratorThread . execute ( ( ) -> { if ( complete ) { return ; } long n = outern ; try { while ( iterator . hasNext ( ) && n -- > 0 ) { self ( ) . onNext ( iterator . next ( ) ) ; } if ( ! iterator . hasNext ( ) ) { complete = true ; self ( ) . onComplete ( ) ; } } catch ( Throwable t ) { self ( ) . onError ( t ) ; } } ) ; } @ Override public void cancel ( ) { } } ; processor = in -> ( OUT ) in ; onSubscribe ( producer ) ; Thread . currentThread ( ) . setName ( Thread . currentThread ( ) + " (rx async stream processor)" ) ; }
acts as an pull based event producer then