com.aol.cyclops.reactor

Class Fluxes

java.lang.Object

com.aol.cyclops.reactor.Fluxes

public class Fluxes
extends java.lang.Object

Companion class for working with Reactor Flux types

Constructor Summary

Constructors

Constructor and Description
Fluxes()

Method Summary

Modifier and Type	Method and Description
static <T> AnyMSeq<T>	anyM(reactor.core.publisher.Flux<T> flux) Construct an AnyM type from a Flux.
static <T> reactor.core.publisher.Flux<T>	combine(reactor.core.publisher.Flux<T> stream, java.util.function.BiPredicate<? super T,? super T> predicate, java.util.function.BinaryOperator<T> op) Combine adjacent elements within a Flux for as long as the supplied predicate holds.
static <T> reactor.core.publisher.Flux<T>	cycle(reactor.core.publisher.Flux<T> stream, Monoid<T> m, int times) Convert to a Flux with the result of a reduction operation repeated specified times
static <T> reactor.core.publisher.Flux<T>	cycleUntil(reactor.core.publisher.Flux<T> stream, java.util.function.Predicate<? super T> predicate) Repeat in a Stream until specified predicate holds
static <T> reactor.core.publisher.Flux<T>	cycleWhile(reactor.core.publisher.Flux<T> stream, java.util.function.Predicate<? super T> predicate) Repeat in a Flux while specified predicate holds
static <T> reactor.core.publisher.Flux<T>	deleteBetween(reactor.core.publisher.Flux<T> stream, int start, int end) Delete elements between given indexes in a Flux
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 com.aol.cyclops.reactor.Fluxes.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, TriFunction<? 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, TriFunction<? super T1,? super R1,? super R2,java.lang.Boolean> filterFunction, TriFunction<? 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, TriFunction<? super T1,? super R1,? super R2,? extends org.reactivestreams.Publisher<R3>> value4, QuadFunction<? 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, TriFunction<? super T1,? super R1,? super R2,? extends org.reactivestreams.Publisher<R3>> value4, QuadFunction<? super T1,? super R1,? super R2,? super R3,java.lang.Boolean> filterFunction, QuadFunction<? super T1,? super R1,? super R2,? super R3,? extends R> yieldingFunction) Perform a For Comprehension over a Flux, accepting 3 generating functions.
static <T,K,A,D> reactor.core.publisher.Flux<org.jooq.lambda.tuple.Tuple2<K,D>>	grouped(reactor.core.publisher.Flux<T> flux, java.util.function.Function<? super T,? extends K> classifier) Classify a Flux by the supplied classifying function.
static <T,K,A,D> reactor.core.publisher.Flux<org.jooq.lambda.tuple.Tuple2<K,D>>	grouped(reactor.core.publisher.Flux<T> flux, java.util.function.Function<? super T,? extends K> classifier, java.util.stream.Collector<? super T,A,D> downstream) Classify a Flux by the supplied classifying function, materialize into collection with the supplied Collector
static <T> reactor.core.publisher.Flux<ListX<T>>	grouped(reactor.core.publisher.Flux<T> flux, int size) Group elements in a Flux
static <T,C extends java.util.Collection<? super T>> reactor.core.publisher.Flux<C>	grouped(reactor.core.publisher.Flux<T> stream, int groupSize, java.util.function.Supplier<C> factory) Batch elements in a Flux by size into a collection created by the supplied factory
static <T> reactor.core.publisher.Flux<ListX<T>>	groupedStatefullyUntil(reactor.core.publisher.Flux<T> flux, java.util.function.BiPredicate<ListX<? super T>,? super T> predicate) Create Flux of ListX where each ListX is populated while the supplied bipredicate holds.
static <T> reactor.core.publisher.Flux<ListX<T>>	groupedUntil(reactor.core.publisher.Flux<T> stream, java.util.function.Predicate<? super T> predicate) Create a Flux batched by List, where each batch is populated while the predicate doesn't hold
static <T> reactor.core.publisher.Flux<ListX<T>>	groupedWhile(reactor.core.publisher.Flux<T> stream, java.util.function.Predicate<? super T> predicate) Create a Flux batched by List, where each batch is populated while the predicate holds
static <T,C extends java.util.Collection<? super T>> reactor.core.publisher.Flux<C>	groupedWhile(reactor.core.publisher.Flux<T> stream, java.util.function.Predicate<? super T> predicate, java.util.function.Supplier<C> factory) Create a Flux batched by List, where each batch is populated while the predicate holds
static <T> reactor.core.publisher.Flux<T>	insertAt(reactor.core.publisher.Flux<T> stream, int pos, T... values) Insert data into a Flux at given position
static <T> reactor.core.publisher.Flux<T>	intersperse(reactor.core.publisher.Flux<T> flux, T value) Intersperse the supplied value throughout this Flux
static <T> java.util.Iterator<T>	iterator(reactor.core.publisher.Flux<T> stream) Get a lazy iterator over the data in a Flux
static <T,U> reactor.core.publisher.Flux<U>	ofType(reactor.core.publisher.Flux<T> stream, java.lang.Class<? extends U> type) Keep only those elements in a stream that are of a given type.
static <T> reactor.core.publisher.Flux<T>	onEmpty(reactor.core.publisher.Flux<T> flux, T value) Generate a lazy Flux containing the supplied value if this Flux is empty
static <T> reactor.core.publisher.Flux<T>	onEmptyGet(reactor.core.publisher.Flux<T> flux, java.util.function.Supplier<? extends T> value) Generate a lazy Flux containing a value generated from the provided Supplier if this Flux is empty
static <T> reactor.core.publisher.Flux<T>	onEmptySwitch(reactor.core.publisher.Flux<T> flux, java.util.function.Supplier<? extends reactor.core.publisher.Flux<T>> value) Lazily switch a Flux if it is empty if one created by the provided Supplier
static <T,X extends java.lang.Throwable> reactor.core.publisher.Flux<T>	onEmptyThrow(reactor.core.publisher.Flux<T> flux, java.util.function.Supplier<? extends X> value) Throw the supplied exception, on first use, if the provided Flux is empty
static <T,R> reactor.core.publisher.Flux<R>	patternMatch(reactor.core.publisher.Flux<T> flux, java.util.function.Function<Matchable.CheckValue1<T,R>,Matchable.CheckValue1<T,R>> case1, java.util.function.Supplier<? extends R> otherwise) Transform the elements of this Stream with a Pattern Matching case and default value
static <T> reactor.core.publisher.Flux<T>	removeAll(reactor.core.publisher.Flux<T> flux, java.lang.Iterable<? extends T> iterable) Remove all the values in the supplied Iterable from the Flux
static <T> reactor.core.publisher.Flux<T>	retainAll(reactor.core.publisher.Flux<T> flux, java.lang.Iterable<? extends T> iterable) Retain only the values in the supplied Iterable in the generated Flux
static <T> reactor.core.publisher.Flux<T>	reverse(reactor.core.publisher.Flux<T> flux) Lazily reverse a Flux by materializing it on first use and iterating over the materialized collection in reverse order
static <T> reactor.core.publisher.Flux<T>	scanRight(reactor.core.publisher.Flux<T> flux, Monoid<T> monoid)
static <T,U> reactor.core.publisher.Flux<U>	scanRight(reactor.core.publisher.Flux<T> flux, U identity, java.util.function.BiFunction<? super T,? super U,? extends U> combiner) Apply the identity function / combiner from left to right accumulating partial results in the resulting Flux This is a lazy operation that materializes the provided Flux in order to reverse it on first use.
static <T> reactor.core.publisher.Flux<T>	shuffle(reactor.core.publisher.Flux<T> flux) Lazily shuffle a Flux by materializing it on first use & shuffling it
static <T> reactor.core.publisher.Flux<T>	shuffle(reactor.core.publisher.Flux<T> flux, java.util.Random random) Lazily shuffle a Flux by materializing it on first use & shuffling it using the supplied Random number generator
static <T> reactor.core.publisher.Flux<ListX<T>>	sliding(reactor.core.publisher.Flux<T> flux, int windowSize, int increment) Create a sliding view over this Sequence
static <T> reactor.core.publisher.Flux<T>	sorted(reactor.core.publisher.Flux<T> flux) Lazily shuffle a Flux by materializing it on first use & sorting it
static <T> reactor.core.publisher.Flux<T>	sorted(reactor.core.publisher.Flux<T> flux, java.util.Comparator<? super T> c) Sort Flux using the provided comparator This is a lazy operation that materializes the provided Flux on first use in order to sort it.
static <T,U> reactor.core.publisher.Flux<T>	sorted(reactor.core.publisher.Flux<T> flux, java.util.function.Function<? super T,? extends U> function) Sort the Flux using the supplied Function.
static <T> reactor.core.publisher.Flux<T>	takeUntil(reactor.core.publisher.Flux<T> stream, java.util.function.Predicate<? super T> predicate) Take elements from the Stream until the predicate returns true, after which all elements are excluded.
static <T,R> reactor.core.publisher.Flux<R>	trampoline(reactor.core.publisher.Flux<T> flux, java.util.function.Function<? super T,? extends Trampoline<? extends R>> mapper) Performs a map operation that can call a recursive method without running out of stack space
static <T> reactor.core.publisher.Flux<org.jooq.lambda.tuple.Tuple2<T,java.lang.Long>>	zipWithIndex(reactor.core.publisher.Flux<T> stream) Add an index to the supplied Flux

