com.aol.cyclops.control

Class SimpleReact

java.lang.Object

com.aol.cyclops.control.SimpleReact

All Implemented Interfaces:

ReactBuilder

public class SimpleReact
extends java.lang.Object
implements ReactBuilder

Builder class for SimpleReact Stream types SimpleReact streams are finite eager parallel Streams with a concise API. Awesome for loading discrete batches of files or remote resources. Useful for doing shortcircuiting querying operations against remote services. Confgure Executors Parallelism / concurrent tasks Caching Object pooling E.g implementing a Quorum

  
   new SimpleReact().react(this:query,this:query,this:query,this:query)
                .then(this:process)
                .block(status -> status.getAllCompleted() >2 && status.getElapsedMillis()>200);

     //short circuit if 2 results after 200ms 
 
 
 
 

E.g. loading files

  
   List<File> files;
   
   new SimpleReact().from(files)
                    .thenAsync(FileUtils::load)
                      .then(this:process)
                      .block();

 
 
 
 

In general if you have a small discrete data sets SimpleReact may be a fit. If you need infinite / continuous Stream processing & more advanced features use LazyReact (Even flatMap is relatively limited in SimpleReact Stream - for advanced operations @see LazyReact)

Constructor Summary

Constructors

Constructor and Description
SimpleReact() Construct a SimpleReact builder using standard thread pool.
SimpleReact(java.util.concurrent.Executor executor)
SimpleReact(java.util.concurrent.Executor queueService, java.util.concurrent.Executor executor, com.nurkiewicz.asyncretry.RetryExecutor retrier, java.lang.Boolean async)
SimpleReact(java.util.concurrent.Executor executor, com.nurkiewicz.asyncretry.RetryExecutor retrier) Construct a SimpleReact builder from the provided Executor, Retrier.
SimpleReact(java.util.concurrent.Executor executor, com.nurkiewicz.asyncretry.RetryExecutor retrier, java.lang.Boolean async) Construct a SimpleReact builder from the provided Executor, Retrier.
SimpleReact(java.util.concurrent.Executor executor, com.nurkiewicz.asyncretry.RetryExecutor retrier, java.util.concurrent.Executor queueCopier)

Method Summary

Modifier and Type	Method and Description
<U> SimpleReactStream<U>	construct(java.util.stream.Stream s)
<R> SimpleReactStream<R>	from(java.util.Collection<R> collection) Start a reactive flow from a Collection using an Iterator
<U> SimpleReactStream<U>	from(java.util.concurrent.CompletableFuture<U>... cf) Construct a simpleReactStream from an Array of CompletableFutures
<U> SimpleReactStream<U>	from(java.util.concurrent.CompletableFuture<U> cf)
<U> SimpleReactStream<java.lang.Double>	from(java.util.stream.DoubleStream stream) Start a reactive dataflow from a stream.
<U> SimpleReactStream<java.lang.Integer>	from(java.util.stream.IntStream stream) Start a reactive dataflow from a stream.
<U> SimpleReactStream<U>	from(java.util.Iterator<U> iterator) Start a reactive flow from a JDK Iterator
<U> SimpleReactStream<java.lang.Long>	from(java.util.stream.LongStream stream) Start a reactive dataflow from a stream.
<U> SimpleReactStream<U>	from(java.util.stream.Stream<U> stream) Start a reactive dataflow from a stream.
<U> SimpleReactStream<U>	fromIterable(java.lang.Iterable<U> iter) Start a reactive flow from a JDK Iterator
<U> SimpleReactStream<U>	fromIterableAsync(java.lang.Iterable<? extends java.util.function.Supplier<U>> actions) Start a reactive dataflow with a list of one-off-suppliers
<U> SimpleReactStream<U>	fromIteratorAsync(java.util.Iterator<? extends java.util.function.Supplier<U>> actions) Start a reactive dataflow with a list of one-off-suppliers
<T> SimpleReactStream<T>	fromPublisher(org.reactivestreams.Publisher<? extends T> publisher) Construct a SimpleReactStream from an Publisher
<U> SimpleReactStream<U>	fromStream(java.util.stream.Stream<java.util.concurrent.CompletableFuture<U>> stream) Start a reactive dataflow from a stream of CompletableFutures.
<U> SimpleReactStream<U>	fromStreamAsync(java.util.stream.Stream<? extends java.util.function.Supplier<U>> actions) Start a reactive dataflow with a list of one-off-suppliers
boolean	isAsync()
<U> SimpleReactStream<U>	of(U... array)
<U> SimpleReactStream<U>	ofAsync(java.util.function.Supplier<U>... actions) Start a reactive dataflow with an array of one-off-suppliers
static SimpleReact	parallelBuilder()
static SimpleReact	parallelBuilder(int parallelism) Construct a new SimpleReact builder, with a new task executor and retry executor with configured number of threads
static SimpleReact	parallelCommonBuilder()
SimpleReactStream<java.lang.Integer>	range(int startInclusive, int endExclusive)
static SimpleReact	sequentialBuilder()
static SimpleReact	sequentialCommonBuilder()
SimpleReact	withQueueCopyExecutor(java.util.concurrent.Executor queueCopyExecutor)

