Package org.hibernate.reactive.util.async.impl

Interface AsyncIterator<T>

Type Parameters:
T - Type of object being iterated over.
All Superinterfaces:
AsyncCloseable
public interface AsyncIterator<T> extends AsyncCloseable
Copy of com.ibm.asyncutil.iteration.AsyncIterator from com.ibm.async:asyncutil:0.1.0 without all the methods and imports we don't need for Hibernate Reactive.

A mechanism for asynchronously generating and consuming values

Consider this an async version of Stream.

AsyncIterators have lazy, pull based, evaluation semantics - values are not computed until they are needed. AsyncIterators are not immutable - like streams, each value they produce is consumed only once. Typically you should not apply multiple transformations to the same source AsyncIterator, it almost certainly won't do what you want it to do.

Implementors of the interface need only implement nextStage().

A note on thread safety: This class makes no assumption that nextStage() is thread safe! Many methods that generate transformed iterators assume that nextStage will not be called concurrently, and even stronger, that nextStage won't be called again until the previous stage returned by nextStage has completed.

Parallelization may still be accomplished using the partially eager methods described below. The difference is that the parallelization in that case is from producing values in parallel, not consuming values in parallel.

To implement an AsyncIterator you must only implement the nextStage() method - however, it is recommended that users avoid actually using nextStage to consume the results of iteration. It is less expressive and it can also be error prone; it is easy to cause a stack overflow by incorrectly recursing on calls to nextStage. You should prefer to use the other higher level methods on this interface.

There are 2 main categories of such methods on this interface: Intermediate and Terminal. These methods can be combined to form pipelines, which generally consist of a source (often created with the static constructor methods on this interface (fromIterator(Iterator), etc)), followed by zero or more intermediate operations (such as filter(Predicate), thenApply(Function)), and completed with a terminal operation which returns a CompletionStage (such as forEach(Consumer)). For example, suppose we wanted to accomplish the following (blocking) procedure: 


 // request and lookup records one by one until we get 10 relevant records
 List<Record> records = new ArrayList<>()
 while (records.size() < 10) {
     // ask for a record identifier from a remote service (blocking)
     RecordId response = requestIdFromIdServer();
     // get the actual record from another service (blocking)
     Record record = getRecordFromRecordServer(recordIdentifier);
     // only add relevant records
     if (isRelevant(record)) {
        records.add(record);
     }
 }

If we wanted to do it without doing any blocking, we can use a pipeline and return a CompletionStage of the desired record list. Like the blocking version only one request will be made at a time. 


 CompletionStage<RecordId> requestIdFromIdServer();
 CompletionStage<Record> getRecordFromRecordServer(RecordId recordId);

 CompletionStage<List<Response>> responses =
   AsyncIterator.generate(this::requestIdFromIdServer) // source iterator
  .thenCompose(this::getRecordFromRecordServer)        // intermediate transformation
  .filter(record -> isRelevant(record))                // intermediate transformation
  .take(10)                                            // intermediate transformation
  .collect(Collectors.toList());                       // terminal operation

Intermediate methods - All methods which return AsyncIterators are intermediate methods. They can further be broken down into lazy and partially eager methods. Methods that end with the suffix ahead are partially eager, the rest are lazy. A lazy intermediate transformation will not be evaluated until some downstream eager operation is called. Furthermore, only what is needed to satisfy the eager operation will be evaluated from the previous iterator in the chain. When only requesting a single element from the transformed iterator, only a single element may be evaluated from the previous iterator (ex: thenApply(Function)), or potentially many elements (ex: filter(Predicate)).

Methods ending with the suffix ahead , are partially eager. They can be used when there is an expensive transformation step that should be performed in parallel. They will eagerly consume from their upstream iterator up to a specified amount (still sequentially!) and eagerly apply the transformation step.

Intermediate methods will propagate exceptions similarly to CompletionStage, a dependent AsyncIterator will return exceptional stages if the upstream iterator generated exceptional elements.

Terminal methods - Terminal methods consume the iterator and return a CompletionStage. After a terminal operation is called, the iterator is considered consumed and should not be used further. If any of the stages in the chain that comprise this iterator were exceptional, the CompletionStage returned by a terminal operation will also be exceptional. The exception will short-circuit the terminal operation. For example, a terminal operation such as forEach(Consumer) will not to continue to run on subsequent elements of the iterator and instead immediately complete its returned stage with the error. Unless otherwise noted, this behavior holds for all terminal methods but may not documented explicitly.

The exception propagation scheme should be familiar to users of CompletionStage, upstream errors will appear wherever the AsyncIterator is consumed and the result is observed (with CompletableFuture.join() for instance).

Unless otherwise noted, methods on this interface are free to throw NullPointerException if any of the provided arguments are null.

The behavior of an AsyncIterator if nextStage() is called after the end of iteration marker is returned is left to the implementation.

