reactor.core.composable

Class Stream<T>

java.lang.Object

reactor.core.composable.Composable<T>

reactor.core.composable.Stream<T>

Type Parameters:

T - the type of the values in the stream

All Implemented Interfaces:

Flushable<T>, Pipeline<T>

public class Stream<T>
extends Composable<T>

A Stream is a stateless event processor that provides methods for attaching Consumers to consume the values passing through the stream. Some methods like map(reactor.function.Function), filter(reactor.function.Predicate), first(), last(), and reduce(reactor.function.Function, Object) return new Stream Streams whose values are the result of transformation functions, filtering, and the like.

Typically, new Stream Streams aren't created directly. To create a Stream, create a DeferredStreamSpec and configure it with the appropriate Environment, Dispatcher, and other settings, then call Deferred.compose(), which will return a Stream that handles the values passed into the Deferred.

Field Summary

Fields

Modifier and Type	Field and Description
protected int	batchSize

Fields inherited from class reactor.core.composable.Composable

END_EVENT

Fields inherited from interface reactor.core.action.Flushable

FLUSH_EVENT

Constructor Summary

Constructors

Constructor and Description
Stream(Observable observable, int batchSize, Composable<?> parent, Environment environment) Create a new Stream that will use the Observable to pass its values to registered handlers.
Stream(Observable observable, int batchSize, Composable<?> parent, Tuple2<Selector,java.lang.Object> acceptSelector, Environment environment) Create a new Stream that will use the Observable to pass its values to registered handlers.

Method Summary