Methods inherited from class java.lang.Object

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

Constructor Detail

Fluxes

public Fluxes()

Method Detail

anyM

public static <T> AnyMSeq<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 = Fluxes.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

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,
                                                                            TriFunction<? super T1,? super R1,? super R2,? extends org.reactivestreams.Publisher<R3>> value4,
                                                                            QuadFunction<? 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 com.aol.cyclops.reactor.Fluxes.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,
                                                                            TriFunction<? super T1,? super R1,? super R2,? extends org.reactivestreams.Publisher<R3>> value4,
                                                                            QuadFunction<? super T1,? super R1,? super R2,? super R3,java.lang.Boolean> filterFunction,
                                                                            QuadFunction<? 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 com.aol.cyclops.reactor.Fluxes.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,
                                                                      TriFunction<? 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 com.aol.cyclops.reactor.Fluxes.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,
                                                                      TriFunction<? super T1,? super R1,? super R2,java.lang.Boolean> filterFunction,
                                                                      TriFunction<? 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 com.aol.cyclops.reactor.Fluxes.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 com.aol.cyclops.reactor.Fluxes.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 com.aol.cyclops.reactor.Fluxes.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:

grouped

public static final <T,C extends java.util.Collection<? super T>> reactor.core.publisher.Flux<C> grouped(reactor.core.publisher.Flux<T> stream,
                                                                                                         int groupSize,
                                                                                                         java.util.function.Supplier<C> factory)

