com.aol.cyclops.control

Class ReactorPipes<K,V>

java.lang.Object

com.aol.cyclops.control.ReactorPipes<K,V>

public class ReactorPipes<K,V>
extends java.lang.Object

Store for Pipes for cross-thread communication Connected Streams will not be able to complete collect or reduce style methods unless the underlying Adapter for data transfer is closed. I.e. connected Streams remain connected until either the Adapter is closed, or they disconnect (due to a limit for example).

Constructor Summary

Constructors

Constructor and Description
ReactorPipes()

Method Summary

Modifier and Type	Method and Description
void	clear()
void	close(java.lang.String key)
Maybe<reactor.core.publisher.Flux<V>>	flux(K key)
Maybe<LazyFutureStream<V>>	futureStream(K key)
Maybe<LazyFutureStream<V>>	futureStream(K key, LazyReact reactor)
Maybe<Adapter<V>>	get(K key)
Eval<V>	nextOrNull(K key) Return an Eval that allows retrieval of the next value from the attached pipe when get() is called A value is returned if a value is present, otherwise null is returned if the publisher is complete or an error occurs
Eval<Maybe<V>>	nextValue(K key) Return an Eval that allows retrieval of the next value from the attached pipe when get() is called Maybe.some is returned if a value is present, Maybe.none is returned if the publisher is complete or an error occurs
static <K,V> ReactorPipes<K,V>	of()
static <K,V> ReactorPipes<K,V>	of(java.util.Map<K,Adapter<V>> registered)
Xor<java.lang.Throwable,V>	oneOrError(K key)
reactor.core.publisher.Mono<V>	oneOrErrorAsync(K key, java.util.concurrent.Executor ex)
Maybe<V>	oneValue(K key) Extract one value from the selected pipe, if it exists
Maybe<Try<V,java.lang.Throwable>>	oneValueOrError(K key)
<X extends java.lang.Throwable> Maybe<Try<V,X>>	oneValueOrError(K key, java.lang.Class<X>... classes)
void	publishTo(K key, org.reactivestreams.Publisher<V> publisher)
void	publishToAsync(K key, org.reactivestreams.Publisher<V> publisher)
void	push(K key, V value) Push the supplied value through the Adapter identified by the supplied key
Maybe<ReactiveSeq<V>>	reactiveSeq(K key)
void	register(K key, Adapter<V> adapter) Register a Queue, and get back a listening LazyFutureStream that runs on a single thread (not the calling thread)
PMapX<K,Adapter<V>>	registered()
int	size()
void	subscribeTo(K key, org.reactivestreams.Subscriber<V> subscriber) Subscribe synchronously to a pipe
void	subscribeTo(K key, org.reactivestreams.Subscriber<V> subscriber, java.util.concurrent.Executor subscribeOn) Subscribe asynchronously to a pipe
LazyListX<V>	xValues(K key, long x)

Methods inherited from class java.lang.Object

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

Constructor Detail

ReactorPipes

public ReactorPipes()

Method Detail

size

public int size()

Returns:

Size of registered pipes

registered

public PMapX<K,Adapter<V>> registered()

Returns:

Persistent map of all registered pipes

of

public static <K,V> ReactorPipes<K,V> of()

Returns:

Construct an empty Pipes instance

of

public static <K,V> ReactorPipes<K,V> of(java.util.Map<K,Adapter<V>> registered)

push

public void push(K key,
                 V value)

Push the supplied value through the Adapter identified by the supplied key

  
 
  Queue<String> q = new Queue<>();
        pipes.register("hello", q);
        pipes.push("hello", "world");
        pipes.push("hello", "world2");
        q.close();
        pipes.oneOrErrorAsync("hello", ex).block(); //"world"
               
 

Parameters:

key - Adapter key

value - Value to push

get

public Maybe<Adapter<V>> get(K key)

Parameters:

key - : Adapter identifier

Returns:

selected Queue

futureStream

public Maybe<LazyFutureStream<V>> futureStream(K key)

Parameters:

key - : Adapter identifier

Returns:

LazyFutureStream from selected Queue

futureStream

public Maybe<LazyFutureStream<V>> futureStream(K key,
                                               LazyReact reactor)

Parameters:

key - : Adapter identifier

Returns:

LazyFutureStream from selected Queue

reactiveSeq

public Maybe<ReactiveSeq<V>> reactiveSeq(K key)

Parameters:

key - : Adapter identifier

Returns:

LazyFutureStream from selected Queue

flux

public Maybe<reactor.core.publisher.Flux<V>> flux(K key)

Parameters:

key - : Adapter identifier

Returns:

LazyFutureStream from selected Queue

xValues

public LazyListX<V> xValues(K key,
                            long x)

oneValue

public Maybe<V> oneValue(K key)

Extract one value from the selected pipe, if it exists

Parameters:

key - : Adapter identifier

Returns:

LazyFutureStream from selected Queue

oneOrError

public Xor<java.lang.Throwable,V> oneOrError(K key)

oneValueOrError

public <X extends java.lang.Throwable> Maybe<Try<V,X>> oneValueOrError(K key,
                                                                       java.lang.Class<X>... classes)

oneValueOrError

public Maybe<Try<V,java.lang.Throwable>> oneValueOrError(K key)

oneOrErrorAsync

public reactor.core.publisher.Mono<V> oneOrErrorAsync(K key,
                                                      java.util.concurrent.Executor ex)

nextValue

public Eval<Maybe<V>> nextValue(K key)

Return an Eval that allows retrieval of the next value from the attached pipe when get() is called Maybe.some is returned if a value is present, Maybe.none is returned if the publisher is complete or an error occurs

Parameters:

key -

Returns:

nextOrNull

public Eval<V> nextOrNull(K key)

Return an Eval that allows retrieval of the next value from the attached pipe when get() is called A value is returned if a value is present, otherwise null is returned if the publisher is complete or an error occurs

Parameters:

key -

Returns:

register

public void register(K key,
                     Adapter<V> adapter)

Register a Queue, and get back a listening LazyFutureStream that runs on a single thread (not the calling thread)

 
 Pipes.register("test", QueueFactories.
                                            <String>boundedNonBlockingQueue(100)
                                                .build());
        LazyFutureStream<String> stream =  PipesToLazyStreams.cpuBoundStream("test");
        stream.filter(it->it!=null).peek(System.out::println).run();
 
 

Parameters:

key - : Adapter identifier

adapter -

clear

public void clear()

subscribeTo

public void subscribeTo(K key,
                        org.reactivestreams.Subscriber<V> subscriber)

Subscribe synchronously to a pipe

Parameters:

key - for registered simple-react async.Adapter

subscriber - Reactive Streams subscriber for data on this pipe

subscribeTo

public void subscribeTo(K key,
                        org.reactivestreams.Subscriber<V> subscriber,
                        java.util.concurrent.Executor subscribeOn)

Subscribe asynchronously to a pipe

Parameters:

key - for registered simple-react async.Adapter

subscriber - Reactive Streams subscriber for data on this pipe

publishTo

public void publishTo(K key,
                      org.reactivestreams.Publisher<V> publisher)

Parameters:

key - for registered simple-react async.Adapter

publisher - Reactive Streams publisher to push data onto this pipe

publishToAsync

public void publishToAsync(K key,
                           org.reactivestreams.Publisher<V> publisher)

Parameters:

key - for registered simple-react async.Adapter

publisher - Reactive Streams publisher to push data onto this pipe

close

public void close(java.lang.String key)

Parameters:

key - Close the Adapter defined by the supplied key