Modifier and Type	Method and Description
Stream<T>	buffer() Use Observable.batchNotify(Object) to group dispatching by the current batch size.
Stream<T>	buffer(int batchSize) Use Observable.batchNotify(Object) to group dispatch by the passed .
Stream<T>	bufferWithErrors() Collect incoming values into a List that will be pushed separately into the returned Stream every time batchSize has been reached.
Stream<T>	bufferWithErrors(int batchSize) Collect incoming values into a List that will be pushed separately into the returned Stream every time batchSize has been reached.
Stream<java.util.List<T>>	collect() Collect incoming values into a List that will be pushed into the returned Stream every time batchSize or flush is triggered has been reached.
Stream<java.util.List<T>>	collect(int batchSize) Collect incoming values into a List that will be pushed into the returned Stream every time batchSize has been reached.
Stream<T>	connect(Composable<T> consumer) Attach another Composable to this one that will cascade the value or error received by this Composable into the next.
Stream<T>	connectErrors(Composable<?> composable) Forward any error to the argument.
Stream<T>	connectValues(Composable<T> composable) Attach another Composable to this one that will only cascade the value received by this Composable into the next.
Stream<T>	consume(Consumer<T> consumer) Attach a Consumer to this Composable that will consume any values accepted by this Composable.
Stream<T>	consume(java.lang.Object key, Observable observable) Pass values accepted by this Composable into the given Observable, notifying with the given key.
Stream<T>	consumeFlush(Flushable<?> consumer) Consume flush with the passed Flushable
Stream<T>	count(Stream<java.lang.Long> stream) Count accepted events for each batch (every flush) and pass each accumulated long to the .
Stream<T>	distinct() Create a new Stream that filters out consecutive equals values.
Stream<java.lang.Boolean>	filter() Evaluate each accepted boolean value.
Stream<java.lang.Boolean>	filter(Composable<java.lang.Boolean> elseComposable) Evaluate each accepted boolean value.
Stream<T>	filter(Function<T,java.lang.Boolean> p) Evaluate each accepted value against the given predicate Function.
Stream<T>	filter(Predicate<T> p) Evaluate each accepted value against the given Predicate.
Stream<T>	filter(Predicate<T> p, Composable<T> elseComposable) Evaluate each accepted value against the given Predicate.
Stream<T>	first() Create a new Stream whose values will be only the first value of each batch.
Stream<T>	first(int batchSize) Create a new Stream whose values will be only the first value of each batch.
Stream<T>	flush() Flush any cached or unprocessed values through this Stream.
Stream<T>	flushWhen(Predicate<T> predicate) Consume values and trigger flush when matches.
Stream<T>	last() Create a new Stream whose values will be only the last value of each batch.
Stream<T>	last(int batchSize) Create a new Stream whose values will be only the last value of each batch.
<V> Stream<V>	map(Function<T,V> fn) Assign the given Function to transform the incoming value T into a V and pass it into another Composable.
<V,C extends Composable<V>> Stream<V>	mapMany(Function<T,C> fn) Assign the given Function to transform the incoming value T into a Composable<V> and pass it into another Composable.
Stream<T>	merge(Composable<T>... composables) this#connect(Composable) all the passed to this Composable, merging values streams into the current pipeline.
Stream<java.util.List<T>>	movingWindow(int period, int backlog) Collect incoming values into an internal array, providing a List that will be pushed into the returned Stream every specified in milliseconds.
Stream<java.util.List<T>>	movingWindow(int period, java.util.concurrent.TimeUnit timeUnit, int backlog) Collect incoming values into an internal array, providing a List that will be pushed into the returned Stream every specified time from the and a .
Stream<java.util.List<T>>	movingWindow(int period, java.util.concurrent.TimeUnit timeUnit, int delay, int backlog) Collect incoming values into an internal array, providing a List that will be pushed into the returned Stream every specified time from the and a after an initial in milliseconds.
Stream<java.util.List<T>>	movingWindow(int period, java.util.concurrent.TimeUnit timeUnit, int delay, int backlog, Timer timer) Collect incoming values into an internal array, providing a List that will be pushed into the returned Stream every specified time from the and a after an initial in milliseconds.
protected <V> Stream<V>	newComposable() Create a Composable that is compatible with the subclass of Composable in use.
protected <V> Stream<V>	newComposable(int batchSize)
Stream<T>	propagate(java.lang.Iterable<T> iterable) Create a new Stream whose values will be each iterated item from Every time flush is triggered, is drained.
Stream<T>	propagate(Supplier<T> supplier) Create a new Composable whose values will be generated from .
<A> Stream<A>	reduce(Function<Tuple2<T,A>,A> fn) Reduce the values passing through this Stream into an object A.
<A> Stream<A>	reduce(Function<Tuple2<T,A>,A> fn, A initial) Reduce the values passing through this Stream into an object A.
<A> Stream<A>	reduce(Function<Tuple2<T,A>,A> fn, Supplier<A> accumulators, int batchSize) Reduce the values passing through this Stream into an object A.
<A> Stream<A>	scan(Function<Tuple2<T,A>,A> fn) Scan the values passing through this Stream into an object A.
<A> Stream<A>	scan(Function<Tuple2<T,A>,A> fn, A initial) Scan the values passing through this Stream into an object A.
<A> Stream<A>	scan(Function<Tuple2<T,A>,A> fn, Supplier<A> accumulators) Scan the values passing through this Stream into an object A.
<V> Stream<V>	split() Create a new Stream whose values will be each element E of any Iterable flowing this Stream
<V> Stream<V>	split(int batchSize) Create a new Stream whose values will be each element E of any Iterable flowing this Stream
Tap<T>	tap() Create a Tap that maintains a reference to the last value seen by this Stream.
Stream<T>	timeout(long timeout) Flush the parent if any or the current composable otherwise when the last notification occurred before milliseconds.
Stream<T>	timeout(long timeout, Timer timer) Flush the parent if any or the current composable otherwise when the last notification occurred before milliseconds.
<E extends java.lang.Throwable> Stream<T>	when(java.lang.Class<E> exceptionType, Consumer<E> onError) Assign an error handler to exceptions of the given type.
Stream<java.util.List<T>>	window(int period) Collect incoming values into a List that will be pushed into the returned Stream every specified time from the in milliseconds.
Stream<java.util.List<T>>	window(int period, java.util.concurrent.TimeUnit timeUnit) Collect incoming values into a List that will be pushed into the returned Stream every specified time from the and a .
Stream<java.util.List<T>>	window(int period, java.util.concurrent.TimeUnit timeUnit, int delay) Collect incoming values into a List that will be pushed into the returned Stream every specified time from the , after an initial in milliseconds.
Stream<java.util.List<T>>	window(int period, java.util.concurrent.TimeUnit timeUnit, int delay, Timer timer) Collect incoming values into a List that will be pushed into the returned Stream every specified time from the , after an initial in milliseconds.

