Package org.eclipse.microprofile.reactive.streams.operators

Interface ProcessorBuilder<T,R>

Type Parameters:
T - The type of the elements that the processor consumes.
R - The type of the elements that the processor emits.
All Superinterfaces:
ConnectingOperators<R>, ConsumingOperators<R>, ErrorHandlingOperators<R>, FilteringOperators<R>, PeekingOperators<R>, TransformingOperators<R>
public interface ProcessorBuilder<T,R> extends TransformingOperators<R>, FilteringOperators<R>, PeekingOperators<R>, ConsumingOperators<R>, ErrorHandlingOperators<R>, ConnectingOperators<R>
A builder for a Processor.

The documentation for each operator uses marble diagrams to visualize how the operator functions. Each element flowing in and out of the stream is represented as a coloured marble that has a value, with the operator applying some transformation or some side effect, termination and error signals potentially being passed, and for operators that subscribe to the stream, an output value being redeemed at the end.

Below is an example diagram labelling all the parts of the stream.

Example marble diagram

See Also:

  • Method Details

    • map

      <S> ProcessorBuilder<T,S> map(Function<? super R,? extends S> mapper)
      Map the elements emitted by this stream using the mapper function.

      map marbles diagram

      Specified by:
      map in interface TransformingOperators<T>
      Type Parameters:
      S - The type of elements that the mapper function emits.
      Parameters:
      mapper - The function to use to map the elements.
      Returns:
      A new stream builder that emits the mapped elements.

    • flatMap

      <S> ProcessorBuilder<T,S> flatMap(Function<? super R,? extends PublisherBuilder<? extends S>> mapper)
      Map the elements to publishers, and flatten so that the elements emitted by publishers produced by the mapper function are emitted from this stream.

      flatMap marbles diagram

      This method operates on one publisher at a time. The result is a concatenation of elements emitted from all the publishers produced by the mapper function.

      Unlike TransformingOperators.flatMapRsPublisher(Function)}, the mapper function returns a org.eclipse.microprofile.reactive.streams type instead of an org.reactivestreams type.

      Specified by:
      flatMap in interface TransformingOperators<T>
      Type Parameters:
      S - The type of the elements emitted from the new stream.
      Parameters:
      mapper - The mapper function.
      Returns:
      A new stream builder.

    • flatMapRsPublisher

      <S> ProcessorBuilder<T,S> flatMapRsPublisher(Function<? super R,? extends org.reactivestreams.Publisher<? extends S>> mapper)
      Map the elements to publishers, and flatten so that the elements emitted by publishers produced by the mapper function are emitted from this stream.

      flatMapRsPublisher marble diagram

      This method operates on one publisher at a time. The result is a concatenation of elements emitted from all the publishers produced by the mapper function.

      Unlike TransformingOperators.flatMap(Function), the mapper function returns a org.eclipse.microprofile.reactive.streams builder instead of an org.reactivestreams type.

      Specified by:
      flatMapRsPublisher in interface TransformingOperators<T>
      Type Parameters:
      S - The type of the elements emitted from the new stream.
      Parameters:
      mapper - The mapper function.
      Returns:
      A new stream builder.

    • flatMapCompletionStage

      <S> ProcessorBuilder<T,S> flatMapCompletionStage(Function<? super R,? extends CompletionStage<? extends S>> mapper)
      Map the elements to CompletionStage, and flatten so that the elements the values redeemed by each CompletionStage are emitted from this stream.

      flatMapCompletionStage marble diagram

      This method only works with one element at a time. When an element is received, the mapper function is executed, and the next element is not consumed or passed to the mapper function until the previous CompletionStage is redeemed. Hence this method also guarantees that ordering of the stream is maintained.

      Specified by:
      flatMapCompletionStage in interface TransformingOperators<T>
      Type Parameters:
      S - The type of the elements emitted from the new stream.
      Parameters:
      mapper - The mapper function.
      Returns:
      A new stream builder.

    • flatMapIterable

      <S> ProcessorBuilder<T,S> flatMapIterable(Function<? super R,? extends Iterable<? extends S>> mapper)
      Map the elements to Iterable's, and flatten so that the elements contained in each iterable are emitted by this stream.

      flatMapIterable marble diagram

      This method operates on one iterable at a time. The result is a concatenation of elements contain in all the iterables returned by the mapper function.

      Specified by:
      flatMapIterable in interface TransformingOperators<T>
      Type Parameters:
      S - The type of the elements emitted from the new stream.
      Parameters:
      mapper - The mapper function.
      Returns:
      A new stream builder.

    • filter

      ProcessorBuilder<T,R> filter(Predicate<? super R> predicate)
      Filter elements emitted by this publisher using the given Predicate.

      Any elements that return true when passed to the Predicate will be emitted, all other elements will be dropped.

      filter marbles diagram

      Specified by:
      filter in interface FilteringOperators<T>
      Parameters:
      predicate - The predicate to apply to each element.
      Returns:
      A new stream builder.

    • distinct

      ProcessorBuilder<T,R> distinct()
      Creates a stream consisting of the distinct elements (according to Object.equals(Object)) of this stream.

      distinct marbles diagram

      Specified by:
      distinct in interface FilteringOperators<T>
      Returns:
      A new stream builder emitting the distinct elements from this stream.

    • limit

      ProcessorBuilder<T,R> limit(long maxSize)
      Truncate this stream, ensuring the stream is no longer than maxSize elements in length.

      limit marble diagram

      If maxSize is reached, the stream will be completed, and upstream will be cancelled. Completion of the stream will occur immediately when the element that satisfies the maxSize is received.

      Specified by:
      limit in interface FilteringOperators<T>
      Parameters:
      maxSize - The maximum size of the returned stream.
      Returns:
      A new stream builder.

    • skip

      ProcessorBuilder<T,R> skip(long n)
      Discard the first n of this stream. If this stream contains fewer than n elements, this stream will effectively be an empty stream.

      skip marble diagram

      Specified by:
      skip in interface FilteringOperators<T>
      Parameters:
      n - The number of elements to discard.
      Returns:
      A new stream builder.

    • takeWhile

      ProcessorBuilder<T,R> takeWhile(Predicate<? super R> predicate)
      Take the longest prefix of elements from this stream that satisfy the given predicate.

      takeWhile marble diagram

      When the predicate returns false, the stream will be completed, and upstream will be cancelled.

      Specified by:
      takeWhile in interface FilteringOperators<T>
      Parameters:
      predicate - The predicate.
      Returns:
      A new stream builder.

    • dropWhile

      ProcessorBuilder<T,R> dropWhile(Predicate<? super R> predicate)
      Drop the longest prefix of elements from this stream that satisfy the given predicate.

      dropWhile marble diagram

      As long as the predicate returns true, no elements will be emitted from this stream. Once the first element is encountered for which the predicate returns false, all subsequent elements will be emitted, and the predicate will no longer be invoked.

      Specified by:
      dropWhile in interface FilteringOperators<T>
      Parameters:
      predicate - The predicate.
      Returns:
      A new stream builder.

    • peek

      ProcessorBuilder<T,R> peek(Consumer<? super R> consumer)
      Returns a stream containing all the elements from this stream, additionally performing the provided action on each element.

      peek marbles diagram

      Specified by:
      peek in interface PeekingOperators<T>
      Parameters:
      consumer - The function called for every element.
      Returns:
      A new stream builder that consumes elements of type T and emits the same elements. In between, the given function is called for each element.

    • onError

      ProcessorBuilder<T,R> onError(Consumer<Throwable> errorHandler)
      Returns a stream containing all the elements from this stream, additionally performing the provided action if this stream conveys an error. The given consumer is called with the failure.

      onError marble diagram

      Specified by:
      onError in interface PeekingOperators<T>
      Parameters:
      errorHandler - The function called with the failure.
      Returns:
      A new stream builder that consumes elements of type T and emits the same elements. If the stream conveys a failure, the given error handler is called.

    • onTerminate

      ProcessorBuilder<T,R> onTerminate(Runnable action)
      Returns a stream containing all the elements from this stream, additionally performing the provided action when this stream completes or failed. The given action does not know if the stream failed or completed. If you need to distinguish use PeekingOperators.onError(Consumer) and PeekingOperators.onComplete(Runnable). In addition, the action is called if the stream is cancelled downstream.

      onTerminate marble diagram

      Specified by:
      onTerminate in interface PeekingOperators<T>
      Parameters:
      action - The function called when the stream completes or failed.
      Returns:
      A new stream builder that consumes elements of type T and emits the same elements. The given action is called when the stream completes or fails.

    • onComplete

      ProcessorBuilder<T,R> onComplete(Runnable action)
      Returns a stream containing all the elements from this stream, additionally performing the provided action when this stream completes.

      onComplete marble diagram

      Specified by:
      onComplete in interface PeekingOperators<T>
      Parameters:
      action - The function called when the stream completes.
      Returns:
      A new stream builder that consumes elements of type T and emits the same elements. The given action is called when the stream completes.

    • forEach

      SubscriberBuilder<T,Void> forEach(Consumer<? super R> action)
      Performs an action for each element on this stream.

      forEach marble diagram

      The returned CompletionStage will be redeemed when the stream completes, either successfully if the stream completes normally, or with an error if the stream completes with an error or if the action throws an exception.

      Specified by:
      forEach in interface ConsumingOperators<T>
      Parameters:
      action - The action.
      Returns:
      A SubscriberBuilder that will invoke the action for each element of the stream.

    • ignore

      SubscriberBuilder<T,Void> ignore()
      Ignores each element of this stream.

      ignore marble diagram

      The returned CompletionStage will be redeemed when the stream completes, either successfully if the stream completes normally, or with an error if the stream completes with an error or if the action throws an exception.

      Specified by:
      ignore in interface ConsumingOperators<T>
      Returns:
      A new SubscriberBuilder for the stream.

    • cancel

      SubscriberBuilder<T,Void> cancel()
      Cancels the stream as soon as it is run.

      The returned CompletionStage will be immediately redeemed as soon as the stream is run.

      Specified by:
      cancel in interface ConsumingOperators<T>
      Returns:
      A new SubscriberBuilder for the stream.

    • reduce

      SubscriberBuilder<T,R> reduce(R identity, BinaryOperator<R> accumulator)
      Perform a reduction on the elements of this stream, using the provided identity value and the accumulation function.

      reduce marble diagram

      The result of the reduction is returned in the CompletionStage.

      Specified by:
      reduce in interface ConsumingOperators<T>
      Parameters:
      identity - The identity value.
      accumulator - The accumulator function.
      Returns:
      A new SubscriberBuilder for the reduction.

    • reduce

      SubscriberBuilder<T,Optional<R>> reduce(BinaryOperator<R> accumulator)
      Perform a reduction on the elements of this stream, using the provided accumulation function.

      reduce marble diagram

      The result of the reduction is returned as an Optional<T> in the CompletionStage. If there are no elements in this stream, empty will be returned.

      Specified by:
      reduce in interface ConsumingOperators<T>
      Parameters:
      accumulator - The accumulator function.
      Returns:
      A new SubscriberBuilder for the reduction.

    • collect

      <S, A> SubscriberBuilder<T,S> collect(Collector<? super R,A,S> collector)
      Collect the elements emitted by this stream using the given Collector.

      Since Reactive Streams are intrinsically sequential, only the accumulator of the collector will be used, the combiner will not be used.

      Specified by:
      collect in interface ConsumingOperators<T>
      Type Parameters:
      S - The result of the collector.
      A - The accumulator type.
      Parameters:
      collector - The collector to collect the elements.
      Returns:
      A new SubscriberBuilder that will emit the collected result.

    • collect

      <S> SubscriberBuilder<T,S> collect(Supplier<S> supplier, BiConsumer<S,? super R> accumulator)
      Collect the elements emitted by this stream using a Collector built from the given supplier and accumulator.

      collect marble diagram

      Since Reactive Streams are intrinsically sequential, the combiner will not be used. This is why this method does not accept a combiner method.

      Specified by:
      collect in interface ConsumingOperators<T>
      Type Parameters:
      S - The result of the collector.
      Parameters:
      supplier - a function that creates a new result container. It creates objects of type <A>.
      accumulator - an associative, non-interfering, stateless function for incorporating an additional element into a result
      Returns:
      A new SubscriberBuilder that will emit the collected result.

    • toList

      SubscriberBuilder<T,List<R>> toList()
      Collect the elements emitted by this stream into a List.

      toList marble diagram

      Specified by:
      toList in interface ConsumingOperators<T>
      Returns:
      A new SubscriberBuilder that will emit the list.

    • findFirst

      SubscriberBuilder<T,Optional<R>> findFirst()
      Find the first element emitted by the Publisher, and return it in a CompletionStage.

      findFirst marble diagram

      If the stream is completed before a single element is emitted, then Optional.empty() will be emitted.

      Specified by:
      findFirst in interface ConsumingOperators<T>
      Returns:
      A new SubscriberBuilder.

    • onErrorResume

      ProcessorBuilder<T,R> onErrorResume(Function<Throwable,? extends R> errorHandler)
      Returns a stream containing all the elements from this stream. Additionally, in the case of failure, it invokes the given function and emits the result as final event of the stream.

      onErrorResume marble diagram

      By default, when a stream encounters an error that prevents it from emitting the expected item to its subscriber, the stream invokes its subscriber's onError method, and then terminates without invoking any more of its subscriber's methods. This operator changes this behavior. If the current stream encounters an error, instead of invoking its subscriber's onError method, it will instead emit the return value of the passed function. This operator prevents errors from propagating or to supply fallback data should errors be encountered.

      Specified by:
      onErrorResume in interface ErrorHandlingOperators<T>
      Parameters:
      errorHandler - the function returning the value that needs to be emitting instead of the error. The function must not return null.
      Returns:
      The new stream builder.

    • onErrorResumeWith

      ProcessorBuilder<T,R> onErrorResumeWith(Function<Throwable,? extends PublisherBuilder<? extends R>> errorHandler)
      Returns a stream containing all the elements from this stream. Additionally, in the case of failure, it invokes the given function and emits the elements from the returned PublisherBuilder instead.

      onErrorResumeWith marble diagram

      By default, when a stream encounters an error that prevents it from emitting the expected item to its subscriber, the stream invokes its subscriber's onError method, and then terminates without invoking any more of its subscriber's methods. This operator changes this behavior. If the current stream encounters an error, instead of invoking its subscriber's onError method, it will instead relinquish control to the PublisherBuilder returned from the given function, which invokes the subscriber's onNext method if it is able to do so. The subscriber's original Subscription is used to control the flow of elements both before and after any error occurring. In such a case, because no publisher necessarily invokes onError on the stream's subscriber, it may never know that an error happened.

      Specified by:
      onErrorResumeWith in interface ErrorHandlingOperators<T>
      Parameters:
      errorHandler - the function returning the stream that needs to be emitting instead of the error. The function must not return null.
      Returns:
      The new stream builder.

    • onErrorResumeWithRsPublisher

      ProcessorBuilder<T,R> onErrorResumeWithRsPublisher(Function<Throwable,? extends org.reactivestreams.Publisher<? extends R>> errorHandler)
      Returns a stream containing all the elements from this stream. Additionally, in the case of failure, it invokes the given function and emits the elements from the returned Publisher instead.

      onErrorResumeWithRsPublisher marble diagram

      By default, when a stream encounters an error that prevents it from emitting the expected item to its subscriber, the stream invokes its subscriber's onError method, and then terminates without invoking any more of its subscriber's methods. This operator changes this behavior. If the current stream encounters an error, instead of invoking its subscriber's onError method, the subscriber will be fed from the Publisher returned from the given function, and the subscriber's onNext method is called as the returned Publisher publishes. The subscriber's original Subscription is used to control the flow of both the original and the onError Publishers' elements. In such a case, because no publisher necessarily invokes onError, the subscriber may never know that an error happened.

      Specified by:
      onErrorResumeWithRsPublisher in interface ErrorHandlingOperators<T>
      Parameters:
      errorHandler - the function returning the stream that need to be emitting instead of the error. The function must not return null.
      Returns:
      The new stream builder.

    • to

      SubscriberBuilder<T,Void> to(org.reactivestreams.Subscriber<? super R> subscriber)
      Connect the outlet of the Publisher built by this builder to the given Subscriber. The Reactive Streams specification states that a subscriber should cancel any new stream subscription it receives if it already has an active subscription. The returned result of this method is a stream that creates a subscription for the subscriber passed in, so the resulting stream should only be run once. For the same reason, the subscriber passed in should not have any active subscriptions and should not be used in more than one call to this method.
      Specified by:
      to in interface ConnectingOperators<T>
      Parameters:
      subscriber - The subscriber to connect.
      Returns:
      A completion builder that completes when the stream completes.

    • to

      <S> SubscriberBuilder<T,S> to(SubscriberBuilder<? super R,? extends S> subscriber)
      Connect the outlet of this stream to the given SubscriberBuilder graph. The Reactive Streams specification states that a subscriber should cancel any new stream subscription it receives if it already has an active subscription. For this reason, a subscriber builder, particularly any that represents a graph that includes a user supplied Subscriber or Processor stage, should not be used in the creation of more than one stream instance.
      Specified by:
      to in interface ConnectingOperators<T>
      Parameters:
      subscriber - The subscriber builder to connect.
      Returns:
      A completion builder that completes when the stream completes.

    • via

      <S> ProcessorBuilder<T,S> via(ProcessorBuilder<? super R,? extends S> processor)
      Connect the outlet of the Publisher built by this builder to the given ProcessorBuilder. The Reactive Streams specification states that a subscribing processor should cancel any new stream subscription it receives if it already has an active subscription. For this reason, a processor builder, particularly any that represents a graph that includes a user supplied Processor stage, should not be used in the creation of more than one stream instance.
      Specified by:
      via in interface ConnectingOperators<T>
      Parameters:
      processor - The processor builder to connect.
      Returns:
      A stream builder that represents the passed in processor's outlet.

    • via

      <S> ProcessorBuilder<T,S> via(org.reactivestreams.Processor<? super R,? extends S> processor)
      Connect the outlet of this stream to the given Processor. The Reactive Streams specification states that a subscribing processor should cancel any new stream subscription it receives if it already has an active subscription. The returned result of this method is a stream that creates a subscription for the processor passed in, so the resulting stream should only be run once. For the same reason, the processor passed in should not have any active subscriptions and should not be used in more than one call to this method.
      Specified by:
      via in interface ConnectingOperators<T>
      Parameters:
      processor - The processor builder to connect.
      Returns:
      A stream builder that represents the passed in processor builder's outlet.

    • buildRs

      org.reactivestreams.Processor<T,R> buildRs()
      Build this stream, using the first ReactiveStreamsEngine found by the ServiceLoader.
      Returns:
      A Processor that will run this stream.

    • buildRs

      org.reactivestreams.Processor<T,R> buildRs(ReactiveStreamsEngine engine)
      Build this stream, using the supplied ReactiveStreamsEngine.
      Parameters:
      engine - The engine to run the stream with.
      Returns:
      A Processor that will run this stream.