Methods inherited from class java.lang.Object

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

Constructor Detail

SimpleReact

public SimpleReact()

Construct a SimpleReact builder using standard thread pool. By default, unless ThreadPools is configured otherwise this will be sized to the available processors

See Also:

ThreadPools.getStandard()

SimpleReact

public SimpleReact(java.util.concurrent.Executor executor,
                   com.nurkiewicz.asyncretry.RetryExecutor retrier,
                   java.lang.Boolean async)

Construct a SimpleReact builder from the provided Executor, Retrier.

Parameters:

executor - Task executor to execute tasks on

retrier - Retrier to use for asyncrhonous retry

async - If false, subsequent tasks are executed on the completing thread If true each subsequent task is resubmitted to a task executor,

SimpleReact

public SimpleReact(java.util.concurrent.Executor executor)

Parameters:

executor - Executor this SimpleReact instance will use to execute concurrent tasks.

SimpleReact

public SimpleReact(java.util.concurrent.Executor executor,
                   com.nurkiewicz.asyncretry.RetryExecutor retrier)

Construct a SimpleReact builder from the provided Executor, Retrier.

Parameters:

executor - Task executor to execute tasks on

retrier - Retrier to use for asyncrhonous retry

SimpleReact

public SimpleReact(java.util.concurrent.Executor executor,
                   com.nurkiewicz.asyncretry.RetryExecutor retrier,
                   java.util.concurrent.Executor queueCopier)

Parameters:

executor - Task executor to execute tasks on

retrier -

queueCopier - Task executor to transfer results during flatMap operations

SimpleReact

public SimpleReact(java.util.concurrent.Executor queueService,
                   java.util.concurrent.Executor executor,
                   com.nurkiewicz.asyncretry.RetryExecutor retrier,
                   java.lang.Boolean async)

Method Detail

construct

public <U> SimpleReactStream<U> construct(java.util.stream.Stream s)

withQueueCopyExecutor

public SimpleReact withQueueCopyExecutor(java.util.concurrent.Executor queueCopyExecutor)

fromStreamAsync

public <U> SimpleReactStream<U> fromStreamAsync(java.util.stream.Stream<? extends java.util.function.Supplier<U>> actions)

Start a reactive dataflow with a list of one-off-suppliers

Parameters:

actions - Stream of Suppliers to provide data (and thus events) that downstream jobs will react too

Returns:

Next stage in the reactive flow

fromPublisher

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

Construct a SimpleReactStream from an Publisher

Parameters:

publisher - to construct SimpleReactStream from

Returns:

SimpleReactStream

fromIteratorAsync

public <U> SimpleReactStream<U> fromIteratorAsync(java.util.Iterator<? extends java.util.function.Supplier<U>> actions)

Start a reactive dataflow with a list of one-off-suppliers

Parameters:

actions - Iterator over Suppliers to provide data (and thus events) that downstream jobs will react too

Returns:

Next stage in the reactive flow

fromIterableAsync

public <U> SimpleReactStream<U> fromIterableAsync(java.lang.Iterable<? extends java.util.function.Supplier<U>> actions)

Start a reactive dataflow with a list of one-off-suppliers

Parameters:

actions - Stream of Suppliers to provide data (and thus events) that downstream jobs will react too

