cyclops.companion.reactor

Class Fluxs

java.lang.Object

cyclops.companion.reactor.Fluxs

public class Fluxs
extends java.lang.Object

Companion class for working with Reactor Flux types

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static interface	Fluxs.FluxNested
static class	Fluxs.Instances Companion class for creating Type Class instances for working with Fluxs
static interface	Fluxs.NestedFlux

Constructor Summary

Constructors

Constructor and Description
Fluxs()

Method Summary

Modifier and Type	Method and Description
static <T> Active<ReactorWitness.flux,T>	allTypeclasses(reactor.core.publisher.Flux<T> type)
static <T> AnyMSeq<ReactorWitness.flux,T>	anyM(reactor.core.publisher.Flux<T> flux) Construct an AnyM type from a Flux.
static <W1,T> Coproduct<W1,ReactorWitness.flux,T>	coproduct(reactor.core.publisher.Flux<T> list, InstanceDefinitions<W1> def1)
static <W1,T> Coproduct<W1,ReactorWitness.flux,T>	coproduct(InstanceDefinitions<W1> def1, T... values)
static <T> ReactiveSeq<T>	create(java.util.function.Consumer<? super reactor.core.publisher.FluxSink<T>> emitter)
static <T> ReactiveSeq<T>	create(java.util.function.Consumer<? super reactor.core.publisher.FluxSink<T>> emitter, reactor.core.publisher.FluxSink.OverflowStrategy backpressure)
static <T> ReactiveSeq<T>	defer(java.util.function.Supplier<? extends org.reactivestreams.Publisher<T>> supplier)
static <T> ReactiveSeq<T>	empty()
static <T> ReactiveSeq<T>	error(java.lang.Throwable error)
static <O> ReactiveSeq<O>	error(java.lang.Throwable throwable, boolean whenRequested)
static <I> ReactiveSeq<I>	firstEmitting(java.lang.Iterable<? extends org.reactivestreams.Publisher<? extends I>> sources)
static <I> ReactiveSeq<I>	firstEmitting(org.reactivestreams.Publisher<? extends I>... sources)
static <T> reactor.core.publisher.Flux<T>	flux(AnyM<ReactorWitness.flux,T> flux)
static <T> reactor.core.publisher.Flux<T>	fluxFrom(ReactiveSeq<T> stream)
static <W extends WitnessType<W>,T> StreamT<W,T>	fluxify(StreamT<W,T> nested)
static <T,R1,R> reactor.core.publisher.Flux<R>	forEach(reactor.core.publisher.Flux<? extends T> value1, java.util.function.Function<? super T,? extends org.reactivestreams.Publisher<R1>> value2, java.util.function.BiFunction<? super T,? super R1,java.lang.Boolean> filterFunction, java.util.function.BiFunction<? super T,? super R1,? extends R> yieldingFunction) {@code import static cyclops.companion.reactor.Fluxs.forEach; forEach(Flux.range(1, 10), i -> Flux.range(i, 10),(a,b) -> a>2 && b<10,Tuple::tuple) .subscribe(System.out::println); //(3, 3) (3, 4) (3, 5) (3, 6) (3, 7) (3, 8) (3, 9) ...
static <T,R1,R> reactor.core.publisher.Flux<R>	forEach(reactor.core.publisher.Flux<? extends T> value1, java.util.function.Function<? super T,reactor.core.publisher.Flux<R1>> value2, java.util.function.BiFunction<? super T,? super R1,? extends R> yieldingFunction) Perform a For Comprehension over a Flux, accepting an additonal generating function.
static <T1,T2,R1,R2,R> reactor.core.publisher.Flux<R>	forEach3(reactor.core.publisher.Flux<? extends T1> value1, java.util.function.Function<? super T1,? extends org.reactivestreams.Publisher<R1>> value2, java.util.function.BiFunction<? super T1,? super R1,? extends org.reactivestreams.Publisher<R2>> value3, Fn3<? super T1,? super R1,? super R2,? extends R> yieldingFunction) Perform a For Comprehension over a Flux, accepting 2 generating functions.
static <T1,T2,R1,R2,R> reactor.core.publisher.Flux<R>	forEach3(reactor.core.publisher.Flux<? extends T1> value1, java.util.function.Function<? super T1,? extends org.reactivestreams.Publisher<R1>> value2, java.util.function.BiFunction<? super T1,? super R1,? extends org.reactivestreams.Publisher<R2>> value3, Fn3<? super T1,? super R1,? super R2,java.lang.Boolean> filterFunction, Fn3<? super T1,? super R1,? super R2,? extends R> yieldingFunction) Perform a For Comprehension over a Flux, accepting 2 generating functions.
static <T1,T2,T3,R1,R2,R3,R> reactor.core.publisher.Flux<R>	forEach4(reactor.core.publisher.Flux<? extends T1> value1, java.util.function.Function<? super T1,? extends org.reactivestreams.Publisher<R1>> value2, java.util.function.BiFunction<? super T1,? super R1,? extends org.reactivestreams.Publisher<R2>> value3, Fn3<? super T1,? super R1,? super R2,? extends org.reactivestreams.Publisher<R3>> value4, Fn4<? super T1,? super R1,? super R2,? super R3,? extends R> yieldingFunction) Perform a For Comprehension over a Flux, accepting 3 generating functions.
static <T1,T2,T3,R1,R2,R3,R> reactor.core.publisher.Flux<R>	forEach4(reactor.core.publisher.Flux<? extends T1> value1, java.util.function.Function<? super T1,? extends org.reactivestreams.Publisher<R1>> value2, java.util.function.BiFunction<? super T1,? super R1,? extends org.reactivestreams.Publisher<R2>> value3, Fn3<? super T1,? super R1,? super R2,? extends org.reactivestreams.Publisher<R3>> value4, Fn4<? super T1,? super R1,? super R2,? super R3,java.lang.Boolean> filterFunction, Fn4<? super T1,? super R1,? super R2,? super R3,? extends R> yieldingFunction) Perform a For Comprehension over a Flux, accepting 3 generating functions.
static <T> ReactiveSeq<T>	from(org.reactivestreams.Publisher<? extends T> source)
static <T> ReactiveSeq<T>	fromIterable(java.lang.Iterable<? extends T> it)
static <T,W extends WitnessType<W>> AnyM<W,reactor.core.publisher.Flux<T>>	fromStream(AnyM<W,java.util.stream.Stream<T>> anyM)
static <T> ReactiveSeq<T>	fromStream(java.util.stream.Stream<? extends T> s)
static <T,S> ReactiveSeq<T>	generate(java.util.concurrent.Callable<S> stateSupplier, java.util.function.BiFunction<S,reactor.core.publisher.SynchronousSink<T>,S> generator)
static <T,S> ReactiveSeq<T>	generate(java.util.concurrent.Callable<S> stateSupplier, java.util.function.BiFunction<S,reactor.core.publisher.SynchronousSink<T>,S> generator, java.util.function.Consumer<? super S> stateConsumer)
static <T> ReactiveSeq<T>	generate(java.util.function.Consumer<reactor.core.publisher.SynchronousSink<T>> generator)
static ReactiveSeq<java.lang.Long>	interval(java.time.Duration period)
static ReactiveSeq<java.lang.Long>	interval(java.time.Duration delay, java.time.Duration period)
static ReactiveSeq<java.lang.Long>	intervalMillis(long period)
static <T> ReactiveSeq<T>	just(T... data)
static <T> ReactiveSeq<T>	just(T data)
static <W extends WitnessType<W>,T> StreamT<W,T>	liftM(AnyM<W,reactor.core.publisher.Flux<T>> nested)
static <T,W extends WitnessType<W>> StreamT<W,T>	liftM(reactor.core.publisher.Flux<T> opt, W witness)
static <T,W2,R> Nested<ReactorWitness.flux,W2,R>	mapM(reactor.core.publisher.Flux<T> type, java.util.function.Function<? super T,? extends Higher<W2,R>> fn, InstanceDefinitions<W2> defs)
static <T> reactor.core.publisher.Flux<T>	narrow(reactor.core.publisher.Flux<? extends T> observable)
static <W extends WitnessType<W>,T> AnyM<W,reactor.core.publisher.Flux<T>>	nestedFlux(StreamT<W,T> nested)
static <W extends WitnessType<W>,T,R> R	nestedFlux(StreamT<W,T> nested, java.util.function.Function<? super AnyM<W,reactor.core.publisher.Flux<T>>,? extends R> mapper)
static <T> ReactiveSeq<T>	of(T... data)
static <T> ReactiveSeq<T>	of(T value)
static <T> ReactiveSeq<T>	ofNullable(T nullable)
static ReactiveSeq<java.lang.Integer>	range(int start, int end)
static <T> reactor.core.publisher.Flux<T>	raw(AnyM<ReactorWitness.flux,T> anyM)
static <T> ReactiveSeq<T>	reactiveSeq(reactor.core.publisher.Flux<T> flux)
static <T> ReactiveSeq<T>	reactiveSeq(org.reactivestreams.Publisher<T> flux)
static <T,R> reactor.core.publisher.Flux<R>	tailRec(T initial, java.util.function.Function<? super T,? extends reactor.core.publisher.Flux<? extends Xor<T,R>>> fn)
static <W1 extends WitnessType<W1>,T> XorM<W1,ReactorWitness.flux,T>	xorM(reactor.core.publisher.Flux<T> type)

Methods inherited from class java.lang.Object

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

Constructor Detail

Fluxs

public Fluxs()

Method Detail

coproduct

public static <W1,T> Coproduct<W1,ReactorWitness.flux,T> coproduct(reactor.core.publisher.Flux<T> list,
                                                                   InstanceDefinitions<W1> def1)

coproduct

public static <W1,T> Coproduct<W1,ReactorWitness.flux,T> coproduct(InstanceDefinitions<W1> def1,
                                                                   T... values)

xorM

public static <W1 extends WitnessType<W1>,T> XorM<W1,ReactorWitness.flux,T> xorM(reactor.core.publisher.Flux<T> type)

raw

public static <T> reactor.core.publisher.Flux<T> raw(AnyM<ReactorWitness.flux,T> anyM)

fromStream

public static <T,W extends WitnessType<W>> AnyM<W,reactor.core.publisher.Flux<T>> fromStream(AnyM<W,java.util.stream.Stream<T>> anyM)

tailRec

public static <T,R> reactor.core.publisher.Flux<R> tailRec(T initial,
                                                           java.util.function.Function<? super T,? extends reactor.core.publisher.Flux<? extends Xor<T,R>>> fn)

narrow

public static <T> reactor.core.publisher.Flux<T> narrow(reactor.core.publisher.Flux<? extends T> observable)

fluxFrom

public static <T> reactor.core.publisher.Flux<T> fluxFrom(ReactiveSeq<T> stream)

fluxify

public static <W extends WitnessType<W>,T> StreamT<W,T> fluxify(StreamT<W,T> nested)

nestedFlux

public static <W extends WitnessType<W>,T,R> R nestedFlux(StreamT<W,T> nested,
                                                          java.util.function.Function<? super AnyM<W,reactor.core.publisher.Flux<T>>,? extends R> mapper)

nestedFlux

public static <W extends WitnessType<W>,T> AnyM<W,reactor.core.publisher.Flux<T>> nestedFlux(StreamT<W,T> nested)

liftM

public static <T,W extends WitnessType<W>> StreamT<W,T> liftM(reactor.core.publisher.Flux<T> opt,
                                                              W witness)

liftM

public static <W extends WitnessType<W>,T> StreamT<W,T> liftM(AnyM<W,reactor.core.publisher.Flux<T>> nested)

reactiveSeq

public static <T> ReactiveSeq<T> reactiveSeq(reactor.core.publisher.Flux<T> flux)

reactiveSeq

public static <T> ReactiveSeq<T> reactiveSeq(org.reactivestreams.Publisher<T> flux)

range

public static ReactiveSeq<java.lang.Integer> range(int start,
                                                   int end)

of

public static <T> ReactiveSeq<T> of(T... data)

of

public static <T> ReactiveSeq<T> of(T value)

ofNullable

public static <T> ReactiveSeq<T> ofNullable(T nullable)

create

public static <T> ReactiveSeq<T> create(java.util.function.Consumer<? super reactor.core.publisher.FluxSink<T>> emitter)

create

public static <T> ReactiveSeq<T> create(java.util.function.Consumer<? super reactor.core.publisher.FluxSink<T>> emitter,
                                        reactor.core.publisher.FluxSink.OverflowStrategy backpressure)

defer

public static <T> ReactiveSeq<T> defer(java.util.function.Supplier<? extends org.reactivestreams.Publisher<T>> supplier)

empty

public static <T> ReactiveSeq<T> empty()

error

public static <T> ReactiveSeq<T> error(java.lang.Throwable error)

error

public static <O> ReactiveSeq<O> error(java.lang.Throwable throwable,
                                       boolean whenRequested)

firstEmitting

@SafeVarargs
public static <I> ReactiveSeq<I> firstEmitting(org.reactivestreams.Publisher<? extends I>... sources)

firstEmitting

public static <I> ReactiveSeq<I> firstEmitting(java.lang.Iterable<? extends org.reactivestreams.Publisher<? extends I>> sources)

from

public static <T> ReactiveSeq<T> from(org.reactivestreams.Publisher<? extends T> source)

fromIterable

public static <T> ReactiveSeq<T> fromIterable(java.lang.Iterable<? extends T> it)

fromStream

public static <T> ReactiveSeq<T> fromStream(java.util.stream.Stream<? extends T> s)

generate

public static <T> ReactiveSeq<T> generate(java.util.function.Consumer<reactor.core.publisher.SynchronousSink<T>> generator)

generate

public static <T,S> ReactiveSeq<T> generate(java.util.concurrent.Callable<S> stateSupplier,
                                            java.util.function.BiFunction<S,reactor.core.publisher.SynchronousSink<T>,S> generator)

generate

public static <T,S> ReactiveSeq<T> generate(java.util.concurrent.Callable<S> stateSupplier,
                                            java.util.function.BiFunction<S,reactor.core.publisher.SynchronousSink<T>,S> generator,
                                            java.util.function.Consumer<? super S> stateConsumer)

interval

public static ReactiveSeq<java.lang.Long> interval(java.time.Duration period)

interval

public static ReactiveSeq<java.lang.Long> interval(java.time.Duration delay,
                                                   java.time.Duration period)

intervalMillis

public static ReactiveSeq<java.lang.Long> intervalMillis(long period)

just

@SafeVarargs
public static <T> ReactiveSeq<T> just(T... data)

just

public static <T> ReactiveSeq<T> just(T data)

anyM

public static <T> AnyMSeq<ReactorWitness.flux,T> anyM(reactor.core.publisher.Flux<T> flux)

Construct an AnyM type from a Flux. This allows the Flux to be manipulated according to a standard interface along with a vast array of other Java Monad implementations

  
    
    AnyMSeq<Integer> flux = Fluxs.anyM(Flux.just(1,2,3));
    AnyMSeq<Integer> transformedFlux = myGenericOperation(flux);
    
    public AnyMSeq<Integer> myGenericOperation(AnyMSeq<Integer> monad);
 
 

Parameters:

flux - To wrap inside an AnyM

Returns:

AnyMSeq wrapping a flux

flux

public static <T> reactor.core.publisher.Flux<T> flux(AnyM<ReactorWitness.flux,T> flux)

forEach4

public static <T1,T2,T3,R1,R2,R3,R> reactor.core.publisher.Flux<R> forEach4(reactor.core.publisher.Flux<? extends T1> value1,
                                                                            java.util.function.Function<? super T1,? extends org.reactivestreams.Publisher<R1>> value2,
                                                                            java.util.function.BiFunction<? super T1,? super R1,? extends org.reactivestreams.Publisher<R2>> value3,
                                                                            Fn3<? super T1,? super R1,? super R2,? extends org.reactivestreams.Publisher<R3>> value4,
                                                                            Fn4<? super T1,? super R1,? super R2,? super R3,? extends R> yieldingFunction)

Perform a For Comprehension over a Flux, accepting 3 generating functions. This results in a four level nested internal iteration over the provided Publishers.

 
    
   import static cyclops.companion.reactor.Fluxs.forEach4;
    
          forEach4(Flux.range(1,10), 
                  a-> ReactiveSeq.iterate(a,i->i+1).limit(10),
                  (a,b) -> Maybe.<Integer>of(a+b),
                  (a,b,c) -> Mono.<Integer>just(a+b+c),
                  Tuple::tuple)
 
 
 

Parameters:

value1 - top level Flux

value2 - Nested publisher

value3 - Nested publisher

value4 - Nested publisher

yieldingFunction - Generates a result per combination

Returns:

Flux with an element per combination of nested publishers generated by the yielding function

forEach4

public static <T1,T2,T3,R1,R2,R3,R> reactor.core.publisher.Flux<R> forEach4(reactor.core.publisher.Flux<? extends T1> value1,
                                                                            java.util.function.Function<? super T1,? extends org.reactivestreams.Publisher<R1>> value2,
                                                                            java.util.function.BiFunction<? super T1,? super R1,? extends org.reactivestreams.Publisher<R2>> value3,
                                                                            Fn3<? super T1,? super R1,? super R2,? extends org.reactivestreams.Publisher<R3>> value4,
                                                                            Fn4<? super T1,? super R1,? super R2,? super R3,java.lang.Boolean> filterFunction,
                                                                            Fn4<? super T1,? super R1,? super R2,? super R3,? extends R> yieldingFunction)

Perform a For Comprehension over a Flux, accepting 3 generating functions. This results in a four level nested internal iteration over the provided Publishers.

 
 
  import static cyclops.companion.reactor.Fluxs.forEach4;
   
  forEach4(Flux.range(1,10), 
                            a-> ReactiveSeq.iterate(a,i->i+1).limit(10),
                            (a,b) -> Maybe.<Integer>just(a+b),
                            (a,b,c) -> Mono.<Integer>just(a+b+c),
                            (a,b,c,d) -> a+b+c+d <100,
                            Tuple::tuple);
 
 
 

Parameters:

value1 - top level Flux

value2 - Nested publisher

value3 - Nested publisher

value4 - Nested publisher

filterFunction - A filtering function, keeps values where the predicate holds

yieldingFunction - Generates a result per combination

Returns:

Flux with an element per combination of nested publishers generated by the yielding function

forEach3

public static <T1,T2,R1,R2,R> reactor.core.publisher.Flux<R> forEach3(reactor.core.publisher.Flux<? extends T1> value1,
                                                                      java.util.function.Function<? super T1,? extends org.reactivestreams.Publisher<R1>> value2,
                                                                      java.util.function.BiFunction<? super T1,? super R1,? extends org.reactivestreams.Publisher<R2>> value3,
                                                                      Fn3<? super T1,? super R1,? super R2,? extends R> yieldingFunction)

Perform a For Comprehension over a Flux, accepting 2 generating functions. This results in a three level nested internal iteration over the provided Publishers.

  
 
 import static cyclops.companion.reactor.Fluxs.forEach;
 
 forEach(Flux.range(1,10), 
                            a-> ReactiveSeq.iterate(a,i->i+1).limit(10),
                            (a,b) -> Maybe.<Integer>of(a+b),
                            Tuple::tuple);
 
 
 

Parameters:

value1 - top level Flux

value2 - Nested publisher

value3 - Nested publisher

yieldingFunction - Generates a result per combination

Returns:

Flux with an element per combination of nested publishers generated by the yielding function

forEach3

public static <T1,T2,R1,R2,R> reactor.core.publisher.Flux<R> forEach3(reactor.core.publisher.Flux<? extends T1> value1,
                                                                      java.util.function.Function<? super T1,? extends org.reactivestreams.Publisher<R1>> value2,
                                                                      java.util.function.BiFunction<? super T1,? super R1,? extends org.reactivestreams.Publisher<R2>> value3,
                                                                      Fn3<? super T1,? super R1,? super R2,java.lang.Boolean> filterFunction,
                                                                      Fn3<? super T1,? super R1,? super R2,? extends R> yieldingFunction)

Perform a For Comprehension over a Flux, accepting 2 generating functions. This results in a three level nested internal iteration over the provided Publishers.

 

 import static cyclops.companion.reactor.Fluxs.forEach;

 forEach(Flux.range(1,10),
                       a-> ReactiveSeq.iterate(a,i->i+1).limit(10),
                       (a,b) -> Maybe.<Integer>of(a+b),
                       (a,b,c) ->a+b+c<10,
                       Tuple::tuple).toListX();
 
 

Parameters:

value1 - top level Flux

value2 - Nested publisher

value3 - Nested publisher

filterFunction - A filtering function, keeps values where the predicate holds

yieldingFunction - Generates a result per combination

Returns:

forEach

public static <T,R1,R> reactor.core.publisher.Flux<R> forEach(reactor.core.publisher.Flux<? extends T> value1,
                                                              java.util.function.Function<? super T,reactor.core.publisher.Flux<R1>> value2,
                                                              java.util.function.BiFunction<? super T,? super R1,? extends R> yieldingFunction)

Perform a For Comprehension over a Flux, accepting an additonal generating function. This results in a two level nested internal iteration over the provided Publishers.

  
 
  import static cyclops.companion.reactor.Fluxs.forEach;
  forEach(Flux.range(1, 10), i -> Flux.range(i, 10), Tuple::tuple)
              .subscribe(System.out::println);
              
       //(1, 1)
         (1, 2)
         (1, 3)
         (1, 4)
         ...
 
 

Parameters:

value1 - top level Flux

value2 - Nested publisher

yieldingFunction - Generates a result per combination

Returns:

forEach

public static <T,R1,R> reactor.core.publisher.Flux<R> forEach(reactor.core.publisher.Flux<? extends T> value1,
                                                              java.util.function.Function<? super T,? extends org.reactivestreams.Publisher<R1>> value2,
                                                              java.util.function.BiFunction<? super T,? super R1,java.lang.Boolean> filterFunction,
                                                              java.util.function.BiFunction<? super T,? super R1,? extends R> yieldingFunction)

  
 
   import static cyclops.companion.reactor.Fluxs.forEach;
 
   forEach(Flux.range(1, 10), i -> Flux.range(i, 10),(a,b) -> a>2 && b<10,Tuple::tuple)
               .subscribe(System.out::println);
               
       //(3, 3)
         (3, 4)
         (3, 5)
         (3, 6)
         (3, 7)
         (3, 8)
         (3, 9)
         ...
     
 
 

Parameters:

value1 - top level Flux

value2 - Nested publisher

filterFunction - A filtering function, keeps values where the predicate holds

yieldingFunction - Generates a result per combination

Returns:

allTypeclasses

public static <T> Active<ReactorWitness.flux,T> allTypeclasses(reactor.core.publisher.Flux<T> type)

mapM

public static <T,W2,R> Nested<ReactorWitness.flux,W2,R> mapM(reactor.core.publisher.Flux<T> type,
                                                             java.util.function.Function<? super T,? extends Higher<W2,R>> fn,
                                                             InstanceDefinitions<W2> defs)