Batch elements in a Flux by size into a collection created by the supplied factory

 
 Fluxes.grouped(Flux.just(1,2,3,1,1,1),3,()->new TreeSet<>())
 
 //Flux[Set[1,2,4],Set[1]]
    
 
 

Parameters:

stream - Flux to group

groupSize - Size of each batch

factory - A supplier used to create the colleciton used to store groups

Returns:

Grouped Flux

iterator

public static <T> java.util.Iterator<T> iterator(reactor.core.publisher.Flux<T> stream)

Get a lazy iterator over the data in a Flux

Parameters:

stream - Flux to get iterator for

Returns:

Iterator over data in the supplied Flux

combine

public static <T> reactor.core.publisher.Flux<T> combine(reactor.core.publisher.Flux<T> stream,
                                                         java.util.function.BiPredicate<? super T,? super T> predicate,
                                                         java.util.function.BinaryOperator<T> op)

Combine adjacent elements within a Flux for as long as the supplied predicate holds. This is a stateful grouping & reduction operation. The output of a combination may in turn be combined with it's neighbour

  
  Fluxes.combine(Flux.just(1,1,2,3),(a, b)->a.equals(b),Semigroups.intSum)
                   
                   
  //Flux(3,4) 
 

Parameters:

stream - Flux to combine elements

predicate - Determines if subsequent element should be combine

op - Combiner function

Returns:

Flux with partial reduction of internal elements by combiner where predicate holds

cycle

public static final <T> reactor.core.publisher.Flux<T> cycle(reactor.core.publisher.Flux<T> stream,
                                                             Monoid<T> m,
                                                             int times)

Convert to a Flux with the result of a reduction operation repeated specified times

  
   List<Integer> list = Fluxes.cycle(Flux.just(1,2,2)),Reducers.toCountInt(),3)
                                 .collect(Collectors.toList());
   //List[3,3,3];
   
 