Returns:

Next stage in the reactive flow

ofAsync

@SafeVarargs
public final <U> SimpleReactStream<U> ofAsync(java.util.function.Supplier<U>... actions)

Start a reactive dataflow with an array of one-off-suppliers

Parameters:

actions - Array of Suppliers to provide data (and thus events) that downstream jobs will react too

Returns:

Next stage in the reactive flow

from

public <U> SimpleReactStream<U> from(java.util.stream.Stream<U> stream)

Start a reactive dataflow from a stream.

Parameters:

stream - that will be used to drive the reactive dataflow

Returns:

Next stage in the reactive flow

from

public <R> SimpleReactStream<R> from(java.util.Collection<R> collection)

Start a reactive flow from a Collection using an Iterator

Parameters:

collection - - Collection SimpleReact will iterate over at the start of the flow

Returns:

Next stage in the reactive flow

isAsync

public boolean isAsync()

parallelBuilder

public static SimpleReact parallelBuilder()

Returns:

An Eager SimpleReact instance

See Also:

SimpleReact()

parallelBuilder

public static SimpleReact parallelBuilder(int parallelism)

Construct a new SimpleReact builder, with a new task executor and retry executor with configured number of threads

Parameters:

parallelism - Number of threads task executor should have

Returns:

eager SimpleReact instance

parallelCommonBuilder

public static SimpleReact parallelCommonBuilder()

Returns:

new eager SimpleReact builder configured with standard parallel executor By default this is the ForkJoinPool common instance but is configurable in the ThreadPools class

See Also:

see RetryBuilder#getDefaultInstance()

sequentialBuilder

public static SimpleReact sequentialBuilder()

Returns:

new eager SimpleReact builder configured to run on a separate thread (non-blocking current thread), sequentially New ForkJoinPool will be created

sequentialCommonBuilder

public static SimpleReact sequentialCommonBuilder()

Returns:

new eager SimpleReact builder configured to run on a separate thread (non-blocking current thread), sequentially Common free thread Executor from

range

public SimpleReactStream<java.lang.Integer> range(int startInclusive,
                                                  int endExclusive)

from

public <U> SimpleReactStream<U> from(java.util.Iterator<U> iterator)

Start a reactive flow from a JDK Iterator

Parameters:

iterator - SimpleReact will iterate over this iterator concurrently to start the reactive dataflow

Returns:

Next stage in the reactive flow

fromIterable

public <U> SimpleReactStream<U> fromIterable(java.lang.Iterable<U> iter)

Start a reactive flow from a JDK Iterator

Parameters:

iter - SimpleReact will iterate over this iterator concurrently to start the reactive dataflow

Returns:

Next stage in the reactive flow

fromStream

public <U> SimpleReactStream<U> fromStream(java.util.stream.Stream<java.util.concurrent.CompletableFuture<U>> stream)

Start a reactive dataflow from a stream of CompletableFutures.

Parameters:

stream - of CompletableFutures that will be used to drive the reactive dataflow

Returns:

Next stage in the reactive flow

from

public <U> SimpleReactStream<java.lang.Integer> from(java.util.stream.IntStream stream)

Start a reactive dataflow from a stream.

Parameters:

stream - that will be used to drive the reactive dataflow

Returns:

Next stage in the reactive flow

from

public <U> SimpleReactStream<java.lang.Double> from(java.util.stream.DoubleStream stream)

Start a reactive dataflow from a stream.

Parameters:

stream - that will be used to drive the reactive dataflow

Returns:

Next stage in the reactive flow

from

public <U> SimpleReactStream<java.lang.Long> from(java.util.stream.LongStream stream)

Start a reactive dataflow from a stream.

Parameters:

stream - that will be used to drive the reactive dataflow

Returns:

Next stage in the reactive flow

of

public <U> SimpleReactStream<U> of(U... array)

from

public <U> SimpleReactStream<U> from(java.util.concurrent.CompletableFuture<U> cf)

from

public <U> SimpleReactStream<U> from(java.util.concurrent.CompletableFuture<U>... cf)

Construct a simpleReactStream from an Array of CompletableFutures

Parameters:

cf - CompletableFutures to turn into a Stream

Returns:

SimpleReactStream from an Array of CompletableFutures