Methods inherited from class reactor.core.composable.Composable

add, consumeErrorAndFlush, consumeEvent, debug, getAcceptKey, getAcceptSelector, getEnvironment, getError, getFlush, getObservable, getParent

Methods inherited from class java.lang.Object

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

Field Detail

batchSize

protected final int batchSize

Constructor Detail

Stream

public Stream(@Nullable
              Observable observable,
              int batchSize,
              @Nullable
              Composable<?> parent,
              @Nullable
              Environment environment)

Create a new Stream that will use the Observable to pass its values to registered handlers.

The stream will batch values into batches of the given batchSize, affecting the values that are passed to the first() and last() substreams. A size of -1 indicates that the stream should not be batched.

Once all event handler have been registered, flush() must be called to pass the initial values to those handlers. The stream will accept errors from the given parent.

Parameters:

observable - The observable used to drive event handlers

batchSize - The size of the batches, or -1 for no batching

parent - The stream's parent. May be null

Stream

public Stream(@Nullable
              Observable observable,
              int batchSize,
              @Nullable
              Composable<?> parent,
              Tuple2<Selector,java.lang.Object> acceptSelector,
              @Nullable
              Environment environment)

Create a new Stream that will use the Observable to pass its values to registered handlers.

The stream will batch values into batches of the given batchSize, affecting the values that are passed to the first() and last() substreams. A size of -1 indicates that the stream should not be batched.

Once all event handler have been registered, flush() must be called to pass the initial values to those handlers. The stream will accept errors from the given parent.

Parameters:

observable - The observable used to drive event handlers

batchSize - The size of the batches, or -1 for no batching

parent - The stream's parent. May be null

acceptSelector - The tuple Selector/Key to accept values on this observable. May be null

Method Detail

consume

public Stream<T> consume(@Nonnull
                         Consumer<T> consumer)

Description copied from class: Composable

Attach a Consumer to this Composable that will consume any values accepted by this Composable.

Overrides:

consume in class Composable<T>

Parameters:

consumer - the conumer to invoke on each value

Returns:

this

connect

public Stream<T> connect(@Nonnull
                         Composable<T> consumer)

Description copied from class: Composable

Attach another Composable to this one that will cascade the value or error received by this Composable into the next.

Overrides:

connect in class Composable<T>

Parameters:

consumer - the next Composable to cascade events to

Returns:

this

consume

public Stream<T> consume(@Nonnull
                         java.lang.Object key,
                         @Nonnull
                         Observable observable)

Description copied from class: Composable

Pass values accepted by this Composable into the given Observable, notifying with the given key.

Overrides:

consume in class Composable<T>

Parameters:

key - the key to notify on

observable - the Observable to notify

Returns:

this

flush

public Stream<T> flush()

Description copied from class: Composable

Flush any cached or unprocessed values through this Stream.

Specified by:

flush in interface Flushable<T>

Overrides:

flush in class Composable<T>

Returns:

this

when

public <E extends java.lang.Throwable> Stream<T> when(@Nonnull
                                                      java.lang.Class<E> exceptionType,
                                                      @Nonnull
                                                      Consumer<E> onError)

Description copied from class: Composable

Assign an error handler to exceptions of the given type.

Overrides:

when in class Composable<T>

Type Parameters:

E - type of the exception to handle

Parameters:

exceptionType - the type of exceptions to handle

onError - the error handler for each exception

Returns:

this

map

public <V> Stream<V> map(@Nonnull
                         Function<T,V> fn)

Description copied from class: Composable

Assign the given Function to transform the incoming value T into a V and pass it into another Composable.

Overrides:

map in class Composable<T>

Type Parameters:

V - the type of the return value of the transformation function

Parameters:

fn - the transformation function

Returns:

a new Composable containing the transformed values

mapMany

public <V,C extends Composable<V>> Stream<V> mapMany(@Nonnull
                                                     Function<T,C> fn)