Parameters:

stream - Flux to cycle

m - Monoid to be used in reduction

times - number of times to cycle

Returns:

Stream with reduced values repeated

cycleWhile

public static final <T> reactor.core.publisher.Flux<T> cycleWhile(reactor.core.publisher.Flux<T> stream,
                                                                  java.util.function.Predicate<? super T> predicate)

Repeat in a Flux while specified predicate holds

  
  int count =0;
  
        assertThat(Fluxes.cycleWhile(Flux.just(1,2,2)
                                            ,next -> count++<6 )
                                            .collect(Collectors.toList()),equalTo(Arrays.asList(1,2,2,1,2,2)));
 
 

Parameters:

predicate - repeat while true

Returns:

Repeating Stream

takeUntil

public static final <T> reactor.core.publisher.Flux<T> takeUntil(reactor.core.publisher.Flux<T> stream,
                                                                 java.util.function.Predicate<? super T> predicate)

Take elements from the Stream until the predicate returns true, after which all elements are excluded.

  
  Fluxes.limitUntil(Flux.justf(4,3,6,7),i->i==6).collect(Collectors.toList())
  //Arrays.asList(4,3)
 

Parameters:

stream - Flux to take elements from

predicate - Take until predicate is true

Returns:

Stream with limited elements

cycleUntil

public static final <T> reactor.core.publisher.Flux<T> cycleUntil(reactor.core.publisher.Flux<T> stream,
                                                                  java.util.function.Predicate<? super T> predicate)

Repeat in a Stream until specified predicate holds

  
  count =0;
        assertThat(Fluxes.cycleUntil(Flux.just(1,2,2,3)
                                            ,next -> count++>10 )
                                            .collect(Collectors.toList()),equalTo(Arrays.asList(1, 2, 2, 3, 1, 2, 2, 3, 1, 2, 2)));
    
 
 

Parameters:

predicate - repeat while true

Returns:

Repeating Stream

ofType

public static <T,U> reactor.core.publisher.Flux<U> ofType(reactor.core.publisher.Flux<T> stream,
                                                          java.lang.Class<? extends U> type)

Keep only those elements in a stream that are of a given type. assertThat(Arrays.asList(1, 2, 3), equalTo( Fluxes.ofType(Flux.just(1, "a", 2, "b", 3,Integer.class));

groupedWhile

public static final <T> reactor.core.publisher.Flux<ListX<T>> groupedWhile(reactor.core.publisher.Flux<T> stream,
                                                                           java.util.function.Predicate<? super T> predicate)

Create a Flux batched by List, where each batch is populated while the predicate holds

  
 Fluxes.groupedWhile(Flux.just(1,2,3,4,5,6),i->i%3!=0))
           
  //Flux[[1,2,3],[4,5,6]] 
 
 

Parameters:

stream - Batch while predicate holds, then open next batch

predicate - batched into lists determined by the predicate supplied

Returns:

Grouped flux

groupedWhile

public static final <T,C extends java.util.Collection<? super T>> reactor.core.publisher.Flux<C> groupedWhile(reactor.core.publisher.Flux<T> stream,
                                                                                                              java.util.function.Predicate<? super T> predicate,
                                                                                                              java.util.function.Supplier<C> factory)

Create a Flux batched by List, where each batch is populated while the predicate holds

  
 Fluxes.groupedWhile(Flux.just(1,2,3,4,5,6),i->i%3!=0),()->SetX.empty())
           
  //Flux[[1,2,3],[4,5,6]] 
 
 

Parameters:

stream - Batch while predicate holds, then open next batch

predicate - batched into lists determined by the predicate supplied

factory - Used to create the collection to store the batch

Returns:

Grouped flux

groupedUntil

public static final <T> reactor.core.publisher.Flux<ListX<T>> groupedUntil(reactor.core.publisher.Flux<T> stream,
                                                                           java.util.function.Predicate<? super T> predicate)

Create a Flux batched by List, where each batch is populated while the predicate doesn't hold

  
 
 Fluxes.groupedUntil(Flux.just(1,2,3,4,5,6),i->i%3!=0),()->SetX.empty())
    
 
 [1]
       [2]
       [3, 4]
       [5]
       [6]
    
 
 

Parameters:

stream - Batch while predicate doesn't hold, then open next batch

