org.eclipse.microprofile.reactive.streams.operators

## Class ReactiveStreams

- java.lang.Object
- - org.eclipse.microprofile.reactive.streams.operators.ReactiveStreams

    public class ReactiveStreams
    extends Object

Primary entry point into the Reactive Streams utility API.

This class provides factory methods for publisher and processor builders, which can then be subsequently manipulated using their respective APIs.

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.

### Method Summary

All Methods Static Methods Concrete Methods
`static <T> ProcessorBuilder<T,T>` `builder()` Create a `ProcessorBuilder`.
`static <T> PublisherBuilder<T>` `concat(PublisherBuilder<? extends T> a, PublisherBuilder<? extends T> b)` Concatenates two publishers.
`static <T,R> ProcessorBuilder<T,R>` `coupled(org.reactivestreams.Subscriber<? super T> subscriber, org.reactivestreams.Publisher<? extends R> publisher)` Creates a `ProcessorBuilder` by coupling a `Subscriber` to a `Publisher`.
`static <T,R> ProcessorBuilder<T,R>` `coupled(SubscriberBuilder<? super T,?> subscriber, PublisherBuilder<? extends R> publisher)` Creates a `ProcessorBuilder` by coupling a `SubscriberBuilder` to a `PublisherBuilder`.
`static <T> PublisherBuilder<T>` `empty()` Create an empty `PublisherBuilder`.
`static <T> PublisherBuilder<T>` `failed(Throwable t)` Create a failed `PublisherBuilder`.
`static <T> PublisherBuilder<T>` `fromCompletionStage(CompletionStage<? extends T> completionStage)` Creates a publisher from a `CompletionStage`.
`static <T> PublisherBuilder<T>` `fromCompletionStageNullable(CompletionStage<? extends T> completionStage)` Creates a publisher from a `CompletionStage`.
`static <T> PublisherBuilder<T>` `fromIterable(Iterable<? extends T> ts)` Create a `PublisherBuilder` that will emits the elements produced by the passed in `Iterable`.
`static <T,R> ProcessorBuilder<T,R>` `fromProcessor(org.reactivestreams.Processor<? super T,? extends R> processor)` Create a `ProcessorBuilder` from the given `Processor`.
`static <T> PublisherBuilder<T>` `fromPublisher(org.reactivestreams.Publisher<? extends T> publisher)` Create a `PublisherBuilder` from the given `Publisher`.
`static <T> SubscriberBuilder<T,Void>` `fromSubscriber(org.reactivestreams.Subscriber<? super T> subscriber)` Create a `SubscriberBuilder` from the given `Subscriber`.
`static <T> PublisherBuilder<T>` `generate(Supplier<? extends T> s)` Creates an infinite stream that emits elements supplied by the supplier `s`.
`static <T> PublisherBuilder<T>` `iterate(T seed, UnaryOperator<T> f)` Creates an infinite stream produced by the iterative application of the function `f` to an initial element `seed` consisting of `seed`, `f(seed)`, `f(f(seed))`, etc.
`static <T> PublisherBuilder<T>` `of(T... ts)` Create a `PublisherBuilder` that emits the given elements.
`static <T> PublisherBuilder<T>` `of(T t)` Create a `PublisherBuilder` that emits a single element.
`static <T> PublisherBuilder<T>` `ofNullable(T t)` Create a `PublisherBuilder` that will emit a single element if `t` is not null, otherwise will be empty.

### Methods inherited from class java.lang.Object

`clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait`

### Method Detail

#### fromPublisher

    public static <T> PublisherBuilder<T> fromPublisher(org.reactivestreams.Publisher<? extends T> publisher)

Create a `PublisherBuilder` from the given `Publisher`.

Type Parameters:
  `T` - The type of the elements that the publisher produces.
Parameters:
  `publisher` - The publisher to wrap.
Returns:
  A publisher builder that wraps the given publisher.

#### of

    public static <T> PublisherBuilder<T> of(T t)

Create a `PublisherBuilder` that emits a single element.

Type Parameters:
  `T` - The type of the element.
Parameters:
  `t` - The element to emit.
Returns:
  A publisher builder that will emit the element.

#### of

    public static <T> PublisherBuilder<T> of(T... ts)

Create a `PublisherBuilder` that emits the given elements.