Description copied from class: Composable

Assign the given Function to transform the incoming value T into a Composable<V> and pass it into another Composable.

Overrides:

mapMany in class Composable<T>

Type Parameters:

V - the type of the return value of the transformation function

Parameters:

fn - the transformation function

Returns:

a new Composable containing the transformed values

filter

public Stream<java.lang.Boolean> filter()

Description copied from class: Composable

Evaluate each accepted boolean value. If the predicate test succeeds, the value is passed into the new Composable. If the predicate test fails, the value is ignored.

Overrides:

filter in class Composable<T>

Returns:

a new Composable containing only values that pass the predicate test

filter

public Stream<java.lang.Boolean> filter(@Nonnull
                                        Composable<java.lang.Boolean> elseComposable)

Description copied from class: Composable

Evaluate each accepted boolean value. If the predicate test succeeds, the value is passed into the new Composable. the value is propagated into the .

Overrides:

filter in class Composable<T>

Parameters:

elseComposable - the Composable to test values against

Returns:

a new Composable containing only values that pass the predicate test

filter

public Stream<T> filter(@Nonnull
                        Function<T,java.lang.Boolean> p)

Description copied from class: Composable

Evaluate each accepted value against the given predicate Function. If the predicate test succeeds, the value is passed into the new Composable. If the predicate test fails, an exception is propagated into the new Composable.

Overrides:

filter in class Composable<T>

Parameters:

p - the predicate Function to test values against

Returns:

a new Composable containing only values that pass the predicate test

filter

public Stream<T> filter(@Nonnull
                        Predicate<T> p)

Description copied from class: Composable

Evaluate each accepted value against the given Predicate. If the predicate test succeeds, the value is passed into the new Composable. If the predicate test fails, the value is ignored.

Overrides:

filter in class Composable<T>

Parameters:

p - the Predicate to test values against

Returns:

a new Composable containing only values that pass the predicate test

filter

public Stream<T> filter(@Nonnull
                        Predicate<T> p,
                        Composable<T> elseComposable)

Description copied from class: Composable

Evaluate each accepted value against the given Predicate. If the predicate test succeeds, the value is passed into the new Composable. If the predicate test fails, the value is propagated into the elseComposable.

Overrides:

filter in class Composable<T>

Parameters:

p - the Predicate to test values against

elseComposable - the optional Composable to pass rejected values

Returns:

a new Composable containing only values that pass the predicate test

connectValues

public Stream<T> connectValues(@Nonnull
                               Composable<T> composable)

Description copied from class: Composable

Attach another Composable to this one that will only cascade the value received by this Composable into the next.

Overrides:

connectValues in class Composable<T>

Parameters:

composable - the next Composable to cascade events to

Returns:

this

timeout

public Stream<T> timeout(long timeout,
                         Timer timer)

Description copied from class: Composable

Flush the parent if any or the current composable otherwise when the last notification occurred before milliseconds. Timeout is run on the provided .

Overrides:

timeout in class Composable<T>

Parameters:

timeout - the timeout in milliseconds between two notifications on this composable

timer - the reactor timer to run the timeout on

Returns:

this Composable

timeout

public Stream<T> timeout(long timeout)

Description copied from class: Composable

Flush the parent if any or the current composable otherwise when the last notification occurred before milliseconds. Timeout is run on the environment root timer.

Overrides:

timeout in class Composable<T>

Parameters:

timeout - the timeout in milliseconds between two notifications on this composable

Returns:

this Composable

merge

public Stream<T> merge(Composable<T>... composables)

Description copied from class: Composable

this#connect(Composable) all the passed to this Composable, merging values streams into the current pipeline.

Overrides:

merge in class Composable<T>

Parameters:

composables - the the composables to connect

Returns:

this composable

consumeFlush

public Stream<T> consumeFlush(@Nonnull
                              Flushable<?> consumer)

Description copied from interface: Pipeline

Consume flush with the passed Flushable

Specified by:

consumeFlush in interface Pipeline<T>

Overrides:

consumeFlush in class Composable<T>

Parameters:

consumer - the action listening for flush

Returns:

this

connectErrors

public Stream<T> connectErrors(Composable<?> composable)

Description copied from class: Composable

Forward any error to the argument.

Overrides:

connectErrors in class Composable<T>

Parameters:

composable - the target sink for errores and flushes

Returns:

this

propagate

public Stream<T> propagate(java.lang.Iterable<T> iterable)

Create a new Stream whose values will be each iterated item from Every time flush is triggered, is drained.

Parameters:

iterable - the iterable to drain

Returns:

a new Stream whose values are the iterated one on flush

Since:

1.1

propagate

public Stream<T> propagate(Supplier<T> supplier)

Description copied from class: Composable

Create a new Composable whose values will be generated from . Every time flush is triggered, is called.

Overrides:

propagate in class Composable<T>

Parameters:

supplier - the supplier to drain

Returns:

a new Composable whose values are generated on each flush

flushWhen

public Stream<T> flushWhen(Predicate<T> predicate)

Consume values and trigger flush when matches.

Parameters:

predicate - the test returning true to trigger flush

Returns:

the current Stream

Since:

1.1

first

public Stream<T> first()

Create a new Stream whose values will be only the first value of each batch. Requires a batchSize to have been set.

When a new batch is triggered, the first value of that next batch will be pushed into this Stream.

Returns:

a new Stream whose values are the first value of each batch

first

public Stream<T> first(int batchSize)

Create a new Stream whose values will be only the first value of each batch. Requires a batchSize to have been set.

When a new batch is triggered, the first value of that next batch will be pushed into this Stream.

Parameters:

batchSize - the batch size to use

Returns:

a new Stream whose values are the first value of each batch)

last

public Stream<T> last()

Create a new Stream whose values will be only the last value of each batch. Requires a batchSize

Returns:

a new Stream whose values are the last value of each batch

last

public Stream<T> last(int batchSize)

Create a new Stream whose values will be only the last value of each batch. Requires a batchSize

Parameters:

batchSize - the batch size to use

Returns:

a new Stream whose values are the last value of each batch

distinct

public Stream<T> distinct()

Create a new Stream that filters out consecutive equals values.

Returns:

a new Stream whose values are the last value of each batch

Since:

1.1

split

public <V> Stream<V> split()

Create a new Stream whose values will be each element E of any Iterable flowing this Stream

When a new batch is triggered, the last value of that next batch will be pushed into this Stream.

Returns:

a new Stream whose values result from the iterable input

Since:

1.1

split

public <V> Stream<V> split(int batchSize)

Create a new Stream whose values will be each element E of any Iterable flowing this Stream

When a new batch is triggered, the last value of that next batch will be pushed into this Stream.

Parameters:

batchSize - the batch size to use

Returns:

a new Stream whose values result from the iterable input

Since:

1.1

tap

public Tap<T> tap()

Create a Tap that maintains a reference to the last value seen by this Stream. The Tap is continually updated when new values pass through the Stream.

Returns:

the new Tap

See Also:

Consumer

buffer

public Stream<T> buffer()

Use Observable.batchNotify(Object) to group dispatching by the current batch size. When a new batch is triggered, the last value of that next batch will be pushed into this Stream.

Returns:

a new Stream whose values result from the iterable input

Since:

1.1

buffer

public Stream<T> buffer(int batchSize)

Use Observable.batchNotify(Object) to group dispatch by the passed . When a new batch is triggered, the last value of that next batch will be pushed into this Stream.

Parameters:

batchSize - the batch size to use

Returns:

a new Stream whose values result from the iterable input

Since:

1.1

bufferWithErrors

public Stream<T> bufferWithErrors()

Collect incoming values into a List that will be pushed separately into the returned Stream every time batchSize has been reached. All errors are also captured until current batchSize or flush is called.

Returns:

a new Stream whose values of all values in this batch

Since:

1.1

bufferWithErrors

public Stream<T> bufferWithErrors(int batchSize)

Collect incoming values into a List that will be pushed separately into the returned Stream every time batchSize has been reached. All errors are also captured until batchSize or flush is called.

Parameters:

batchSize - the collected size

