Composable (reactor-core 1.1.6.RELEASE API)

java.lang.Object
- reactor.core.composable.Composable<T>

Type Parameters:

T - The type of the values

All Implemented Interfaces:

Flushable<T>, Pipeline<T>

Direct Known Subclasses:

Promise, Stream
```
public abstract class Composable<T>
extends java.lang.Object
implements Pipeline<T>
```
Abstract base class for components designed to provide a succinct API for working with future values. Provides base functionality and an internal contract for subclasses that make use of the map(reactor.function.Function) and filter(reactor.function.Predicate) methods.

Field Summary

Fields
Modifier and Type Field and Description

static Event<java.lang.Object> END_EVENT
- Fields inherited from interface reactor.core.action.Flushable
  FLUSH_EVENT

Fields
Modifier and Type	Field and Description
`static Event<java.lang.Object>`	`END_EVENT`

Constructor Summary

Constructors
Modifier	Constructor and Description
`protected`	`Composable(Observable observable, Composable<U> parent)`
`protected`	`Composable(Observable observable, Composable<U> parent, Tuple2<Selector,java.lang.Object> acceptSelectorTuple, Environment environment)`

Method Summary

All Methods Instance Methods Abstract Methods Concrete Methods
Modifier and Type	Method and Description
`Composable<T>`	`add(Action<T> action)` Consume events with the passed `Action`
`Composable<T>`	`connect(Composable<T> composable)` Attach another `Composable` to this one that will cascade the value or error received by this `Composable` into the next.
`Composable<T>`	`connectErrors(Composable<?> composable)` Forward any error to the argument.
`Composable<T>`	`connectValues(Composable<T> composable)` Attach another `Composable` to this one that will only cascade the value received by this `Composable` into the next.
`Composable<T>`	`consume(Consumer<T> consumer)` Attach a `Consumer` to this `Composable` that will consume any values accepted by this `Composable`.
`Composable<T>`	`consume(java.lang.Object key, Observable observable)` Pass values accepted by this `Composable` into the given `Observable`, notifying with the given key.
`protected Composable<T>`	`consumeErrorAndFlush(Composable<?> composable)` Forward any error or flush to the argument.
`Composable<T>`	`consumeEvent(Consumer<Event<T>> consumer)` Attach a `Consumer` to this `Composable` that will consume any values accepted by this `Composable`.
`Composable<T>`	`consumeFlush(Flushable<?> action)` Consume flush with the passed `Flushable`
`java.lang.String`	`debug()` Print a debugged form of the root composable relative to this.
`Composable<java.lang.Boolean>`	`filter()` Evaluate each accepted boolean value.
`Composable<java.lang.Boolean>`	`filter(Composable<java.lang.Boolean> elseComposable)` Evaluate each accepted boolean value.
`Composable<T>`	`filter(Function<T,java.lang.Boolean> fn)` Evaluate each accepted value against the given predicate `Function`.
`Composable<T>`	`filter(Predicate<T> p)` Evaluate each accepted value against the given `Predicate`.
`Composable<T>`	`filter(Predicate<T> p, Composable<T> elseComposable)` Evaluate each accepted value against the given `Predicate`.
`Composable<T>`	`flush()` Flush any cached or unprocessed values through this Stream.
`protected java.lang.Object`	`getAcceptKey()` Get the anonymous `Selector` and notification key for doing accepts.
`protected Selector`	`getAcceptSelector()` Get the anonymous `Selector` and notification key for doing accepts.
`protected Environment`	`getEnvironment()` Get the assigned `Environment`.
`protected Selector`	`getError()` Get the anonymous `Selector` and notification key for doing errors.
`protected Selector`	`getFlush()` Get the anonymous flush `Selector` for batch consuming.
`protected Observable`	`getObservable()` Get the current `Observable`.
`protected Composable<?>`	`getParent()` Get the parent `Composable` for callback callback.
`<V> Composable<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>> Composable<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`.
`Composable<T>`	`merge(Composable<T>... composables)` `this#connect(Composable)` all the passed to this `Composable`, merging values streams into the current pipeline.
`protected abstract <V> Composable<V>`	`newComposable()` Create a `Composable` that is compatible with the subclass of `Composable` in use.
`Composable<T>`	`propagate(Supplier<T> supplier)` Create a new `Composable` whose values will be generated from .
`Composable<T>`	`timeout(long timeout)` Flush the parent if any or the current composable otherwise when the last notification occurred before milliseconds.
`Composable<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> Composable<T>`	`when(java.lang.Class<E> exceptionType, Consumer<E> onError)` Assign an error handler to exceptions of the given type.

Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