Type Parameters:
  `T` - The type of the elements.
Parameters:
  `ts` - The elements to emit.
Returns:
  A publisher builder that will emit the elements.

#### empty

    public static <T> PublisherBuilder<T> empty()

Create an empty `PublisherBuilder`.

Type Parameters:
  `T` - The type of the publisher builder.
Returns:
  A publisher builder that will just emit a completion signal.

#### ofNullable

    public static <T> PublisherBuilder<T> ofNullable(T t)

Create a `PublisherBuilder` that will emit a single element if `t` is not null, otherwise will be empty.

Type Parameters:
  `T` - The type of the element.
Parameters:
  `t` - The element to emit, `null` if to element should be emitted.
Returns:
  A publisher builder that optionally emits a single element.

#### fromIterable

    public static <T> PublisherBuilder<T> fromIterable(Iterable<? extends T> ts)

Create a `PublisherBuilder` that will emits the elements produced by the passed in `Iterable`.

Type Parameters:
  `T` - The type of the elements.
Parameters:
  `ts` - The elements to emit.
Returns:
  A publisher builder that emits the elements of the iterable.

#### failed

    public static <T> PublisherBuilder<T> failed(Throwable t)

Create a failed `PublisherBuilder`.

This publisher will just emit an error.

Type Parameters:
  `T` - The type of the publisher builder.
Parameters:
  `t` - The error te emit.
Returns:
  A publisher builder that completes the stream with an error.

#### builder

    public static <T> ProcessorBuilder<T,T> builder()

Create a `ProcessorBuilder`. This builder will start as an identity processor.

Type Parameters:
  `T` - The type of elements that the processor consumes and emits.
Returns:
  The identity processor builder.

#### fromProcessor

    public static <T,R> ProcessorBuilder<T,R> fromProcessor(org.reactivestreams.Processor<? super T,? extends R> processor)

Create a `ProcessorBuilder` from the given `Processor`.

Type Parameters:
  `T` - The type of the elements that the processor consumes.
  `R` - The type of the elements that the processor emits.
Parameters:
  `processor` - The processor to be wrapped.
Returns:
  A processor builder that wraps the processor.

#### fromSubscriber

    public static <T> SubscriberBuilder<T,Void> fromSubscriber(org.reactivestreams.Subscriber<? super T> subscriber)

Create a `SubscriberBuilder` from the given `Subscriber`. The subscriber can only be used to create a single subscriber builder.

Type Parameters:
  `T` - The type of elements that the subscriber consumes.
Parameters:
  `subscriber` - The subscriber to be wrapped.
Returns:
  A subscriber builder that wraps the subscriber.

#### iterate

    public static <T> PublisherBuilder<T> iterate(T seed,
                                                  UnaryOperator<T> f)

Creates an infinite stream produced by the iterative application of the function `f` to an initial element `seed` consisting of `seed`, `f(seed)`, `f(f(seed))`, etc.

Type Parameters:
  `T` - The type of stream elements.
Parameters:
  `seed` - The initial element.
  `f` - A function applied to the previous element to produce the next element.
Returns:
  A publisher builder.

#### generate

    public static <T> PublisherBuilder<T> generate(Supplier<? extends T> s)

Creates an infinite stream that emits elements supplied by the supplier `s`.

Type Parameters:
  `T` - The type of stream elements.
Parameters:
  `s` - The supplier.
Returns:
  A publisher builder.

#### concat

    public static <T> PublisherBuilder<T> concat(PublisherBuilder<? extends T> a,
                                                 PublisherBuilder<? extends T> b)

Concatenates two publishers.

The resulting stream will be produced by subscribing to the first publisher, and emitting the elements it emits, until it emits a completion signal, at which point the second publisher will be subscribed to, and its elements will be emitted.

If the first publisher completes with an error signal, then the second publisher will be subscribed to but immediately cancelled, none of its elements will be emitted. This ensures that hot publishers are cleaned up. If downstream emits a cancellation signal before the first publisher finishes, it will be passed to both publishers.

Type Parameters:
  `T` - The type of stream elements.
Parameters:
  `a` - The first publisher.
  `b` - The second publisher.
Returns:
  A publisher builder.

#### fromCompletionStage

    public static <T> PublisherBuilder<T> fromCompletionStage(CompletionStage<? extends T> completionStage)

