public < Original , E > OnCloseIterator < E > start ( Address origin , Supplier < Stream < Original > > streamSupplier , Iterable < IntermediateOperation > intOps , Object requestId ) { if ( trace ) { log . tracef ( "Iterator requested from %s using requestId %s" , origin , requestId ) ; } BaseStream stream = streamSupplier . get ( ) ; for ( IntermediateOperation intOp : intOps ) { stream = intOp . perform ( stream ) ; } OnCloseIterator < E > iter = new IteratorCloser <> ( ( CloseableIterator < E > ) Closeables . iterator ( stream ) ) ; // When the iterator is closed make sure to clean up iter . onClose ( ( ) -> closeIterator ( origin , requestId ) ) ; currentRequests . put ( requestId , iter ) ; // This will be null if there have been no iterators for this node. // If the originating node died before we start this iterator this could be null as well. In this case the // iterator will be closed on the next view change. Set < Object > ids = ownerRequests . computeIfAbsent ( origin , k -> ConcurrentHashMap . newKeySet ( ) ) ; ids . add ( requestId ) ; return iter ; }
Starts an iteration process from the given stream that converts the stream to a subsequent stream using the given intermediate operations and then creates a managed iterator for the caller to subsequently retrieve .