predicate - batched into lists determined by the predicate supplied

Returns:

Grouped flux

trampoline

public static <T,R> reactor.core.publisher.Flux<R> trampoline(reactor.core.publisher.Flux<T> flux,
                                                              java.util.function.Function<? super T,? extends Trampoline<? extends R>> mapper)

Performs a map operation that can call a recursive method without running out of stack space

 
 
       Fluxes.trampoline(Flux.just(10,20,30,40),i-> fibonacci(i))
                .forEach(System.out::println); 
                
       Trampoline<Long> fibonacci(int i){
           return fibonacci(i,1,0);
       }
       Trampoline<Long> fibonacci(int n, long a, long b) {
           return n == 0 ? Trampoline.done(b) : Trampoline.more( ()->fibonacci(n-1, a+b, a));
       }        
                
 55
       6765
       832040
       102334155
 
 
 
       Fluxes.trampoline(Flux.just(10_000,200_000,3_000_000,40_000_000),i-> fibonacci(i))
                .forEach(System.out::println);
                
                
 completes successfully
 
 

Parameters:

mapper - TCO Transformation function

Returns:

Functor transformed by the supplied transformation function

patternMatch

public static <T,R> reactor.core.publisher.Flux<R> patternMatch(reactor.core.publisher.Flux<T> flux,
                                                                java.util.function.Function<Matchable.CheckValue1<T,R>,Matchable.CheckValue1<T,R>> case1,
                                                                java.util.function.Supplier<? extends R> otherwise)

Transform the elements of this Stream with a Pattern Matching case and default value

 
 List<String> result = Fluxes.patternMatch(Flux.just(1,2,3,4),
                                                      c->c.valuesWhere(i->"even", (Integer i)->i%2==0 )
                                                    )
 
 // CollectionX["odd","even","odd","even"]
 

Parameters:

case1 - Function to generate a case (or chain of cases as a single case)

otherwise - Value if supplied case doesn't match

Returns:

CollectionX where elements are transformed by pattern matching

reverse

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

Lazily reverse a Flux by materializing it on first use and iterating over the materialized collection in reverse order

Parameters:

flux - Flux to reverse

Returns:

Reversed Flux

shuffle

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

Lazily shuffle a Flux by materializing it on first use & shuffling it

Parameters:

flux - Flux to shuffle

Returns:

Shuffled flux

shuffle

public static <T> reactor.core.publisher.Flux<T> shuffle(reactor.core.publisher.Flux<T> flux,
                                                         java.util.Random random)

Lazily shuffle a Flux by materializing it on first use & shuffling it using the supplied Random number generator

Parameters:

flux - Flux to shuffle

random - the source of randomness to use to shuffle the list.

Returns:

Shuffled flux

sorted

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

Lazily shuffle a Flux by materializing it on first use & sorting it

Parameters:

flux - Flux to sort

Returns:

Sorted Flux

onEmpty

public static <T> reactor.core.publisher.Flux<T> onEmpty(reactor.core.publisher.Flux<T> flux,
                                                         T value)

Generate a lazy Flux containing the supplied value if this Flux is empty

  
    Flux<Integer> empty = Flux.empty();
    Flux<Integer> nonEmpty = Fluxes.onEmpty(empty,10);
    //Flux[10]
 
 

Parameters:

flux - Flux to replace with supplied value if empty

value - Value to use if Flux is empty

Returns:

Non-empty Flux

onEmptySwitch

public static <T> reactor.core.publisher.Flux<T> onEmptySwitch(reactor.core.publisher.Flux<T> flux,
                                                               java.util.function.Supplier<? extends reactor.core.publisher.Flux<T>> value)

Lazily switch a Flux if it is empty if one created by the provided Supplier

  
    Flux<Integer> empty = Flux.empty();
    Flux<Integer> nonEmpty = Fluxes.onEmpty(empty,()->Flux.just(10));
    //Flux[10]
 
 

Parameters:

flux - Flux to replace with supplied Flux if empty

value - Supplier that generates the replacement Flux

Returns:

Non-empty Flux

onEmptyGet

public static <T> reactor.core.publisher.Flux<T> onEmptyGet(reactor.core.publisher.Flux<T> flux,
                                                            java.util.function.Supplier<? extends T> value)