- Field Detail
  - END_EVENT
```
public static final Event<java.lang.Object> END_EVENT
```
- Constructor Detail
  - Composable
```
protected Composable(@Nullable
                     Observable observable,
                     @Nullable
                     Composable<U> parent)
```
  - Composable
```
protected Composable(@Nullable
                     Observable observable,
                     @Nullable
                     Composable<U> parent,
                     @Nullable
                     Tuple2<Selector,java.lang.Object> acceptSelectorTuple,
                     @Nullable
                     Environment environment)
```
- Method Detail
  - when
```
public <E extends java.lang.Throwable> Composable<T> when(@Nonnull
                                                          java.lang.Class<E> exceptionType,
                                                          @Nonnull
                                                          Consumer<E> onError)
```
    Assign an error handler to exceptions of the given type.
    
    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
  - connect
```
public Composable<T> connect(@Nonnull
                             Composable<T> composable)
```
    Attach another Composable to this one that will cascade the value or error received by this Composable into the next.
    
    Parameters:
    
    composable - the next Composable to cascade events to
    
    Returns:
    
    this
    
    Since:
    
    1.1
  - connectValues
```
public Composable<T> connectValues(@Nonnull
                                   Composable<T> composable)
```
    Attach another Composable to this one that will only cascade the value received by this Composable into the next.
    
    Parameters:
    
    composable - the next Composable to cascade events to
    
    Returns:
    
    this
  - consume
```
public Composable<T> consume(@Nonnull
                             Consumer<T> consumer)
```
    Attach a Consumer to this Composable that will consume any values accepted by this Composable.
    
    Parameters:
    
    consumer - the conumer to invoke on each value
    
    Returns:
    
    this
  - consumeEvent
```
public Composable<T> consumeEvent(@Nonnull
                                  Consumer<Event<T>> consumer)
```
    Attach a Consumer to this Composable that will consume any values accepted by this Composable.
    
    Parameters:
    
    consumer - the conumer to invoke on each value
    
    Returns:
    
    this
  - consume
```
public Composable<T> consume(@Nonnull
                             java.lang.Object key,
                             @Nonnull
                             Observable observable)
```
    Pass values accepted by this Composable into the given Observable, notifying with the given key.
    
    Parameters:
    
    key - the key to notify on
    
    observable - the Observable to notify
    
    Returns:
    
    this
  - map
```
public <V> Composable<V> map(@Nonnull
                             Function<T,V> fn)
```
    Assign the given Function to transform the incoming value T into a V and pass it into another Composable.
    
    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>> Composable<V> mapMany(@Nonnull
                                                         Function<T,C> fn)
```
    Assign the given Function to transform the incoming value T into a Composable<V> and pass it into another Composable.
    
    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
    
    Since:
    
    1.1
  - merge
```
public Composable<T> merge(Composable<T>... composables)
```
    this#connect(Composable) all the passed to this Composable, merging values streams into the current pipeline.
    
    Parameters:
    
    composables - the the composables to connect
    
    Returns:
    
    this composable
    
    Since:
    
    1.1
  - filter
```
public Composable<T> filter(@Nonnull
                            Function<T,java.lang.Boolean> fn)
```
    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.
    
    Parameters:
    
    fn - the predicate Function to test values against
    
    Returns:
    
    a new Composable containing only values that pass the predicate test
  - filter
```
public Composable<java.lang.Boolean> filter()
```
    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.
    
    Returns:
    
    a new Composable containing only values that pass the predicate test
    
    Since:
    
    1.1
  - filter
```
public Composable<java.lang.Boolean> filter(@Nonnull
                                            Composable<java.lang.Boolean> elseComposable)
```
    Evaluate each accepted boolean value. If the predicate test succeeds, the value is passed into the new Composable. the value is propagated into the .
    
    Parameters:
    
    elseComposable - the Composable to test values against
    
    Returns:
    
    a new Composable containing only values that pass the predicate test
    
    Since:
    
    1.1
  - filter
```
public Composable<T> filter(@Nonnull
                            Predicate<T> p)
```
    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.
    
    Parameters:
    
    p - the Predicate to test values against
    
    Returns:
    
    a new Composable containing only values that pass the predicate test
  - filter
```
public Composable<T> filter(@Nonnull
                            Predicate<T> p,
                            Composable<T> elseComposable)
