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;
   }
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 ) ) ; iter . onClose ( ( ) -> closeIterator ( origin , requestId ) ) ; currentRequests . put ( requestId , iter ) ; 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.
@param streamSupplier the supplied stream
@param intOps the intermediate operations to perform on the stream
@param <Original> original stream type
@param <E> resulting type
@return the id of the managed iterator