This interface extends AsyncCloseable, if there are resources associated with this iterator that must be relinquished after iteration is complete, the close() method should be implemented. Because the majority of methods do not have a manually managed resource, a default implementation of close which does nothing is provided. Terminal methods on this interface do not call close(), it is generally the user's responsibility..

See Also:

  • Method Details

    • nextStage

      CompletionStage<Either<AsyncIterator.End,T>> nextStage()
      Returns a stage that will be completed with the next element of this iterator when it becomes available, or AsyncIterator.End if there are no more elements.

      This is not a terminal method, it can be safely called multiple times. However, this method is not thread safe, and should only be called in a single-threaded fashion. Moreover, sequential calls should not be made until the CompletionStage returned by the previous call has completed. That is to say, 

      
 // illegal
 pool.execute(() -> nextStage())
 pool.execute(() -> nextStage())

 // just as illegal
 f1 = nextStage();
 f2 = nextStage();

 // good
 nextStage().thenCompose(t -> nextStage());

      Though this is not a terminal method, if a terminal method has been called it is no longer safe to call this method. When nextStage returns AsyncIterator.End, the iterator has no more elements. After an iterator emits an AsyncIterator.End indicator, the result of subsequent calls to nextStage is undefined.

      An AsyncIterator may be capable of producing normally completing stages after having producing exceptionally completed stages. nextStage is unique in that it can safely continue to be called even after a returned stage completes exceptionally, whereas all terminal operations short circuit when encountering an exception. If a user wishes to continue iteration after exception, they must use nextStage directly.

      Returns:
      A CompletionStage of the next element for iteration held in the Either.right() position, or an instance of AsyncIterator.End held in the Either.left() position indicating the end of iteration.

    • close

      default CompletionStage<Void> close()
      Relinquishes any resources associated with this iterator.

      This method should be overridden if manual resource management is required, the default implementation does nothing. This method is not thread safe, and must not be called concurrently with calls to nextStage(). This method is not automatically called by terminal methods, and must be explicitly called after iteration is complete if the underlying iterator has resources to release. Similar to the situation with BaseStream.close(), because the common case requires no resources the user should only call close if it is possible that the AsyncIterator has resources. Special care needs to be taken to call close even in the case of an exception. 

      
 class SocketBackedIterator implements AsyncIterator<byte[]> {
  ...
  {@literal @Override}
  CompletionStage<Void> close() { return socket.close(); }
 }
 AsyncCloseable.tryComposeWith(new SocketBackedIterator(socket), socketIt -> socketIt
  .thenCompose(this::deserialize)
  .filter(this::isRelevantMessage)
  .forEach(message -> System.out.println(message)));

      Intermediate methods will pass calls to close to their upstream iterators, so it is safe to call close on an intermediate result of an iterator instead of on it directly. For example, 

      
 AsyncIterator<byte[]> original = new SocketBackedIterator(socket);
 AsyncIterator<Message> transformed = original.thenCompose(this::deserialize).filter(this::isRelevantMessage);

 transformed.close() // will close on original
      Specified by:
      close in interface AsyncCloseable
      Returns:
      a CompletionStage that completes when all resources associated with this iterator have been relinquished.

    • thenApply

      default <U> AsyncIterator<U> thenApply(Function<? super T,? extends U> fn)
      Transforms this into a new AsyncIterator that iterates over the results of fn applied to the outcomes of stages in this iterator when they complete normally. When stages in this iterator complete exceptionally the returned iterator will emit an exceptional stage without applying fn. 
       
 intIterator // 1,2,3,...
     .thenApply(Integer::toString) //"1","2","3"...

      This is a lazy intermediate method.

      Parameters:
      fn - A function which produces a U from the given T
      Returns:
      A new AsyncIterator which produces stages of fn applied to the result of the stages from this iterator

    • thenCompose

      default <U> AsyncIterator<U> thenCompose(Function<? super T,? extends CompletionStage<U>> fn)
      Transforms this into a new AsyncIterator using the produced stages of fn applied to the output from the stages of this. When stages in this iterator complete exceptionally the returned iterator will emit an exceptional stage without applying fn. 
       
 CompletableFuture<String> asyncToString(final int i);
 intIterator // 1, 2, 3
   .thenCompose(this::asyncToString); //"1", "2", "3"...

      This is a lazy intermediate method.

      Parameters:
      fn - A function which produces a new CompletionStage from a T
      Returns:
      A new AsyncIterator which produces stages of fn composed with the result of the stages from this iterator

    • filter

      default AsyncIterator<T> filter(Predicate<? super T> predicate)
      Transforms the AsyncIterator into one which will only produce results that match predicate.

      This is a lazy intermediate method.

      Parameters:
      predicate - A function that takes a T and returns true if it should be returned by the new iterator, and false otherwise
      Returns:
      a new AsyncIterator which will only return results that match predicate

    • batch

      default <A, R> AsyncIterator<R> batch(Collector<? super T,A,R> collector, BiPredicate<? super A,? super T> shouldAddToBatch)
      Collects the results of this iterator in batches, returning an iterator of those batched collections.

      This may be useful for performing bulk operations on many elements, rather than on one element at a time.

      This is a lazy intermediate method.

      Parameters:
      collector - a Collector used to collect the elements of this iterator into individual batches. Each batch will be created by invoking the collector's Collector.supplier() method
      shouldAddToBatch - a predicate which determines whether a given element encountered during iteration should be added to the given (current) batch. If this predicate returns true for the given element and container, the element will be added to the container, and the batching operation will continue to draw from the underlying iterator. If this predicate returns false, the element will not be added and the current batch will be finished and returned by the batching iterator. The element which did not meet the predicate will be tested again by the next batch
      Returns:
      an AsyncIterator which invokes several iterations of the underlying iterator with each advance, collecting these elements into containers provided by the given Collector.

    • batch

      default <A, R> AsyncIterator<R> batch(Collector<? super T,A,R> collector, int batchSize)
      A convenience method provided to invoke batch(Collector, BiPredicate) with a predicate that limits batches to a fixed size.

      Each batch will be as large as the given batchSize except possibly the last one, which may be smaller due to exhausting the underlying iterator.

      This is a lazy intermediate method.

      See Also:

    • fold

      default <U> CompletionStage<U> fold(U identity, BiFunction<U,? super T,U> accumulator)
      Sequentially accumulates the elements of type T in this iterator into a U. This provides an immutable style terminal reduction operation as opposed to the mutable style supported by collect(java.util.stream.Collector<? super T, A, R>). For example, to sum the lengths of Strings in an AsyncIterator, stringIt.fold(0, (acc, s) -> acc + s.length()).

      This is a terminal method.

      Parameters:
      accumulator - a function that produces a new accumulation from an existing accumulation and a new element
      identity - a starting U value
      Returns:
      a CompletionStage containing the resulting U from repeated application of accumulator

    • collect

      default <R, A> CompletionStage<R> collect(Collector<? super T,A,R> collector)
      Performs a mutable reduction operation using collector and return a CompletionStage of the result.

      This is a terminal method.

      Type Parameters:
      A - The intermediate type of the accumulated object
      R - The final type of the accumulated object
      Parameters:
      collector - a Collector which will sequentially collect the contents of this iterator into an R
      Returns:
      a CompletionStage which will complete with the collected value
      See Also:

    • collect

      default <R> CompletionStage<R> collect(Supplier<R> supplier, BiConsumer<R,? super T> accumulator)
      Performs a mutable reduction operation and return a CompletionStage of the result. A mutable reduction is one where the accumulator has mutable state and additional elements are incorporated by updating that state.

      This is a terminal method.

      Parameters:
      supplier - a supplier for a stateful accumulator
      accumulator - a function which can incorporate T elements into a stateful accumulation
      Returns:
      a CompletionStage which will complete with the accumulated value
      See Also:

    • range

      static AsyncIterator<Long> range(long start, long end)
      Creates an AsyncIterator for a range.

      Similar to for(i = start; i < end; i++).

      The stages returned by nextStage will be already completed.

      Parameters:
      start - the start point of iteration (inclusive)
      end - the end point of iteration (exclusive)
      Returns:
      an AsyncIterator that will return longs from start to end

    • forEach

      default CompletionStage<Void> forEach(Consumer<? super T> action)
      Performs the side effecting action until the end of iteration is reached

      This is a terminal method.

      Parameters:
      action - a side-effecting action that takes a T
      Returns:
      a CompletionStage that returns when there are no elements left to apply action to, or an exception has been encountered.

    • find

      default CompletionStage<Optional<T>> find(Predicate<? super T> predicate)
      Gets the first element that satisfies predicate, or empty if no such element exists

      This is a terminal method.

      Parameters:
      predicate - the predicate that returns true for the desired element
      Returns:
      a CompletionStage that completes with the first T to satisfy predicate, or empty if no such T exists

    • empty

      static <T> AsyncIterator<T> empty()
      Creates an empty AsyncIterator.
      Returns:
      an AsyncIterator that will immediately produce an AsyncIterator.End marker

    • fromIterator

      static <T> AsyncIterator<T> fromIterator(Iterator<? extends T> iterator)
      Creates an AsyncIterator from an Iterator
      Parameters:
      iterator - an Iterator of T elements
      Returns:
      A new AsyncIterator which will yield the elements of iterator

    • generate

      static <T> AsyncIterator<T> generate(Supplier<? extends CompletionStage<T>> supplier)
      Creates an infinite AsyncIterator of type T.
      Parameters:
      supplier - supplies stages for elements to be yielded by the returned iterator
      Returns:
      AsyncIterator returning values generated from supplier