Generate a lazy Flux containing a value generated from the provided Supplier if this Flux is empty

  
    Flux<Integer> empty = Flux.empty();
    Flux<Integer> nonEmpty = Fluxes.onEmpty(empty,()->10);
    //Flux[10]
 
 

Parameters:

flux - Flux to replace with supplied Flux if empty

value - Supplier that generates the replacement value

Returns:

Non-empty Flux

onEmptyThrow

public static <T,X extends java.lang.Throwable> reactor.core.publisher.Flux<T> onEmptyThrow(reactor.core.publisher.Flux<T> flux,
                                                                                            java.util.function.Supplier<? extends X> value)

Throw the supplied exception, on first use, if the provided Flux is empty

  
    Flux<Integer> empty = Flux.empty();
    Flux<Integer> willThrowOnUse = Fluxes.onEmpty(empty,()->new RuntimeException("boo!"));
   
 
 

Parameters:

flux - Flux to check if empty

value - Supplier that generates Exception to be thrown if Flux is empty

Returns:

Non-empty Flux

sorted

public static <T,U> reactor.core.publisher.Flux<T> sorted(reactor.core.publisher.Flux<T> flux,
                                                          java.util.function.Function<? super T,? extends U> function)

Sort the Flux using the supplied Function. This is a lazy operation that materializes the provided Flux on first use in order to sort it.

Parameters:

flux - Flux to sort

function - To sort Flux with

Returns:

Sorted Flux

sorted

public static <T> reactor.core.publisher.Flux<T> sorted(reactor.core.publisher.Flux<T> flux,
                                                        java.util.Comparator<? super T> c)

Sort Flux using the provided comparator This is a lazy operation that materializes the provided Flux on first use in order to sort it.

Parameters:

flux - Flux to sort

c - Comparator to sort Flux

Returns:

Sorted Flux

scanRight

public static <T,U> reactor.core.publisher.Flux<U> scanRight(reactor.core.publisher.Flux<T> flux,
                                                             U identity,
                                                             java.util.function.BiFunction<? super T,? super U,? extends U> combiner)

Apply the identity function / combiner from left to right accumulating partial results in the resulting Flux This is a lazy operation that materializes the provided Flux in order to reverse it on first use.

  
 assertThat(Fluxes.scanRight(Flux.just("a", "ab", "abc").map(str->str.length()),0, (t, u) -> u + t).toList().size(),
             is(asList(0, 3, 5, 6).size()));
 
 
 

Parameters:

flux - Flux to scanRight

identity - Identity value that leaves the current value unchanged when applied in the combiner function

combiner - Function for combining two values

Returns:

Flux with scanRight applied

scanRight

public static <T> reactor.core.publisher.Flux<T> scanRight(reactor.core.publisher.Flux<T> flux,
                                                           Monoid<T> monoid)

Parameters:

flux -

monoid -

Returns:

removeAll

public static <T> reactor.core.publisher.Flux<T> removeAll(reactor.core.publisher.Flux<T> flux,
                                                           java.lang.Iterable<? extends T> iterable)

Remove all the values in the supplied Iterable from the Flux

Parameters:

flux - Flux to filter values out

iterable - Iterable of values to remove

Returns:

Flux with values in the supplied iterable removed

retainAll

public static <T> reactor.core.publisher.Flux<T> retainAll(reactor.core.publisher.Flux<T> flux,
                                                           java.lang.Iterable<? extends T> iterable)

Retain only the values in the supplied Iterable in the generated Flux

Parameters:

flux - Flux to filter values from

iterable - Values to retain

Returns:

Flux with only values in Iterable retained (if present in original Flux)

intersperse

public static <T> reactor.core.publisher.Flux<T> intersperse(reactor.core.publisher.Flux<T> flux,
                                                             T value)

Intersperse the supplied value throughout this Flux

Parameters:

flux - Flux to intersperse value in

value - Value to intersperse

Returns:

Flux with value interspersed

sliding

public static <T> reactor.core.publisher.Flux<ListX<T>> sliding(reactor.core.publisher.Flux<T> flux,
                                                                int windowSize,
                                                                int increment)