```
    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.
    
    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
  - timeout
```
public Composable<T> timeout(long timeout)
```
    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.
    
    Parameters:
    
    timeout - the timeout in milliseconds between two notifications on this composable
    
    Returns:
    
    this Composable
    
    Since:
    
    1.1
  - timeout
```
public Composable<T> timeout(long timeout,
                             Timer timer)
```
    Flush the parent if any or the current composable otherwise when the last notification occurred before milliseconds. Timeout is run on the provided .
    
    Parameters:
    
    timeout - the timeout in milliseconds between two notifications on this composable
    
    timer - the reactor timer to run the timeout on
    
    Returns:
    
    this Composable
    
    Since:
    
    1.1
  - propagate
```
public Composable<T> propagate(Supplier<T> supplier)
```
    Create a new Composable whose values will be generated from . Every time flush is triggered, is called.
    
    Parameters:
    
    supplier - the supplier to drain
    
    Returns:
    
    a new Composable whose values are generated on each flush
    
    Since:
    
    1.1
  - flush
```
public Composable<T> flush()
```
    Flush any cached or unprocessed values through this Stream.
    
    Specified by:
    
    flush in interface Flushable<T>
    
    Returns:
    
    this
  - debug
```
public java.lang.String debug()
```
    Print a debugged form of the root composable relative to this. The output will be an acyclic directed graph of composed actions.
    
    Since:
    
    1.1
  - add
```
public Composable<T> add(Action<T> action)
```
    Consume events with the passed Action
    
    Specified by:
    
    add in interface Pipeline<T>
    
    Parameters:
    
    action - the action listening for values
    
    Returns:
    
    this
    
    Since:
    
    1.1
  - consumeFlush
```
public Composable<T> consumeFlush(Flushable<?> action)
```
    Description copied from interface: Pipeline
    
    Consume flush with the passed Flushable
    
    Specified by:
    
    consumeFlush in interface Pipeline<T>
    
    Parameters:
    
    action - the action listening for flush
    
    Returns:
    
    this
  - connectErrors
```
public Composable<T> connectErrors(Composable<?> composable)
```
    Forward any error to the argument.
    
    Parameters:
    
    composable - the target sink for errores and flushes
    
    Returns:
    
    this
    
    Since:
    
    1.1
  - consumeErrorAndFlush
```
protected Composable<T> consumeErrorAndFlush(Composable<?> composable)
```
    Forward any error or flush to the argument.
    
    Parameters:
    
    composable - the target sink for errores and flushes
    
    Returns:
    
    this
    
    Since:
    
    1.1
  - newComposable
```
protected abstract <V> Composable<V> newComposable()
```
    Create a Composable that is compatible with the subclass of Composable in use.
    
    Type Parameters:
    
    V - type the Composable handles
    
    Returns:
    
    a new Composable compatible with the current subclass.
  - getObservable
```
protected Observable getObservable()
```
    Get the current Observable.
    
    Returns:
  - getAcceptKey
```
protected java.lang.Object getAcceptKey()
```
    Get the anonymous Selector and notification key for doing accepts.
    
    Returns:
  - getAcceptSelector
```
protected Selector getAcceptSelector()
```
    Get the anonymous Selector and notification key for doing accepts.
    
    Returns:
  - getFlush
```
protected Selector getFlush()
```
    Get the anonymous flush Selector for batch consuming.
    
    Returns:
  - getError
```
protected Selector getError()
```
    Get the anonymous Selector and notification key for doing errors.
    
    Returns:
  - getParent
```
protected Composable<?> getParent()
```
    Get the parent Composable for callback callback.
    
    Returns:
  - getEnvironment
```
protected Environment getEnvironment()
```
    Get the assigned Environment.
    
    Returns:

Class Composable<T>

Field Summary

Fields inherited from interface reactor.core.action.Flushable

Constructor Summary

Method Summary

Methods inherited from class java.lang.Object

Field Detail

END_EVENT

Constructor Detail

Composable

Composable

Method Detail

when

connect

connectValues

consume

consumeEvent

consume

map

mapMany

merge

filter

filter

filter

filter

filter

timeout

timeout

propagate

flush

debug

add

consumeFlush

connectErrors

consumeErrorAndFlush

newComposable

getObservable

getAcceptKey

getAcceptSelector

getFlush

getError

getParent

getEnvironment