Returns:

a new Stream whose values are all values in this batch

Since:

1.1

collect

public Stream<java.util.List<T>> collect()

Collect incoming values into a List that will be pushed into the returned Stream every time batchSize or flush is triggered has been reached.

Returns:

a new Stream whose values are a List of all values in this batch

collect

public Stream<java.util.List<T>> collect(int batchSize)

Collect incoming values into a List that will be pushed into the returned Stream every time batchSize has been reached.

Parameters:

batchSize - the collected size

Returns:

a new Stream whose values are a List of all values in this batch

window

public Stream<java.util.List<T>> window(int period)

Collect incoming values into a List that will be pushed into the returned Stream every specified time from the in milliseconds. The window runs on a timer from the stream this#environment.

Parameters:

period - the time period when each window close and flush the attached consumer

Returns:

a new Stream whose values are a List of all values in this window

Since:

1.1

movingWindow

public Stream<java.util.List<T>> movingWindow(int period,
                                              int backlog)

Collect incoming values into an internal array, providing a List that will be pushed into the returned Stream every specified in milliseconds. The window runs on a timer from the stream this#environment. After accepting of items, every old item will be dropped. Resulting List will be at most items long.

Parameters:

period - the time period when each window close and flush the attached consumer

backlog - maximum amount of items to keep

Returns:

a new Stream whose values are a List of all values in this window

Since:

1.1

window

public Stream<java.util.List<T>> window(int period,
                                        java.util.concurrent.TimeUnit timeUnit)

Collect incoming values into a List that will be pushed into the returned Stream every specified time from the and a . The window runs on a timer from the stream this#environment.

Parameters:

period - the time period when each window close and flush the attached consumer

timeUnit - the time unit used for the period

Returns:

a new Stream whose values are a List of all values in this window

Since:

1.1

movingWindow

public Stream<java.util.List<T>> movingWindow(int period,
                                              java.util.concurrent.TimeUnit timeUnit,
                                              int backlog)

Collect incoming values into an internal array, providing a List that will be pushed into the returned Stream every specified time from the and a .

Parameters:

period - the time period when each window close and flush the attached consumer

timeUnit - the time unit used for the period

backlog - maximum amount of items to keep

Returns:

a new Stream whose values are a List of all values in this window

Since:

1.1

window

public Stream<java.util.List<T>> window(int period,
                                        java.util.concurrent.TimeUnit timeUnit,
                                        int delay)

Collect incoming values into a List that will be pushed into the returned Stream every specified time from the , after an initial in milliseconds. The window runs on a timer from the stream this#environment.

Parameters:

period - the time period when each window close and flush the attached consumer

timeUnit - the time unit used for the period

delay - the initial delay in milliseconds

Returns:

a new Stream whose values are a List of all values in this window

Since:

1.1

movingWindow

public Stream<java.util.List<T>> movingWindow(int period,
                                              java.util.concurrent.TimeUnit timeUnit,
                                              int delay,
                                              int backlog)

Collect incoming values into an internal array, providing a List that will be pushed into the returned Stream every specified time from the and a after an initial in milliseconds.

Parameters:

period - the time period when each window close and flush the attached consumer

timeUnit - the time unit used for the period

delay - the initial delay in milliseconds

backlog - maximum amount of items to keep

Returns:

a new Stream whose values are a List of all values in this window

Since:

1.1

window

public Stream<java.util.List<T>> window(int period,
                                        java.util.concurrent.TimeUnit timeUnit,
                                        int delay,
                                        Timer timer)

Collect incoming values into a List that will be pushed into the returned Stream every specified time from the , after an initial in milliseconds. The window runs on a supplied .

Parameters:

period - the time period when each window close and flush the attached consumer

timeUnit - the time unit used for the period

delay - the initial delay in milliseconds

timer - the reactor timer to run the window on

Returns:

a new Stream whose values are a List of all values in this window

Since:

1.1

movingWindow

public Stream<java.util.List<T>> movingWindow(int period,
                                              java.util.concurrent.TimeUnit timeUnit,
                                              int delay,
                                              int backlog,
                                              Timer timer)