Create a sliding view over this Sequence

 {
  
  List<List<Integer>> list = Fluxes.sliding(Flux.just(1, 2, 3, 4, 5, 6),3, 2).collect(Collectors.toList());
 
  assertThat(list.get(0), hasItems(1, 2, 3));
  assertThat(list.get(1), hasItems(3, 4, 5));
 
 
 
 

Parameters:

flux - Flux to create a sliding window over

windowSize - number of elements in each batch

increment - for each window

Returns:

Flux with sliding view

grouped

public static <T> reactor.core.publisher.Flux<ListX<T>> grouped(reactor.core.publisher.Flux<T> flux,
                                                                int size)

Group elements in a Flux

 {
  
  List<List<Integer>> list = Fluxes.grouped(Flux.just(1, 2, 3, 4, 5, 6),3).collect(Collectors.toList());
 
  //[[1, 2, 3],[4, 5, 6]]
  
 
 

Parameters:

flux - Flux to group

size - Group size

Returns:

Grouped Flux

grouped

public static <T,K,A,D> reactor.core.publisher.Flux<org.jooq.lambda.tuple.Tuple2<K,D>> grouped(reactor.core.publisher.Flux<T> flux,
                                                                                               java.util.function.Function<? super T,? extends K> classifier,
                                                                                               java.util.stream.Collector<? super T,A,D> downstream)

Classify a Flux by the supplied classifying function, materialize into collection with the supplied Collector

Parameters:

flux - Flux to group

classifier - Classifying function

downstream - Collector to build grouping collection

Returns:

Grouped flux

grouped

public static <T,K,A,D> reactor.core.publisher.Flux<org.jooq.lambda.tuple.Tuple2<K,D>> grouped(reactor.core.publisher.Flux<T> flux,
                                                                                               java.util.function.Function<? super T,? extends K> classifier)

Classify a Flux by the supplied classifying function.

Parameters:

flux - Flux to group

classifier - Classifying function

Returns:

Grouped flux

zipWithIndex

public static <T> reactor.core.publisher.Flux<org.jooq.lambda.tuple.Tuple2<T,java.lang.Long>> zipWithIndex(reactor.core.publisher.Flux<T> stream)

Add an index to the supplied Flux

  
    Fluxes.zipWithIndex(Flux.just("a", "b"))
    
    //Flux[Tuple2["a",0l],Tuple2["b",1l]]]
 
 

Parameters:

flux - Flux to add index too

Returns:

Flux with index

deleteBetween

public static final <T> reactor.core.publisher.Flux<T> deleteBetween(reactor.core.publisher.Flux<T> stream,
                                                                     int start,
                                                                     int end)

Delete elements between given indexes in a Flux

  
 List<String> result =    Fluxes.deleteBetween(Flux.just(1,2,3,4,5,6),2,4)
                                           .map(it ->it+"!!")
                                           .collect(Collectors.toList())
                                           .block();
    
           assertThat(result,equalTo(Arrays.asList("1!!","2!!","5!!","6!!")));
 
 

Parameters:

start - index

end - index

Returns:

Stream with elements removed

insertAt

public static final <T> reactor.core.publisher.Flux<T> insertAt(reactor.core.publisher.Flux<T> stream,
                                                                int pos,
                                                                T... values)

Insert data into a Flux at given position

   
 List<String> result =    Fluxes.insertAt(Flux.just(1,2,3),1,100,200,300)
                                       .map(it ->it+"!!")
                                       .collect(Collectors.toList())
                                       .block();
    
          assertThat(result,equalTo(Arrays.asList("1!!","100!!","200!!","300!!","2!!","3!!")));
 
 
 

Parameters:

pos - to insert data at

values - to insert

Returns:

Stream with new data inserted

groupedStatefullyUntil

public static final <T> reactor.core.publisher.Flux<ListX<T>> groupedStatefullyUntil(reactor.core.publisher.Flux<T> flux,
                                                                                     java.util.function.BiPredicate<ListX<? super T>,? super T> predicate)

Create Flux of ListX where each ListX is populated while the supplied bipredicate holds. The bipredicate recieves the ListX from the last window as well as the current value and can choose to aggregate the current value or create a new window

  
 assertThat(Fluxes.groupedStatefullUntil(Flux.just(1,2,3,4,5,6),(s,i)-> s.contains(4) ? true : false)
              .toList().size(),equalTo(5));
 
 

Parameters:

flux - Flux to group

predicate - Window while true

Returns:

Flux windowed while predicate holds