Creates a publisher from a `CompletionStage`.

When the `CompletionStage` is redeemed, the publisher will emit the redeemed element, and then signal completion. If the completion stage is redeemed with `null`, the stream will be failed with a `NullPointerException`.

If the `CompletionStage` is completed with a failure, this failure will be propagated through the stream.

Type Parameters:
  `T` - The type of the `CompletionStage` value.
Parameters:
  `completionStage` - The `CompletionStage` to create the publisher from.
Returns:
  A `PublisherBuilder` representation of this `CompletionStage`.

#### fromCompletionStageNullable

    public static <T> PublisherBuilder<T> fromCompletionStageNullable(CompletionStage<? extends T> completionStage)

Creates a publisher from a `CompletionStage`.

When the `CompletionStage` is redeemed, the publisher will emit the redeemed element, and then signal completion. If the completion stage is redeemed with `null`, the stream will be immediately completed with no element, ie, it will be an empty stream.

If the `CompletionStage` is completed with a failure, this failure will be propagated through the stream.

Type Parameters:
  `T` - The type of the `CompletionStage` value.
Parameters:
  `completionStage` - The `CompletionStage` to create the publisher from.
Returns:
  A `PublisherBuilder` representation of this `CompletionStage`.

#### coupled

    public static <T,R> ProcessorBuilder<T,R> coupled(SubscriberBuilder<? super T,?> subscriber,
                                                      PublisherBuilder<? extends R> publisher)

Creates a `ProcessorBuilder` by coupling a `SubscriberBuilder` to a `PublisherBuilder`.

The resulting processor sends all the elements received to the passed in subscriber, and emits all the elements received from the passed in publisher.

In addition, the lifecycles of the subscriber and publisher are coupled, such that if one terminates or receives a termination signal, the other will be terminated. Below is a table of what signals are emited when.

Lifecycle signal propagation
Returned ProcessorBuilder inlet | Passed in SubscriberBuilder | Passed in PublisherBuilder | Returned ProcessorBuilder outlet
--- | --- | --- | ---
Cause: complete from upstream | Effect: complete | Effect: cancel | Effect: complete
Cause: error from upstream | Effect: error | Effect: cancel | Effect: error
Effect: cancel | Cause: cancels | Effect: cancel | Effect: complete
Effect: cancel | Effect: complete | Cause: completes | Effect: complete
Effect: cancel | Effect: error | Cause: errors | Effect: error
Effect: cancel | Effect: complete | Effect: cancel | Cause: cancel from downstream

Type Parameters:
  `T` - The type of elements received.
  `R` - The type of elements emitted.
Parameters:
  `subscriber` - The subscriber builder to wrap.
  `publisher` - The publisher builder to wrap.
Returns:
  The coupled processor builder.

#### coupled

    public static <T,R> ProcessorBuilder<T,R> coupled(org.reactivestreams.Subscriber<? super T> subscriber,
                                                      org.reactivestreams.Publisher<? extends R> publisher)

Creates a `ProcessorBuilder` by coupling a `Subscriber` to a `Publisher`.

The resulting processor sends all the elements received to the passed in subscriber, and emits all the elements received from the passed in publisher.

In addition, the lifecycles of the subscriber and publisher are coupled, such that if one terminates or receives a termination signal, the other will be terminated. Below is a table of what signals are emited when:

Lifecycle signal propagation
Returned ProcessorBuilder inlet | Passed in SubscriberBuilder | Passed in PublisherBuilder | Returned ProcessorBuilder outlet
--- | --- | --- | ---
Cause: complete from upstream | Effect: complete | Effect: cancel | Effect: complete
Cause: error from upstream | Effect: error | Effect: cancel | Effect: error
Effect: cancel | Cause: cancels | Effect: cancel | Effect: complete
Effect: cancel | Effect: complete | Cause: completes | Effect: complete
Effect: cancel | Effect: error | Cause: errors | Effect: error
Effect: cancel | Effect: complete | Effect: cancel | Cause: cancel from downstream

Type Parameters:
  `T` - The type of elements received.
  `R` - The type of elements emitted.
Parameters:
  `subscriber` - The subscriber builder to wrap.
  `publisher` - The publisher builder to wrap.
Returns:
  The coupled processor builder.
See Also:
  `coupled(SubscriberBuilder, PublisherBuilder)`