Collect incoming values into an internal array, providing a List that will be pushed into the returned Stream every specified time from the and a after an initial in milliseconds.

Parameters:

period - the time period when each window close and flush the attached consumer

timeUnit - the time unit used for the period

delay - the initial delay in milliseconds

backlog - maximum amount of items to keep

timer - the reactor timer to run the window on

Returns:

a new Stream whose values are a List of all values in this window

Since:

1.1

reduce

public <A> Stream<A> reduce(@Nonnull
                            Function<Tuple2<T,A>,A> fn,
                            A initial)

Reduce the values passing through this Stream into an object A. The given initial object will be passed to the function's Tuple2 argument.

Type Parameters:

A - the type of the reduced object

Parameters:

fn - the reduce function

initial - the initial argument to pass to the reduce function

Returns:

a new Stream whose values contain only the reduced objects

reduce

public <A> Stream<A> reduce(@Nonnull
                            Function<Tuple2<T,A>,A> fn,
                            @Nullable
                            Supplier<A> accumulators,
                            int batchSize)

Reduce the values passing through this Stream into an object A. The given Supplier will be used to produce initial accumulator objects either on the first reduce call, in the case of an unbounded Stream, or on the first value of each batch, if a batchSize is set.

In an unbounded Stream, the accumulated value will be published on the returned Stream on flush only. But when a batchSize has been, the accumulated value will only be published on the new Stream at the end of each batch. On the next value (the first of the next batch), the Supplier is called again for a new accumulator object and the reduce starts over with a new accumulator.

Type Parameters:

A - the type of the reduced object

Parameters:

fn - the reduce function

accumulators - the Supplier that will provide accumulators

batchSize - the batch size to use

Returns:

a new Stream whose values contain only the reduced objects

reduce

public <A> Stream<A> reduce(@Nonnull
                            Function<Tuple2<T,A>,A> fn)

Reduce the values passing through this Stream into an object A.

Type Parameters:

A - the type of the reduced object

Parameters:

fn - the reduce function

Returns:

a new Stream whose values contain only the reduced objects

scan

public <A> Stream<A> scan(@Nonnull
                          Function<Tuple2<T,A>,A> fn,
                          A initial)

Scan the values passing through this Stream into an object A. The given initial object will be passed to the function's Tuple2 argument. Behave like Reduce but triggers downstream Stream for every transformation.

Type Parameters:

A - the type of the reduced object

Parameters:

fn - the scan function

initial - the initial argument to pass to the reduce function

Returns:

a new Stream whose values contain only the reduced objects

Since:

1.1

scan

public <A> Stream<A> scan(@Nonnull
                          Function<Tuple2<T,A>,A> fn,
                          @Nullable
                          Supplier<A> accumulators)

Scan the values passing through this Stream into an object A. The given Supplier will be used to produce initial accumulator objects either on the first reduce call, in the case of an unbounded Stream, or on the first value of each batch, if a batchSize is set.

The accumulated value will be published on the returned Stream every time a value is accepted.

Type Parameters:

A - the type of the reduced object

Parameters:

fn - the scan function

accumulators - the Supplier that will provide accumulators

Returns:

a new Stream whose values contain only the reduced objects

Since:

1.1

scan

public <A> Stream<A> scan(@Nonnull
                          Function<Tuple2<T,A>,A> fn)

Scan the values passing through this Stream into an object A.

Type Parameters:

A - the type of the reduced object

Parameters:

fn - the reduce function

Returns:

a new Stream whose values contain only the reduced objects

Since:

1.1

count

public Stream<T> count(Stream<java.lang.Long> stream)

Count accepted events for each batch (every flush) and pass each accumulated long to the .

Parameters:

stream - the stream to consume accumulated number of accepted event between 2 flushes

Since:

1.1

newComposable

protected <V> Stream<V> newComposable()

Description copied from class: Composable

Create a Composable that is compatible with the subclass of Composable in use.

Specified by:

newComposable in class Composable<T>

Type Parameters:

V - type the Composable handles

Returns:

a new Composable compatible with the current subclass.

newComposable

protected <V> Stream<V> newComposable(int batchSize)