com.aol.cyclops.control

Class FluxSource

java.lang.Object

com.aol.cyclops.control.FluxSource

public class FluxSource
extends java.lang.Object

Create Java 8 Streams that data can be pushed into

Method Summary

Modifier and Type	Method and Description
<T> PushableFlux<T>	flux() Create a pushable Flux
static <T> reactor.core.publisher.Flux<T>	flux(Adapter<T> adapter) Create a pushable ReactiveSeq
static <T> LazyFutureStream<T>	futureStream(Adapter<T> adapter, LazyReact react) Create a LazyFutureStream.
<T> PushableLazyFutureStream<T>	futureStream(LazyReact s) Create a pushable LazyFutureStream using the supplied ReactPool
static FluxSource	of(int backPressureAfter) Create a Pushable Flux soure backed by a bounded BlockingQueue.
static FluxSource	of(QueueFactory<?> q) Create a Pushable Flux source backed by a queue created by the supplied queue factory
static <T> MultipleFluxSource<T>	ofMultiple()
static <T> MultipleFluxSource<T>	ofMultiple(int backPressureAfter)
static <T> MultipleFluxSource<T>	ofMultiple(QueueFactory<?> q)
static FluxSource	ofUnbounded() Create a Pushable Flux source backed by an unbounded ConcurrentLinkedQueue
<T> PushableReactiveSeq<T>	reactiveSeq() Create a pushable ReactiveSeq
static <T> ReactiveSeq<T>	reactiveSeq(Adapter<T> adapter) Create a pushable ReactiveSeq
<T> PushableStream<T>	stream() Create a pushable JDK 8 Stream
static <T> java.util.stream.Stream<T>	stream(Adapter<T> adapter) Create a JDK 8 Stream from the supplied Adapter

Methods inherited from class java.lang.Object

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

Method Detail

ofMultiple

public static <T> MultipleFluxSource<T> ofMultiple()

Returns:

a builder that will use Topics to allow multiple Streams from the same data

ofMultiple

public static <T> MultipleFluxSource<T> ofMultiple(int backPressureAfter)

Returns:

a builder that will use Topics to allow multiple Streams from the same data

ofMultiple

public static <T> MultipleFluxSource<T> ofMultiple(QueueFactory<?> q)

Returns:

a builder that will use Topics to allow multiple Streams from the same data

of

public static FluxSource of(QueueFactory<?> q)

Create a Pushable Flux source backed by a queue created by the supplied queue factory

  
  //create a QueueFactory for Agrona OneToOneConcurrentArrayQueue with capacity for 1,000 elements
  
  QueueFactory<Integer> input = QueueFactories.singleWriterboundedNonBlockingQueue(1_000);
 
  FluxSource source = FluxSource.of(input);
  
  PushableFlux flux = source.flux();
  flux.getInput().offer(1);
  
  //on a different thread
  flux.getFlux()
      .map(i->i*2)
      .subscribe(System.out::println);
 
 

Parameters:

q - Queue Factory used to provide data for the Flux stream

Returns:

FluxSource which manages the input queue and the stream

ofUnbounded

public static FluxSource ofUnbounded()

Create a Pushable Flux source backed by an unbounded ConcurrentLinkedQueue

Returns:

FluxSource that allows data to be pushed into a Flux

of

public static FluxSource of(int backPressureAfter)

Create a Pushable Flux soure backed by a bounded BlockingQueue. Queue size is determined by the backPressureAfter parameter

Parameters:

backPressureAfter - Max queue size - once queue is full, producers adding to the Queue will be blocked

Returns:

FluxSource for pushable stream source

futureStream

public <T> PushableLazyFutureStream<T> futureStream(LazyReact s)

Create a pushable LazyFutureStream using the supplied ReactPool

Parameters:

s - ReactPool to use to create the Stream

Returns:

a Tuple2 with a Queue<T> and LazyFutureStream<T> - add data to the Queue to push it to the Stream

futureStream

public static <T> LazyFutureStream<T> futureStream(Adapter<T> adapter,
                                                   LazyReact react)

Create a LazyFutureStream. his will call LazyFutureStream#futureStream(Stream) which creates a sequential LazyFutureStream

Parameters:

adapter - Adapter to create a LazyFutureStream from

Returns:

A LazyFutureStream that will accept values from the supplied adapter

stream

public <T> PushableStream<T> stream()

Create a pushable JDK 8 Stream

Returns:

a Tuple2 with a Queue<T> and Stream<T> - add data to the Queue to push it to the Stream

flux

public <T> PushableFlux<T> flux()

Create a pushable Flux

Returns:

a Tuple2 with a Queue<T> and Fliux<T> - add data to the Queue to push it to the Stream

reactiveSeq

public <T> PushableReactiveSeq<T> reactiveSeq()

Create a pushable ReactiveSeq

Returns:

a Tuple2 with a Queue<T> and Seq<T> - add data to the Queue to push it to the Stream

stream

public static <T> java.util.stream.Stream<T> stream(Adapter<T> adapter)

Create a JDK 8 Stream from the supplied Adapter

Parameters:

adapter - Adapter to create a Steam from

Returns:

Stream that will accept input from supplied adapter

reactiveSeq

public static <T> ReactiveSeq<T> reactiveSeq(Adapter<T> adapter)

Create a pushable ReactiveSeq

Parameters:

adapter - Adapter to create a Seq from

Returns:

A Seq that will accept input from a supplied adapter

flux

public static <T> reactor.core.publisher.Flux<T> flux(Adapter<T> adapter)

Create a pushable ReactiveSeq

Parameters:

adapter - Adapter to create a Flux from

Returns:

A Flux that will accept input from a supplied adapter