public interface FutureStream<U> extends LazySimpleReactStream<U>, LazyStream<U>, ReactiveSeq<U>, LazyToQueue<U>, ConfigurableStream<U,FastFuture<U>>, FutureStreamSynchronousPublisher<U>
ReactiveSeq.Instances| Modifier and Type | Method and Description |
|---|---|
default OperationsOnFutures<U> |
actOnFutures()
FutureStream operators act on the results of the previous stage by default.
|
default AnyMSeq<Witness.reactiveSeq,U> |
anyM() |
default FutureStream<U> |
append(java.lang.Iterable<? extends U> other) |
default FutureStream<U> |
append(U... values)
Append values to the take of this ReactiveSeq
|
default FutureStream<U> |
append(U value) |
default FutureStream<U> |
appendS(java.util.stream.Stream<? extends U> other)
Append Stream to this ReactiveSeq
|
default FutureStream<U> |
async() |
default FutureStream<U> |
boundedWaitFree(int size)
Use an Agrona ManyToOneConcurrentArrayQueue for the next operations (wait-free, mechanical sympathy).
|
static LazyReact |
builder() |
static LazyReact |
builder(int maxActiveTasks,
java.util.concurrent.Executor exec) |
default FutureStream<U> |
capture(java.util.function.Consumer<java.lang.Throwable> errorHandler)
React capture
While onFail is used for disaster recovery (when it is possible to
recover) - capture is used to capture those occasions where the full
pipeline has failed and is unrecoverable.
|
default <U> FutureStream<U> |
cast(java.lang.Class<? extends U> type)
Cast all elements in a reactiveStream to a given type, possibly throwing a
ClassCastException. |
default ReactiveSeq<U> |
changes() |
default java.util.Iterator<java.util.Collection<U>> |
chunkLastReadIterator() |
default FutureStream<java.util.Collection<U>> |
chunkSinceLastRead() |
default void |
closeAll()
closes all open queues.
|
default <R> FutureStream<R> |
coflatMap(java.util.function.Function<? super ReactiveSeq<U>,? extends R> fn)
coflatMap pattern, can be used to perform maybe reductions / collections / folds and other terminal operations
|
default <R> R |
collect(java.util.function.Supplier<R> supplier,
java.util.function.BiConsumer<R,? super U> accumulator,
java.util.function.BiConsumer<R,R> combiner)
Performs a mutable
reduction operation on the elements of this reactiveStream.
|
default <A,R> FutureStream<R> |
collectSeq(java.util.stream.Collector<? super U,A,R> c) |
default <R,A> FutureStream<R> |
collectStream(java.util.stream.Collector<? super U,A,R> collector) |
default FutureStream<ReactiveSeq<U>> |
combinations()
FutureStream.of(1,2,3).combinations()
//FutureStream[Stream[],Stream[1],Stream[2],Stream[3].Stream[1,2],Stream[1,3],Stream[2,3]
,Stream[1,2,3]]
|
default FutureStream<ReactiveSeq<U>> |
combinations(int size)
ReactiveSeq.of(1,2,3).combinations(2)
//ReactiveSeq[ReactiveSeq[1,2],ReactiveSeq[1,3],ReactiveSeq[2,3]]
|
default FutureStream<U> |
combine(java.util.function.BiPredicate<? super U,? super U> predicate,
java.util.function.BinaryOperator<U> op)
Combine two adjacent elements in a traversable using the supplied BinaryOperator
This is a stateful grouping and reduction operation.
|
default FutureStream<U> |
combine(Monoid<U> op,
java.util.function.BiPredicate<? super U,? super U> predicate)
Combine two adjacent elements in a traversable using the supplied BinaryOperator
This is a stateful grouping and reduction operation.
|
default FutureStream<U> |
complete(java.lang.Runnable fn)
Provide a simpler method to handle stream or async operations.
|
default FutureStream<U> |
concat(java.util.stream.Stream<? extends U> other)
Concatenate two streams operating on the underlying futures rather than results.
|
default FutureStream<U> |
concat(U... other)
Concatenate two streams.
|
default FutureStream<U> |
concat(U other)
Concatenate two streams operating on the underlying futures rather than results.
|
default FutureStream<U> |
control(java.util.function.Function<java.util.function.Supplier<U>,java.util.function.Supplier<U>> fn)
Allows clients to control the emission of data for the next phase of the
Stream.
|
default SimpleReactStream<U> |
convertToSimpleReact()
Convert between an Lazy and Eager SimpleReact Stream, can be used to take
advantages of each approach during a singleUnsafe Stream
Allows callers to take advantage of functionality only available in
SimpleReactStreams such as allOf
|
default java.util.List<FutureStream<U>> |
copy(int times)
Copy this Stream the specified number of times
|
default long |
count()
Returns the count of elements in this reactiveStream.
|
default <U1> FutureStream<org.jooq.lambda.tuple.Tuple2<U,U1>> |
crossApply(java.util.function.Function<? super U,? extends java.lang.Iterable<? extends U1>> function) |
default <T> FutureStream<org.jooq.lambda.tuple.Tuple2<U,T>> |
crossJoin(ReactiveSeq<? extends T> other)
Cross join 2 streams into one.
|
default FutureStream<U> |
cycle()
Create a Stream that infinitely cycles this Stream
|
default FutureStream<U> |
cycle(long times)
Create a Stream that finitely cycles this Stream, provided number of times
|
default FutureStream<U> |
cycle(Monoid<U> m,
long times)
Convert to a Stream with the result of a reduction operation repeated
specified times
|
default FutureStream<U> |
cycleUntil(java.util.function.Predicate<? super U> predicate)
Repeat in a Stream until specified predicate holds
|
default FutureStream<U> |
cycleWhile(java.util.function.Predicate<? super U> predicate)
Repeat in a Stream while specified predicate holds
|
default FutureStream<U> |
debounce(long time,
java.util.concurrent.TimeUnit unit)
Can be used to debounce (accept a singleUnsafe data point from a unit of time)
data.
|
default FutureStream<U> |
deleteBetween(int start,
int end)
Delete elements between given indexes in a Stream
|
default FutureStream<U> |
distinct() |
default <R> FutureStream<U> |
distinct(java.util.function.Function<? super U,? extends R> keyExtractor) |
default FutureStream<U> |
drop(long drop)
assertThat(ReactiveSeq.of(4,3,6,7).drop(2).toList(),equalTo(Arrays.asList(6,7))); |
default FutureStream<U> |
dropRight(int num)
Generate a new Traversable that drops the specified number elements from the take of this Traversable
|
default FutureStream<U> |
dropUntil(java.util.function.Predicate<? super U> p)
Generate a new Traversable that drops elements from this Traversable until the predicate holds
|
default FutureStream<U> |
dropWhile(java.util.function.Predicate<? super U> p)
Generate a new Traversable that drops elements from this Traversable as long as the predicate holds
|
default org.jooq.lambda.tuple.Tuple2<ReactiveSeq<U>,ReactiveSeq<U>> |
duplicate()
Duplicate a Stream, buffers intermediate values, leaders may change
positions so a limit can be safely applied to the leading reactiveStream.
|
default org.jooq.lambda.tuple.Tuple2<ReactiveSeq<U>,ReactiveSeq<U>> |
duplicate(java.util.function.Supplier<java.util.Deque<U>> bufferFactory) |
default org.jooq.lambda.tuple.Tuple2<FutureStream<U>,FutureStream<U>> |
duplicateFutureStream() |
default FutureStream<org.jooq.lambda.tuple.Tuple2<U,java.lang.Long>> |
elapsed()
FutureStream.of(1,2,3,4,5)
.elapsed()
.forEach(System.out::println);
|
default boolean |
endsWith(java.util.stream.Stream<U> stream)
assertTrue(ReactiveSeq.of(1,2,3,4,5,6)
.endsWith(Stream.of(5,6)));
|
default boolean |
endsWithIterable(java.lang.Iterable<U> iterable)
assertTrue(ReactiveSeq.of(1,2,3,4,5,6)
.endsWith(Arrays.asList(5,6)));
|
default <R> FutureStream<R> |
fanOut(java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path1,
java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path2) |
default <R> FutureStream<R> |
fanOut(java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path1,
java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path2,
java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path3) |
default <R> ReactiveSeq<R> |
fanOut(java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path1,
java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path2,
java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path3,
java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path4) |
default <R1,R2,R3> FutureStream<R3> |
fanOutZipIn(java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R1>> path1,
java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R2>> path2,
java.util.function.BiFunction<? super R1,? super R2,? extends R3> zipFn) |
default <R1,R2,R3,R4> |
fanOutZipIn(java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R1>> path1,
java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R2>> path2,
java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R3>> path3,
Fn3<? super R1,? super R2,? super R3,? extends R4> zipFn) |
default <R1,R2,R3,R4,R5> |
fanOutZipIn(java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R1>> path1,
java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R2>> path2,
java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R3>> path3,
java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R4>> path4,
Fn4<? super R1,? super R2,? super R3,? super R4,? extends R5> zipFn) |
default FutureStream<U> |
filter(java.util.function.Predicate<? super U> p)
Removes elements that do not match the supplied predicate from the
dataflow
|
default FutureStream<U> |
filterNot(java.util.function.Predicate<? super U> fn)
Remove any elements for which the predicate holds (inverse operation to filter)
e.g.
|
default java.util.Optional<U> |
findAny() |
default java.util.Optional<U> |
findFirst() |
default Either<java.lang.Throwable,U> |
findFirstOrError()
Lazy / reactiveBuffer look up of first value , capturing the first error, if one occurs.
|
default Maybe<U> |
findOne()
Lazy / reactiveBuffer analogue of findFirst / findAny from JDK
For push based reactiveBuffer-streams (created via Spouts.XXX) data will be pushed to the returned Maybe on arrival.
|
default U |
firstValue()
assertThat(ReactiveSeq.of(1,2,3,4)
.map(u->{throw new RuntimeException();})
.recover(e->"hello")
.firstValue(),equalTo("hello"));
|
default FutureStream<U> |
fixedDelay(long time,
java.util.concurrent.TimeUnit unit)
Apply a fixed delay before emitting elements to the next phase of the
Stream.
|
default <R> FutureStream<R> |
flatMap(java.util.function.Function<? super U,? extends java.util.stream.Stream<? extends R>> flatFn)
Allows aggregate values in a Stream to be flatten into a singleUnsafe Stream.
|
default <R> FutureStream<R> |
flatMapAnyM(java.util.function.Function<? super U,AnyM<Witness.stream,? extends R>> flatFn)
Allows flatMap return type to be any Monad type
|
default <R> FutureStream<R> |
flatMapCompletableFuture(java.util.function.Function<? super U,java.util.concurrent.CompletableFuture<? extends R>> flatFn)
Perform a flatMap operation where the CompletableFuture type returned is flattened from the resulting Stream
If in async mode this operation is performed asyncrhonously
If in sync mode this operation is performed synchronously
|
default <R> FutureStream<R> |
flatMapI(java.util.function.Function<? super U,? extends java.lang.Iterable<? extends R>> fn)
FlatMap where the result is a Collection, flattens the resultant
collections into the host ReactiveSeq
|
default <R> FutureStream<R> |
flatMapP(java.util.function.Function<? super U,? extends org.reactivestreams.Publisher<? extends R>> fn) |
default <R> FutureStream<R> |
flatMapP(int maxConcurrency,
java.util.function.Function<? super U,? extends org.reactivestreams.Publisher<? extends R>> fn) |
default <R> FutureStream<R> |
flatMapStream(java.util.function.Function<? super U,java.util.stream.BaseStream<? extends R,?>> fn)
flatMap operation
|
default java.util.stream.DoubleStream |
flatMapToDouble(java.util.function.Function<? super U,? extends java.util.stream.DoubleStream> mapper) |
default java.util.stream.IntStream |
flatMapToInt(java.util.function.Function<? super U,? extends java.util.stream.IntStream> mapper) |
default java.util.stream.LongStream |
flatMapToLong(java.util.function.Function<? super U,? extends java.util.stream.LongStream> mapper) |
static <T1> FutureStream<T1> |
flatten(ReactiveSeq<? extends java.util.stream.Stream<? extends T1>> nested) |
default FutureStream<U> |
fold(Monoid<U> monoid) |
default <R> Future<R> |
foldFuture(java.util.function.Function<? super FoldableTraversable<U>,? extends R> fn) |
default <R> R |
foldRight(R identity,
java.util.function.BiFunction<? super U,? super R,? extends R> accumulator)
Immutable reduction from right to left
|
default void |
forEach(java.util.function.Consumer<? super U> c)
Performs an action for each element of this Stream.
|
default <X extends java.lang.Throwable> |
forEach(java.util.function.Consumer<? super U> consumerElement,
java.util.function.Consumer<? super java.lang.Throwable> consumerError)
Perform a forEach operation over the Stream capturing any elements and errors in the supplied consumers,
|
default <X extends java.lang.Throwable> |
forEach(java.util.function.Consumer<? super U> consumerElement,
java.util.function.Consumer<? super java.lang.Throwable> consumerError,
java.lang.Runnable onComplete)
Perform a forEach operation over the Stream capturing any elements and errors in the supplied consumers
when the entire Stream has been processed an onComplete event will be recieved.
|
default <X extends java.lang.Throwable> |
forEach(long numberOfElements,
java.util.function.Consumer<? super U> consumer)
Perform a forEach operation over the Stream, without closing it, consuming only the specified number of elements from
the Stream, at this time.
|
default <X extends java.lang.Throwable> |
forEach(long numberOfElements,
java.util.function.Consumer<? super U> consumer,
java.util.function.Consumer<? super java.lang.Throwable> consumerError)
Perform a forEach operation over the Stream without closing it, capturing any elements and errors in the supplied consumers, but only consuming
the specified number of elements from the Stream, at this time.
|
default <X extends java.lang.Throwable> |
forEach(long numberOfElements,
java.util.function.Consumer<? super U> consumer,
java.util.function.Consumer<? super java.lang.Throwable> consumerError,
java.lang.Runnable onComplete)
Perform a forEach operation over the Stream without closing it, capturing any elements and errors in the supplied consumers, but only consuming
the specified number of elements from the Stream, at this time.
|
default <R1,R> FutureStream<R> |
forEach2(java.util.function.Function<? super U,? extends java.util.stream.BaseStream<R1,?>> stream1,
java.util.function.BiFunction<? super U,? super R1,? extends R> yieldingFunction)
Perform a two level nested internal iteration over this Stream and the
supplied reactiveStream
|
default <R1,R> FutureStream<R> |
forEach2(java.util.function.Function<? super U,? extends java.util.stream.BaseStream<R1,?>> stream1,
java.util.function.BiFunction<? super U,? super R1,java.lang.Boolean> filterFunction,
java.util.function.BiFunction<? super U,? super R1,? extends R> yieldingFunction)
Perform a two level nested internal iteration over this Stream and the
supplied reactiveStream
|
default <R1,R2,R> FutureStream<R> |
forEach3(java.util.function.Function<? super U,? extends java.util.stream.BaseStream<R1,?>> stream1,
java.util.function.BiFunction<? super U,? super R1,? extends java.util.stream.BaseStream<R2,?>> stream2,
Fn3<? super U,? super R1,? super R2,? extends R> yieldingFunction)
Perform a three level nested internal iteration over this Stream and the
supplied streams
|
default <R1,R2,R> FutureStream<R> |
forEach3(java.util.function.Function<? super U,? extends java.util.stream.BaseStream<R1,?>> stream1,
java.util.function.BiFunction<? super U,? super R1,? extends java.util.stream.BaseStream<R2,?>> stream2,
Fn3<? super U,? super R1,? super R2,java.lang.Boolean> filterFunction,
Fn3<? super U,? super R1,? super R2,? extends R> yieldingFunction)
Perform a three level nested internal iteration over this Stream and the
supplied streams
|
default <R1,R2,R3,R> |
forEach4(java.util.function.Function<? super U,? extends java.util.stream.BaseStream<R1,?>> stream1,
java.util.function.BiFunction<? super U,? super R1,? extends java.util.stream.BaseStream<R2,?>> stream2,
Fn3<? super U,? super R1,? super R2,? extends java.util.stream.BaseStream<R3,?>> stream3,
Fn4<? super U,? super R1,? super R2,? super R3,? extends R> yieldingFunction)
Perform a four level nested internal iteration over this Stream and the
supplied streams
|
default <R1,R2,R3,R> |
forEach4(java.util.function.Function<? super U,? extends java.util.stream.BaseStream<R1,?>> stream1,
java.util.function.BiFunction<? super U,? super R1,? extends java.util.stream.BaseStream<R2,?>> stream2,
Fn3<? super U,? super R1,? super R2,? extends java.util.stream.BaseStream<R3,?>> stream3,
Fn4<? super U,? super R1,? super R2,? super R3,java.lang.Boolean> filterFunction,
Fn4<? super U,? super R1,? super R2,? super R3,? extends R> yieldingFunction)
Perform a four level nested internal iteration over this Stream and the
supplied streams
|
default void |
forEachAsync(java.util.function.Consumer<? super U> action)
A potentially non-blocking analog of
ReactiveSeq.forEach(java.util.function.Consumer<? super T>). |
default void |
forEachOrdered(java.util.function.Consumer<? super U> action) |
default <R> FutureStream<R> |
fromStream(java.util.stream.Stream<R> stream) |
default <R> FutureStream<R> |
fromStreamOfFutures(java.util.stream.Stream<FastFuture<R>> stream) |
default ReactiveStreamsTerminalFutureOperations<U> |
futureOperations() |
LazyReact |
getSimpleReact() |
Continueable |
getSubscription() |
default <C extends java.util.Collection<U>> |
group(java.util.function.Function<java.util.function.Supplier<U>,java.util.function.Supplier<C>> fn)
Batch elements into a Stream of collections with user defined function
|
default FutureStream<ListX<U>> |
grouped(int groupSize)
Group elements in a Stream
|
default <C extends java.util.Collection<? super U>> |
grouped(int size,
java.util.function.Supplier<C> supplier)
Batch the elements in this reactiveStream into Collections of specified size The
type of Collection is determined by the specified supplier
|
default FutureStream<ListX<U>> |
groupedBySizeAndTime(int size,
long time,
java.util.concurrent.TimeUnit unit)
Batch elements by size into a List
|
default <C extends java.util.Collection<? super U>> |
groupedBySizeAndTime(int size,
long time,
java.util.concurrent.TimeUnit unit,
java.util.function.Supplier<C> factory)
Batch elements by size into a toX created by the supplied factory
|
default <C extends java.util.Collection<? super U>,R> |
groupedBySizeAndTime(int size,
long time,
java.util.concurrent.TimeUnit unit,
java.util.function.Supplier<C> factory,
java.util.function.Function<? super C,? extends R> finalizer) |
default FutureStream<ListX<U>> |
groupedByTime(long time,
java.util.concurrent.TimeUnit unit)
Organise elements in a Stream into a Collections based on the time period
they pass through this stage
|
default <C extends java.util.Collection<? super U>> |
groupedByTime(long time,
java.util.concurrent.TimeUnit unit,
java.util.function.Supplier<C> factory)
Organise elements in a Stream into a Collections based on the time period
they pass through this stage
|
default <C extends java.util.Collection<? super U>,R> |
groupedByTime(long time,
java.util.concurrent.TimeUnit unit,
java.util.function.Supplier<C> factory,
java.util.function.Function<? super C,? extends R> finalizer) |
default <C extends java.util.Collection<U>,R> |
groupedStatefullyUntil(java.util.function.BiPredicate<C,? super U> predicate,
java.util.function.Supplier<C> factory,
java.util.function.Function<? super C,? extends R> finalizer) |
default FutureStream<ListX<U>> |
groupedStatefullyUntil(java.util.function.BiPredicate<ListX<? super U>,? super U> predicate)
Create ReactiveSeq of ListX where
each ListX is populated while the supplied bipredicate holds.
|
default <C extends java.util.Collection<U>,R> |
groupedStatefullyWhile(java.util.function.BiPredicate<C,? super U> predicate,
java.util.function.Supplier<C> factory,
java.util.function.Function<? super C,? extends R> finalizer) |
default FutureStream<ListX<U>> |
groupedStatefullyWhile(java.util.function.BiPredicate<ListX<? super U>,? super U> predicate) |
default FutureStream<ListX<U>> |
groupedUntil(java.util.function.Predicate<? super U> predicate)
Create a ReactiveSeq batched by List, where each batch is populated until
the predicate holds
|
default <C extends java.util.Collection<? super U>> |
groupedUntil(java.util.function.Predicate<? super U> predicate,
java.util.function.Supplier<C> factory)
Create a ReactiveSeq batched by a Collection, where each batch is populated
until the predicate holds
|
default FutureStream<ListX<U>> |
groupedWhile(java.util.function.Predicate<? super U> predicate)
Create a ReactiveSeq batched by List, where each batch is populated while
the predicate holds
|
default <C extends java.util.Collection<? super U>> |
groupedWhile(java.util.function.Predicate<? super U> predicate,
java.util.function.Supplier<C> factory)
Create a ReactiveSeq batched by a Collection, where each batch is populated
while the predicate holds
|
default HotStream<U> |
hotStream()
Turns this FutureStream into a HotStream, a connectable Stream, being executed on a thread on the
in it's current task executor, that is producing data
|
default <T> FutureStream<org.jooq.lambda.tuple.Tuple2<U,T>> |
innerJoin(java.util.stream.Stream<? extends T> other,
java.util.function.BiPredicate<? super U,? super T> predicate)
Inner join 2 streams into one.
|
default FutureStream<U> |
insertAt(int pos,
U... values)
Insert data into a reactiveStream at given position
|
default FutureStream<U> |
insertAtS(int pos,
java.util.stream.Stream<U> stream)
Insert a Stream into the middle of this reactiveStream at the specified position
|
default FutureStream<U> |
intersperse(U value)
Returns a reactiveStream with a given value interspersed between any two values
of this reactiveStream.
|
default boolean |
isParallel() |
default CloseableIterator<U> |
iterator() |
default FutureStream<U> |
jitter(long jitterInNanos)
Introduce a random delay between events in a reactiveStream Can be used to
prevent behaviour synchronizing within a system
|
default <R> FutureStream<R> |
jool(java.util.function.Function<? super org.jooq.lambda.Seq<U>,? extends org.jooq.lambda.Seq<R>> mapper) |
default <T> FutureStream<org.jooq.lambda.tuple.Tuple2<U,T>> |
leftOuterJoin(java.util.stream.Stream<? extends T> other,
java.util.function.BiPredicate<? super U,? super T> predicate)
Left outer join 2 streams into one.
|
default FutureStream<U> |
limit(long maxSize)
assertThat(ReactiveSeq.of(4,3,6,7).limit(2).toList(),equalTo(Arrays.asList(4,3)); |
default FutureStream<U> |
limit(long time,
java.util.concurrent.TimeUnit unit)
Return all elements until specified time period has elapsed
|
default FutureStream<U> |
limitLast(int num)
Limit results to the last x elements in a ReactiveSeq
|
default FutureStream<U> |
limitUntil(java.util.function.Predicate<? super U> predicate)
Returns a reactiveStream limited to all elements for which a predicate evaluates
to false.
|
default ReactiveSeq<U> |
limitUntilClosed(java.util.function.Predicate<? super U> p) |
default FutureStream<U> |
limitWhile(java.util.function.Predicate<? super U> predicate)
Returns a reactiveStream limited to all elements for which a predicate evaluates
to true.
|
default FutureStream<U> |
limitWhileClosed(java.util.function.Predicate<? super U> predicate) |
default <R> FutureStream<R> |
map(java.util.function.Function<? super U,? extends R> mapper)
Transform this functor using the supplied transformation function
|
default java.util.stream.DoubleStream |
mapToDouble(java.util.function.ToDoubleFunction<? super U> mapper) |
default java.util.stream.IntStream |
mapToInt(java.util.function.ToIntFunction<? super U> mapper) |
default java.util.stream.LongStream |
mapToLong(java.util.function.ToLongFunction<? super U> mapper) |
FutureStream<U> |
maxActive(int concurrentTasks)
Configure the max active concurrent tasks.
|
default FutureStream<U> |
merge(Adapter<U>... adapters) |
default <R> FutureStream<R> |
mergeLatest(FutureStream<?>... streams)
Merges this reactiveStream and the supplied Streams into a singleUnsafe Stream where the next value
is the next returned across any of the involved Streams.
|
default FutureStream<U> |
mergeP(org.reactivestreams.Publisher<U>... publishers)
A potentially asynchronous merge operation where data from each publisher may arrive out of order (if publishers
are configured to publish asynchronously, users can use the overloaded @see
IterableFunctor#mergePublisher(Collection, QueueFactory)
method to forEachAsync asynchronously also. |
default FutureStream<U> |
mergeP(QueueFactory<U> factory,
org.reactivestreams.Publisher<U>... publishers)
A potentially asynchronous merge operation where data from each publisher may arrive out of order (if publishers
are configured to publish asynchronously.
|
default ListX<ReactiveSeq<U>> |
multicast(int num)
Broadcast the contents of this Stream to multiple downstream Streams (determined by supplier parameter).
|
static <T1> FutureStream<T1> |
narrow(FutureStream<? extends T1> broad) |
default FutureStream<U> |
notNull()
Filter elements retaining only values which are not null
|
default <U> FutureStream<U> |
ofType(java.lang.Class<? extends U> type)
Keep only those elements in a reactiveStream that are of a given type.
|
default FutureStream<U> |
onClose(java.lang.Runnable closeHandler) |
default FutureStream<U> |
onEmpty(U value)
Produce this reactiveStream, or an alternative reactiveStream from the
value, in case this reactiveStream is empty. |
default FutureStream<U> |
onEmptyGet(java.util.function.Supplier<? extends U> supplier)
Produce this reactiveStream, or an alternative reactiveStream from the
supplier, in case this reactiveStream is empty. |
default FutureStream<U> |
onEmptySwitch(java.util.function.Supplier<? extends java.util.stream.Stream<U>> switchTo)
If this Stream is empty one it with a another Stream
|
default <X extends java.lang.Throwable> |
onEmptyThrow(java.util.function.Supplier<? extends X> supplier)
Produce this reactiveStream, or an alternative reactiveStream from the
supplier, in case this reactiveStream is empty. |
default FutureStream<U> |
onePer(long time,
java.util.concurrent.TimeUnit unit)
Slow emissions down, emiting one element per specified time period
|
default FutureStream<U> |
onFail(java.lang.Class<? extends java.lang.Throwable> exceptionClass,
java.util.function.Function<? super SimpleReactFailedStageException,? extends U> fn)
Recover for a particular class of exceptions only.
|
default FutureStream<U> |
onFail(java.util.function.Function<? super SimpleReactFailedStageException,? extends U> fn)
React onFail
Define a function that can be used to recover from exceptions during the
preceeding stage of the dataflow.
|
default <U1> FutureStream<org.jooq.lambda.tuple.Tuple2<U,U1>> |
outerApply(java.util.function.Function<? super U,? extends java.lang.Iterable<? extends U1>> function) |
default FutureStream<U> |
parallel() |
default <R> FutureStream<R> |
parallel(java.util.concurrent.ForkJoinPool fj,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> fn) |
default <R> FutureStream<R> |
parallel(java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> fn) |
default <R> FutureStream<R> |
parallelFanOut(java.util.concurrent.ForkJoinPool fj,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path1,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path2) |
default <R> FutureStream<R> |
parallelFanOut(java.util.concurrent.ForkJoinPool fj,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path1,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path2,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path3) |
default <R> FutureStream<R> |
parallelFanOut(java.util.concurrent.ForkJoinPool fj,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path1,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path2,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path3,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path4) |
default <R1,R2,R3> FutureStream<R3> |
parallelFanOutZipIn(java.util.concurrent.ForkJoinPool fj,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R1>> path1,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R2>> path2,
java.util.function.BiFunction<? super R1,? super R2,? extends R3> zipFn) |
default <R1,R2,R3,R4> |
parallelFanOutZipIn(java.util.concurrent.ForkJoinPool fj,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R1>> path1,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R2>> path2,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R3>> path3,
Fn3<? super R1,? super R2,? super R3,? extends R4> zipFn) |
default <R1,R2,R3,R4,R5> |
parallelFanOutZipIn(java.util.concurrent.ForkJoinPool fj,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R1>> path1,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R2>> path2,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R3>> path3,
java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R4>> path4,
Fn4<? super R1,? super R2,? super R3,? super R4,? extends R5> zipFn) |
default org.jooq.lambda.tuple.Tuple2<ReactiveSeq<U>,ReactiveSeq<U>> |
partition(java.util.function.Predicate<? super U> predicate)
Partition a reactiveStream in two given a predicate.
|
default org.jooq.lambda.tuple.Tuple2<FutureStream<U>,FutureStream<U>> |
partitionFutureStream(java.util.function.Predicate<? super U> predicate)
Partition an FutureStream into two LazyFutureStreams given a
predicate.
|
default FutureStream<U> |
peek(java.util.function.Consumer<? super U> consumer)
Peek asynchronously at the results in the current stage.
|
default FutureStream<U> |
peekSync(java.util.function.Consumer<? super U> consumer)
Synchronous peek operator
|
default FutureStream<ReactiveSeq<U>> |
permutations()
Generate the permutations based on values in the FutureStream
Makes use of Streamable to store intermediate stages in a toX
|
default FutureStream<U> |
prepend(java.lang.Iterable<? extends U> other) |
default FutureStream<U> |
prepend(U... values)
Prepend given values to the skip of the Stream
|
default FutureStream<U> |
prependS(java.util.stream.Stream<? extends U> other)
Prepend Stream to this ReactiveSeq
|
default org.jooq.lambda.tuple.Tuple4<ReactiveSeq<U>,ReactiveSeq<U>,ReactiveSeq<U>,ReactiveSeq<U>> |
quadruplicate()
Makes four copies of a Stream Buffers intermediate values, leaders may
change positions so a limit can be safely applied to the leading reactiveStream.
|
default org.jooq.lambda.tuple.Tuple4<ReactiveSeq<U>,ReactiveSeq<U>,ReactiveSeq<U>,ReactiveSeq<U>> |
quadruplicate(java.util.function.Supplier<java.util.Deque<U>> bufferFactory) |
default <EX extends java.lang.Throwable> |
recover(java.lang.Class<EX> exceptionClass,
java.util.function.Function<? super EX,? extends U> fn)
Recover from a particular exception type
|
default FutureStream<U> |
recover(java.util.function.Function<? super java.lang.Throwable,? extends U> fn)
Recover from an exception with an alternative value
|
default java.util.Optional<U> |
reduce(java.util.function.BinaryOperator<U> accumulator)
An equivalent function to
Stream.reduce(BinaryOperator) |
default <T> T |
reduce(T identity,
java.util.function.BiFunction<T,? super U,T> accumulator)
An equivalent function to
Stream.reduce(Object, BinaryOperator) |
default <T> T |
reduce(T identity,
java.util.function.BiFunction<T,? super U,T> accumulator,
java.util.function.BinaryOperator<T> combiner)
An equivalent function to
Stream.reduce(Object, BiFunction, BinaryOperator) |
default U |
reduce(U identity,
java.util.function.BinaryOperator<U> accumulator)
An equivalent function to
Stream.reduce(Object, BinaryOperator) |
default FutureStream<U> |
reduceAll(U identity,
java.util.function.BinaryOperator<U> accumulator) |
default FutureStream<U> |
remove(U t)
Remove all occurances of the specified element from the Stream
|
default FutureStream<U> |
removeAll(U... values)
Remove all supplied elements from this filterable
|
default FutureStream<U> |
removeAllI(java.lang.Iterable<? extends U> it)
Remove all elements in the supplied Iterable from this filterable
|
default FutureStream<U> |
removeAllS(java.util.stream.Stream<? extends U> stream)
Remove all elements in the supplied Stream from this filterable
|
default FutureStream<U> |
retainAll(U... values)
Retain only the supplied elements in the returned Filters
|
default FutureStream<U> |
retainAllI(java.lang.Iterable<? extends U> it)
Retain only the supplied elements in the returned Filters
|
default FutureStream<U> |
retainAllS(java.util.stream.Stream<? extends U> stream)
Retain only the supplied elements in the returned Filters
|
default <R> FutureStream<R> |
retry(java.util.function.Function<? super U,? extends R> fn)
Retry a transformation if it fails.
|
default <R> FutureStream<R> |
retry(java.util.function.Function<? super U,? extends R> fn,
int retries,
long delay,
java.util.concurrent.TimeUnit timeUnit)
Retry a transformation if it fails.
|
default FutureStream<U> |
reverse()
Reverse a reactiveStream.
|
default <T> FutureStream<org.jooq.lambda.tuple.Tuple2<U,T>> |
rightOuterJoin(java.util.stream.Stream<? extends T> other,
java.util.function.BiPredicate<? super U,? super T> predicate)
Right outer join 2 streams into one.
|
default FutureStream<U> |
scanLeft(Monoid<U> monoid)
Scan left using supplied Monoid
|
default <T> FutureStream<T> |
scanLeft(T seed,
java.util.function.BiFunction<? super T,? super U,? extends T> function)
Scan a reactiveStream to the left.
|
default FutureStream<U> |
scanRight(Monoid<U> monoid)
Scan right
|
default <R> FutureStream<R> |
scanRight(R seed,
java.util.function.BiFunction<? super U,? super R,? extends R> function)
Scan a reactiveStream to the right.
|
default FutureStream<U> |
self(java.util.function.Consumer<FutureStream<U>> consumer)
Give a consumer access to this Stream
|
default FutureStream<U> |
sequential() |
default <K> java.util.Map<K,FutureStream<U>> |
shard(java.util.Map<K,Queue<U>> shards,
java.util.function.Function<? super U,? extends K> sharder)
Break a reactiveStream into multiple Streams based of some characteristic of the
elements of the Stream
e.g.
|
default FutureStream<U> |
shuffle()
Shuffle a reactiveStream
// e.g.
|
default FutureStream<U> |
shuffle(java.util.Random random)
Shuffle a reactiveStream using specified source of randomness
// e.g.
|
default FutureStream<U> |
skip(long n)
assertThat(ReactiveSeq.of(4,3,6,7).skip(2).toList(),equalTo(Arrays.asList(6,7))); |
default FutureStream<U> |
skip(long time,
java.util.concurrent.TimeUnit unit)
Skip all elements until specified time period has passed
|
default FutureStream<U> |
skipLast(int num)
assertThat(ReactiveSeq.of(1,2,3,4,5) .skipLast(2)
.collect(CyclopsCollectors.toList()),equalTo(Arrays.asList(1,2,3)));
|
default FutureStream<U> |
skipUntil(java.util.function.Predicate<? super U> predicate)
Returns a reactiveStream with all elements skipped for which a predicate
evaluates to false.
|
default FutureStream<U> |
skipUntilClosed(java.util.function.Predicate<? super U> p) |
default FutureStream<U> |
skipWhile(java.util.function.Predicate<? super U> predicate)
Returns a reactiveStream with all elements skipped for which a predicate
evaluates to true.
|
default FutureStream<U> |
skipWhileClosed(java.util.function.Predicate<? super U> predicate) |
default FutureStream<U> |
slice(long from,
long to) |
default FutureStream<VectorX<U>> |
sliding(int size)
Create a sliding view over this Stream
|
default FutureStream<VectorX<U>> |
sliding(int size,
int increment)
Create a sliding view over this Stream
|
default FutureStream<U> |
sorted()
assertThat(ReactiveSeq.of(4,3,6,7)).sorted().toList(),equalTo(Arrays.asList(3,4,6,7))); |
default FutureStream<U> |
sorted(java.util.Comparator<? super U> comparator)
assertThat(ReactiveSeq.of(4,3,6,7).sorted((a,b) -> b-a).toList(),equalTo(Arrays.asList(7,6,4,3)));
|
default <R extends java.lang.Comparable<? super R>> |
sorted(java.util.function.Function<? super U,? extends R> function) |
default <U1> FutureStream<U> |
sorted(java.util.function.Function<? super U,? extends U1> function,
java.util.Comparator<? super U1> comparator) |
default org.jooq.lambda.tuple.Tuple2<ReactiveSeq<U>,ReactiveSeq<U>> |
splitAt(int where)
Split at supplied location
|
default org.jooq.lambda.tuple.Tuple2<java.util.Optional<U>,ReactiveSeq<U>> |
splitAtHead()
Split a Stream at it's head (similar to headAndTail)
|
default org.jooq.lambda.tuple.Tuple2<ReactiveSeq<U>,ReactiveSeq<U>> |
splitBy(java.util.function.Predicate<U> splitter)
Split reactiveStream at point where predicate no longer holds
|
default java.util.Spliterator<U> |
spliterator()
Returns a spliterator for the elements of this reactiveStream.
|
default boolean |
startsWith(java.util.stream.Stream<U> iterator)
assertTrue(ReactiveSeq.of(1,2,3,4).startsWith(Stream.of(1,2,3))) |
default boolean |
startsWithIterable(java.lang.Iterable<U> iterable)
assertTrue(ReactiveSeq.of(1,2,3,4).startsWith(Arrays.asList(1,2,3)));
|
default ReactiveSeq<U> |
stream()
Convert this ReactiveSeq into a Stream
|
default void |
subscribe(org.reactivestreams.Subscriber<? super U> s) |
default FutureStream<U> |
subStream(int start,
int end)
Return a Stream with elements before the provided skip index removed, and elements after the provided
take index removed
|
default <R> FutureStream<R> |
switchOnNextValue(java.util.stream.Stream<FutureStream> streams)
Merges this reactiveStream and the supplied Streams into a singleUnsafe Stream where the next value
is the next returned across any of the involved Streams.
|
default FutureStream<U> |
sync() |
default FutureStream<U> |
take(long take)
assertThat(ReactiveSeq.of(4,3,6,7).take(2).toList(),equalTo(Arrays.asList(4,3)); |
default FutureStream<U> |
takeRight(int num)
Generate a new Traversable that takes the specified number elements from the take of this Traversable
|
default FutureStream<U> |
takeUntil(java.util.function.Predicate<? super U> p)
Generate a new Traversable that takes elements from this Traversable until the predicate holds
|
default FutureStream<U> |
takeWhile(java.util.function.Predicate<? super U> p)
Generate a new Traversable that takes elements from this Traversable as long as the predicate holds
|
default <R> FutureStream<R> |
then(java.util.function.Function<? super U,? extends R> fn)
React transform
Unlike 'with' this method is fluent, and returns another Stage Builder
that can represent the next stage in the dataflow.
|
default <R> FutureStream<R> |
then(java.util.function.Function<? super U,? extends R> fn,
java.util.concurrent.Executor service) |
default <R> FutureStream<R> |
thenSync(java.util.function.Function<? super U,? extends R> fn) |
default FutureStream<org.jooq.lambda.tuple.Tuple2<U,java.lang.Long>> |
timestamp()
ReactiveSeq.of(1,2,3,4,5)
.timestamp()
.forEach(System.out::println)
|
default java.lang.Object[] |
toArray() |
default <A> A[] |
toArray(java.util.function.IntFunction<A[]> generator) |
default java.util.concurrent.CompletableFuture<ListX<U>> |
toCompletableFuture() |
default java.util.List<U> |
toList()
Collect the collectable into an
ArrayList. |
default Queue<U> |
toQueue()
Convert the current Stream to a simple-react Queue
|
default java.util.Set<U> |
toSet()
Collect the collectable into a
LinkedHashSet. |
default <R> FutureStream<R> |
trampoline(java.util.function.Function<? super U,? extends Trampoline<? extends R>> mapper)
Performs a map operation that can call a recursive method without running out of stack space
|
default org.jooq.lambda.tuple.Tuple3<ReactiveSeq<U>,ReactiveSeq<U>,ReactiveSeq<U>> |
triplicate()
Triplicates a Stream Buffers intermediate values, leaders may change
positions so a limit can be safely applied to the leading reactiveStream.
|
default org.jooq.lambda.tuple.Tuple3<ReactiveSeq<U>,ReactiveSeq<U>,ReactiveSeq<U>> |
triplicate(java.util.function.Supplier<java.util.Deque<U>> bufferFactory) |
default FutureStream<U> |
unboundedWaitFree()
This is the default setting, internal queues are backed by a ConcurrentLinkedQueue
This operator will return the next stage to using this Queue type if it has been changed
|
default FutureStream<U> |
unordered() |
FutureStream<U> |
withAsync(boolean async) |
<R> FutureStream<R> |
withLastActive(LazyStreamWrapper<R> streamWrapper) |
FutureStream<U> |
withLazyCollector(java.util.function.Supplier<LazyResultConsumer<U>> lazy) |
FutureStream<U> |
withQueueFactory(QueueFactory<U> queue) |
FutureStream<U> |
withSubscription(Continueable sub) |
FutureStream<U> |
withTaskExecutor(java.util.concurrent.Executor e) |
default FutureStream<U> |
xPer(int x,
long time,
java.util.concurrent.TimeUnit unit)
Allows x (specified number of) emissions with a time period before
stopping emmissions until specified time has elapsed since last emission
|
default FutureStream<U> |
zip(java.util.function.BinaryOperator<Zippable<U>> combiner,
Zippable<U> app)
Combine two applicatives together using the provided BinaryOperator (Semigroup, Monoid and Reducer all
extend BinaryOperator - checkout SemigroupK and Monoids for a large number of canned combinations).
|
default <T> FutureStream<org.jooq.lambda.tuple.Tuple2<U,T>> |
zip(java.lang.Iterable<? extends T> other)
Zip (combine) this Zippable with the supplied Iterable combining both into a Tuple2
|
default <T,R> FutureStream<R> |
zip(java.lang.Iterable<? extends T> other,
java.util.function.BiFunction<? super U,? super T,? extends R> zipper)
Zip (combine) this Zippable with the supplied Iterable using the supplied combining function
|
default <S,R> FutureStream<org.jooq.lambda.tuple.Tuple3<U,S,R>> |
zip3(java.lang.Iterable<? extends S> second,
java.lang.Iterable<? extends R> third)
zip 3 Streams into one
|
default <S,U1,R> FutureStream<R> |
zip3(java.lang.Iterable<? extends S> second,
java.lang.Iterable<? extends U1> third,
Fn3<? super U,? super S,? super U1,? extends R> fn3) |
default <T2,T3,T4> FutureStream<org.jooq.lambda.tuple.Tuple4<U,T2,T3,T4>> |
zip4(java.lang.Iterable<? extends T2> second,
java.lang.Iterable<? extends T3> third,
java.lang.Iterable<? extends T4> fourth)
zip 4 Streams into 1
|
default <T2,T3,T4,R> |
zip4(java.lang.Iterable<? extends T2> second,
java.lang.Iterable<? extends T3> third,
java.lang.Iterable<? extends T4> fourth,
Fn4<? super U,? super T2,? super T3,? super T4,? extends R> fn) |
default <U1,R> FutureStream<R> |
zipLatest(org.reactivestreams.Publisher<? extends U1> other,
java.util.function.BiFunction<? super U,? super U1,? extends R> zipper) |
default <T2,R> FutureStream<R> |
zipP(org.reactivestreams.Publisher<? extends T2> publisher,
java.util.function.BiFunction<? super U,? super T2,? extends R> fn)
Zip (combine) this Zippable with the supplied Publisher, using the supplied combining function
|
default <U1> FutureStream<org.jooq.lambda.tuple.Tuple2<U,U1>> |
zipP(org.reactivestreams.Publisher<? extends U1> other) |
default <T> FutureStream<org.jooq.lambda.tuple.Tuple2<U,T>> |
zipS(java.util.stream.Stream<? extends T> other)
Zip 2 streams into one
|
default <T,R> FutureStream<R> |
zipS(java.util.stream.Stream<? extends T> other,
java.util.function.BiFunction<? super U,? super T,? extends R> zipper)
Zip (combine) this Zippable with the supplied Stream, using the supplied combining function
|
default <R> FutureStream<R> |
zipWith(java.lang.Iterable<java.util.function.Function<? super U,? extends R>> fn) |
default FutureStream<org.jooq.lambda.tuple.Tuple2<U,java.lang.Long>> |
zipWithIndex()
Zip a Stream with a corresponding Stream of indexes.
|
default <R> FutureStream<R> |
zipWithP(org.reactivestreams.Publisher<java.util.function.Function<? super U,? extends R>> fn) |
default <R> FutureStream<R> |
zipWithS(java.util.stream.Stream<java.util.function.Function<? super U,? extends R>> fn) |
assureSimpleReactException, copySimpleReactStream, filterSync, flatMapToCompletableFuture, flatMapToCompletableFutureSync, getLastActive, handleExceptions, streamCompletableFutures, withErrorHandlergetErrorHandler, getTaskExecutor, isAsyncempty, freeThread, freeThread, getQueueFactory, getTaskExecutor, isAsync, of, of, parallel, react, react, simpleReactStream, simpleReactStream, simpleReactStream, simpleReactStream, simpleReactStreamFrom, simpleReactStreamFromIterableblock, block, getErrorHandlergetErrorHandler, getLastActive, getLazyCollector, getMaxActive, run, run, runContinuation, runOnCurrent, runThreadallMatch, allTypeclasses, allTypeclasses, ambWith, ambWith, anyMatch, backpressureAware, broadcast, collect, collectAll, combinationsT, combinationsT, concat, concat, concat, concat, concatDoubles, concatInts, concatLongs, coproduct, deferred, deferredI, deferredP, doubles, elementAt, empty, enqueued, enqueued, enqueued, enqueuedAll, enums, enums, enums, enumsFromTo, fill, filterInts, filterLongs, filterLongs, flatMapDoubles, flatMapInts, flatMapLongs, flatten, flattenI, flattenO, foldJool, foldParallel, foldParallel, foldRight, foldRight, foldRightMapToType, format, fromCharSequence, fromDoubleStream, fromIntStream, fromIterable, fromIterator, fromList, fromLongStream, fromPublisher, fromSpliterator, fromStream, generate, generate, get, groupBy, grouped, grouped, headAndTail, hotStream, ints, isEmpty, isReplayable, iterate, iterate, join, join, join, kindCokleisli, kindKleisli, liftM, limitDouble, limitInts, limitLongs, longs, mapDoubles, mapInts, mapLongs, mapM, mapReduce, mapReduce, max, min, narrowK, nested, noneMatch, of, of, ofDoubles, ofInts, ofLongs, ofNullable, oneShotList, oneShotStream, pausableHotStream, permutationsT, prepend, primedHotStream, primedPausableHotStream, print, print, printErr, printOut, product, product, publishTo, publishTo, range, range, rangeLong, rangeLong, reactiveSeq, reduce, reduce, reduce, reversedListOf, reversedOf, schedule, scheduleFixedDelay, scheduleFixedRate, seq, single, single, singleUnsafe, size, skipDoubles, skipInts, skipLongs, tailRec, takeOne, unfold, unit, unzip, unzip3, unzip4, visit, widen, xMatchasFunction, foldFuture, foldLazy, foldTry, forEachSubscribe, forEachSubscribe, forEachSubscribe, runFuture, runLazy, toListX, toSetX, visitcollectors, commonPrefix, commonSuffix, count, countDistinct, countDistinct, countDistinctBy, countDistinctBy, maxBy, minBy, mode, sum, sumDouble, sumInt, sumLong, to, toMap, toMap, toString, toStringfoldLeft, foldLeft, foldLeft, foldLefttraversable, unitIteratorgroupedStatefullyUntilT, groupedT, groupedUntilT, groupedWhileT, slidingT, slidingTadd, addToQueue, getPopulator, returnPopulator, toQueue, toQueuegetQueueFactorycancel, forwardErrors, getLastActivedefault ListX<ReactiveSeq<U>> multicast(int num)
ReactiveSeqReactiveSeq.duplicate()
ReactiveSeq#triplate
ReactiveSeq.quadruplicate()
Which buffer all Stream types and produce a synchronous downstream stream.multicast in interface ReactiveSeq<U>num - Number of downstream Streams to multicast todefault ReactiveSeq<U> changes()
changes in interface ReactiveSeq<U>default Maybe<U> findOne()
ReactiveSeqfindOne in interface ReactiveSeq<U>default Either<java.lang.Throwable,U> findFirstOrError()
ReactiveSeqfindFirstOrError in interface ReactiveSeq<U>default <R> FutureStream<R> parallel(java.util.concurrent.ForkJoinPool fj, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> fn)
parallel in interface ReactiveSeq<U>default <U1,R> FutureStream<R> zipLatest(org.reactivestreams.Publisher<? extends U1> other, java.util.function.BiFunction<? super U,? super U1,? extends R> zipper)
zipLatest in interface ReactiveSeq<U>default FutureStream<U> skipUntilClosed(java.util.function.Predicate<? super U> p)
skipUntilClosed in interface ReactiveSeq<U>default ReactiveSeq<U> limitUntilClosed(java.util.function.Predicate<? super U> p)
limitUntilClosed in interface ReactiveSeq<U>default FutureStream<U> reduceAll(U identity, java.util.function.BinaryOperator<U> accumulator)
reduceAll in interface ReactiveSeq<U>default <R,A> FutureStream<R> collectStream(java.util.stream.Collector<? super U,A,R> collector)
collectStream in interface ReactiveSeq<U>default FutureStream<U> append(U value)
append in interface ReactiveSeq<U>append in interface Traversable<U>default FutureStream<org.jooq.lambda.tuple.Tuple2<U,java.lang.Long>> timestamp()
ReactiveSeq
ReactiveSeq.of(1,2,3,4,5)
.timestamp()
.forEach(System.out::println)
timestamp in interface ReactiveSeq<U>default <R> FutureStream<R> retry(java.util.function.Function<? super U,? extends R> fn, int retries, long delay, java.util.concurrent.TimeUnit timeUnit)
ReactiveSeq
given(serviceMock.applyHKT(anyInt())).willThrow(
new RuntimeException(new SocketException("First")),
new RuntimeException(new IOException("Second"))).willReturn(
"42");
String result = ReactiveSeq.of( 1, 2, 3)
.retry(serviceMock, 7, 2, TimeUnit.SECONDS)
.firstValue();
//result = 42
retry in interface ReactiveSeq<U>retry in interface Transformable<U>fn - Function to retry if failsretries - Number of retriesdelay - Delay in TimeUnitstimeUnit - TimeUnit to use for delaydefault FutureStream<ReactiveSeq<U>> combinations(int size)
ReactiveSeq
ReactiveSeq.of(1,2,3).combinations(2)
//ReactiveSeq[ReactiveSeq[1,2],ReactiveSeq[1,3],ReactiveSeq[2,3]]
combinations in interface ExtendedTraversable<U>combinations in interface ReactiveSeq<U>size - of combinationsdefault FutureStream<U> removeAllS(java.util.stream.Stream<? extends U> stream)
IterableFilterableremoveAllS in interface IterableFilterable<U>removeAllS in interface ReactiveSeq<U>stream - of elements to removedefault FutureStream<U> removeAllI(java.lang.Iterable<? extends U> it)
IterableFilterableremoveAllI in interface IterableFilterable<U>removeAllI in interface ReactiveSeq<U>it - an Iterable of elements to removedefault FutureStream<U> removeAll(U... values)
IterableFilterableremoveAll in interface IterableFilterable<U>removeAll in interface ReactiveSeq<U>values - to removedefault FutureStream<U> retainAllI(java.lang.Iterable<? extends U> it)
IterableFilterableretainAllI in interface IterableFilterable<U>retainAllI in interface ReactiveSeq<U>it - Iterable of elements to retaindefault FutureStream<U> retainAllS(java.util.stream.Stream<? extends U> stream)
IterableFilterableretainAllS in interface IterableFilterable<U>retainAllS in interface ReactiveSeq<U>stream - of elements to retaindefault FutureStream<U> retainAll(U... values)
IterableFilterableretainAll in interface IterableFilterable<U>retainAll in interface ReactiveSeq<U>values - elements to retaindefault FutureStream<U> zip(java.util.function.BinaryOperator<Zippable<U>> combiner, Zippable<U> app)
Zippablezip in interface ReactiveSeq<U>zip in interface Traversable<U>zip in interface Zippable<U>Semigroup,
Semigroups,
Monoid,
To lift any Semigroup (or monoid) up to handling Applicatives use the combineApplicatives operator in SemigroupK
{@see com.aol.cyclops2.SemigroupK#combineApplicatives(BiFunction) } or Monoids
{ {@see com.aol.cyclops2.Monoids#combineApplicatives(java.util.function.Function, com.aol.cyclops2.function.Monoid)
}
{@code
BinaryOperator> sumMaybes = SemigroupK.combineScalarFunctors(SemigroupK.intSum);
Maybe.just(1).zip(sumMaybes, Maybe.just(5))
//Maybe.just(6));
}
default <R> FutureStream<R> zipWith(java.lang.Iterable<java.util.function.Function<? super U,? extends R>> fn)
zipWith in interface ReactiveSeq<U>zipWith in interface Traversable<U>zipWith in interface Zippable<U>default <R> FutureStream<R> zipWithS(java.util.stream.Stream<java.util.function.Function<? super U,? extends R>> fn)
zipWithS in interface ReactiveSeq<U>zipWithS in interface Traversable<U>zipWithS in interface Zippable<U>default <R> FutureStream<R> zipWithP(org.reactivestreams.Publisher<java.util.function.Function<? super U,? extends R>> fn)
zipWithP in interface ReactiveSeq<U>zipWithP in interface Traversable<U>zipWithP in interface Zippable<U>default <U1> FutureStream<org.jooq.lambda.tuple.Tuple2<U,U1>> zipP(org.reactivestreams.Publisher<? extends U1> other)
zipP in interface ReactiveSeq<U>zipP in interface Traversable<U>zipP in interface Zippable<U>default <S,U1,R> FutureStream<R> zip3(java.lang.Iterable<? extends S> second, java.lang.Iterable<? extends U1> third, Fn3<? super U,? super S,? super U1,? extends R> fn3)
zip3 in interface ReactiveSeq<U>zip3 in interface Traversable<U>zip3 in interface Zippable<U>default <T2,T3,T4,R> FutureStream<R> zip4(java.lang.Iterable<? extends T2> second, java.lang.Iterable<? extends T3> third, java.lang.Iterable<? extends T4> fourth, Fn4<? super U,? super T2,? super T3,? super T4,? extends R> fn)
zip4 in interface ReactiveSeq<U>zip4 in interface Traversable<U>zip4 in interface Zippable<U>default FutureStream<U> mergeP(org.reactivestreams.Publisher<U>... publishers)
ReactiveSeqIterableFunctor#mergePublisher(Collection, QueueFactory)
method to forEachAsync asynchronously also. Max concurrency is determined by the publishers toX size, along with a default limit of 5k queued values before
backpressure is applied.mergeP in interface ReactiveSeq<U>publishers - Publishers to mergedefault FutureStream<U> mergeP(QueueFactory<U> factory, org.reactivestreams.Publisher<U>... publishers)
ReactiveSeqQueueFactories
Push based reactiveBuffer-streams signal demand via their subscription.mergeP in interface ReactiveSeq<U>default FutureStream<U> merge(Adapter<U>... adapters)
merge in interface ReactiveSeq<U>default <R1,R2,R3> FutureStream<R3> fanOutZipIn(java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R1>> path1, java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R2>> path2, java.util.function.BiFunction<? super R1,? super R2,? extends R3> zipFn)
fanOutZipIn in interface ReactiveSeq<U>default <R1,R2,R3> FutureStream<R3> parallelFanOutZipIn(java.util.concurrent.ForkJoinPool fj, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R1>> path1, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R2>> path2, java.util.function.BiFunction<? super R1,? super R2,? extends R3> zipFn)
parallelFanOutZipIn in interface ReactiveSeq<U>default <R> FutureStream<R> fanOut(java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path1, java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path2)
fanOut in interface ReactiveSeq<U>default <R> FutureStream<R> parallelFanOut(java.util.concurrent.ForkJoinPool fj, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path1, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path2)
parallelFanOut in interface ReactiveSeq<U>default <R> FutureStream<R> fanOut(java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path1, java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path2, java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path3)
fanOut in interface ReactiveSeq<U>default <R> FutureStream<R> parallelFanOut(java.util.concurrent.ForkJoinPool fj, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path1, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path2, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path3)
parallelFanOut in interface ReactiveSeq<U>default <R1,R2,R3,R4> FutureStream<R4> parallelFanOutZipIn(java.util.concurrent.ForkJoinPool fj, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R1>> path1, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R2>> path2, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R3>> path3, Fn3<? super R1,? super R2,? super R3,? extends R4> zipFn)
parallelFanOutZipIn in interface ReactiveSeq<U>default <R1,R2,R3,R4> FutureStream<R4> fanOutZipIn(java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R1>> path1, java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R2>> path2, java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R3>> path3, Fn3<? super R1,? super R2,? super R3,? extends R4> zipFn)
fanOutZipIn in interface ReactiveSeq<U>default <R> ReactiveSeq<R> fanOut(java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path1, java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path2, java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path3, java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R>> path4)
fanOut in interface ReactiveSeq<U>default <R> FutureStream<R> parallelFanOut(java.util.concurrent.ForkJoinPool fj, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path1, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path2, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path3, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> path4)
parallelFanOut in interface ReactiveSeq<U>default <R1,R2,R3,R4,R5> FutureStream<R5> fanOutZipIn(java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R1>> path1, java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R2>> path2, java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R3>> path3, java.util.function.Function<? super ReactiveSeq<U>,? extends ReactiveSeq<? extends R4>> path4, Fn4<? super R1,? super R2,? super R3,? super R4,? extends R5> zipFn)
fanOutZipIn in interface ReactiveSeq<U>default <R1,R2,R3,R4,R5> FutureStream<R5> parallelFanOutZipIn(java.util.concurrent.ForkJoinPool fj, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R1>> path1, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R2>> path2, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R3>> path3, java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R4>> path4, Fn4<? super R1,? super R2,? super R3,? super R4,? extends R5> zipFn)
parallelFanOutZipIn in interface ReactiveSeq<U>static LazyReact builder(int maxActiveTasks, java.util.concurrent.Executor exec)
default <R> Future<R> foldFuture(java.util.function.Function<? super FoldableTraversable<U>,? extends R> fn)
default ReactiveStreamsTerminalFutureOperations<U> futureOperations()
default <A,R> FutureStream<R> collectSeq(java.util.stream.Collector<? super U,A,R> c)
default FutureStream<U> fold(Monoid<U> monoid)
default <U1> FutureStream<org.jooq.lambda.tuple.Tuple2<U,U1>> crossApply(java.util.function.Function<? super U,? extends java.lang.Iterable<? extends U1>> function)
default <U1> FutureStream<org.jooq.lambda.tuple.Tuple2<U,U1>> outerApply(java.util.function.Function<? super U,? extends java.lang.Iterable<? extends U1>> function)
default FutureStream<U> appendS(java.util.stream.Stream<? extends U> other)
ReactiveSeq
List<String> result = ReactiveSeq.of(1, 2, 3).appendStream(ReactiveSeq.of(100, 200, 300)).map(it -> it + "!!").collect(CyclopsCollectors.toList());
assertThat(result, equalTo(Arrays.asList("1!!", "2!!", "3!!", "100!!", "200!!", "300!!"))); *
appendS in interface ReactiveSeq<U>default FutureStream<U> append(java.lang.Iterable<? extends U> other)
append in interface ReactiveSeq<U>default FutureStream<U> prependS(java.util.stream.Stream<? extends U> other)
ReactiveSeq
List<String> result = ReactiveSeq.of(1, 2, 3)
.prependS(of(100, 200, 300))
.map(it -> it + "!!")
.collect(CyclopsCollectors.toList());
assertThat(result, equalTo(Arrays.asList("100!!", "200!!", "300!!", "1!!", "2!!", "3!!")));
prependS in interface ReactiveSeq<U>prependS in interface Traversable<U>other - to Prependdefault FutureStream<U> prepend(java.lang.Iterable<? extends U> other)
prepend in interface ReactiveSeq<U>default FutureStream<U> cycle(long times)
assertThat(FutureStream.of(1,2,2).cycle(3)
.collect(CyclopsCollectors.toList()),
equalTo(Arrays.asList(1,2,2,1,2,2,1,2,2)));
cycle in interface ReactiveSeq<U>cycle in interface Traversable<U>times - Times values should be repeated within a Streamdefault FutureStream<U> skipWhileClosed(java.util.function.Predicate<? super U> predicate)
skipWhileClosed in interface ReactiveSeq<U>default FutureStream<U> limitWhileClosed(java.util.function.Predicate<? super U> predicate)
limitWhileClosed in interface ReactiveSeq<U>default <U1> FutureStream<U> sorted(java.util.function.Function<? super U,? extends U1> function, java.util.Comparator<? super U1> comparator)
sorted in interface ReactiveSeq<U>default <R> FutureStream<R> coflatMap(java.util.function.Function<? super ReactiveSeq<U>,? extends R> fn)
FutureStream.of(1,2,3)
.map(i->i*2)
.coflatMap(s -> s.reduce(0,(a,b)->a+b))
//FutureStream[12]
coflatMap in interface ReactiveSeq<U>fn - default FutureStream<U> filterNot(java.util.function.Predicate<? super U> fn)
Filters
of(1,2,3).filter(i->i>2);
//[1,2]
default FutureStream<U> notNull()
Filters
of(1,2,null,4).nonNull();
//[1,2,4]
default <R> FutureStream<R> trampoline(java.util.function.Function<? super U,? extends Trampoline<? extends R>> mapper)
ReactiveSeq
ReactiveSeq.of(10,20,30,40)
.trampoline(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
ReactiveSeq.of(10_000,200_000,3_000_000,40_000_000)
.trampoline(i-> fibonacci(i))
.forEach(System.out::println);
completes successfully
trampoline in interface ReactiveSeq<U>trampoline in interface Transformable<U>mapper - TCO Transformation functiondefault <R> R foldRight(R identity,
java.util.function.BiFunction<? super U,? super R,? extends R> accumulator)
Foldsdefault <T2,R> FutureStream<R> zipP(org.reactivestreams.Publisher<? extends T2> publisher, java.util.function.BiFunction<? super U,? super T2,? extends R> fn)
ZippablezipP in interface ReactiveSeq<U>zipP in interface Traversable<U>zipP in interface Zippable<U>publisher - to combine withfn - Zip / combining functiondefault FutureStream<org.jooq.lambda.tuple.Tuple2<U,java.lang.Long>> elapsed()
FutureStream.of(1,2,3,4,5)
.elapsed()
.forEach(System.out::println);
elapsed in interface ReactiveSeq<U>default FutureStream<U> combine(java.util.function.BiPredicate<? super U,? super U> predicate, java.util.function.BinaryOperator<U> op)
Traversable
ReactiveSeq.of(1,1,2,3)
.combine((a, b)->a.equals(b),SemigroupK.intSum)
.listX()
//ListX(3,4)
combine in interface ReactiveSeq<U>combine in interface Traversable<U>predicate - Test to see if two neighbours should be joined. The first parameter to the bi-predicate is the currently
accumulated result and the second is the next elementop - BinaryOperator to combine neighboursdefault FutureStream<U> combine(Monoid<U> op, java.util.function.BiPredicate<? super U,? super U> predicate)
Traversable
ReactiveSeq.of(1,1,2,3)
.combine(Monoids.intMult,(a, b)->a.equals(b))
.listX()
//ListX(1)
Simalar to @see Traversable.combine(BiPredicate, BinaryOperator) but differs in that the first comparison is always to the Monoid zero
This allows us to terminate with just a single valuecombine in interface ReactiveSeq<U>combine in interface Traversable<U>op - Monoid to combine neighbourspredicate - Test to see if two neighbours should be joined. The first parameter to the bi-predicate is the currently
accumulated result and the second is the next elementdefault FutureStream<U> onEmptySwitch(java.util.function.Supplier<? extends java.util.stream.Stream<U>> switchTo)
assertThat(FutureStream.of(4,5,6)
.onEmptySwitch(()->ReactiveSeq.of(1,2,3))
.toList(),
equalTo(Arrays.asList(4,5,6)));
onEmptySwitch in interface OnEmptySwitch<U,java.util.stream.Stream<U>>onEmptySwitch in interface ReactiveSeq<U>switchTo - Supplier that will generate the alternative Streamdefault <R1,R2,R3,R> FutureStream<R> forEach4(java.util.function.Function<? super U,? extends java.util.stream.BaseStream<R1,?>> stream1, java.util.function.BiFunction<? super U,? super R1,? extends java.util.stream.BaseStream<R2,?>> stream2, Fn3<? super U,? super R1,? super R2,? extends java.util.stream.BaseStream<R3,?>> stream3, Fn4<? super U,? super R1,? super R2,? super R3,? extends R> yieldingFunction)
ReactiveSeq
//ReactiveSeq [1,2]
reactiveSeq.forEach4(a->ListX.range(10,13),
(a,b)->ListX.of(""+(a+b),"hello world"),
(a,b,c)->ListX.of(a,b,c)),
(a,b,c,d)->c+":"a+":"+b);
forEach4 in interface ReactiveSeq<U>stream1 - Nested Stream to iterate overstream2 - Nested Stream to iterate overstream3 - Nested Stream to iterate overyieldingFunction - Function with pointers to the current element from both
Streams that generates the new elementsdefault <R1,R2,R3,R> FutureStream<R> forEach4(java.util.function.Function<? super U,? extends java.util.stream.BaseStream<R1,?>> stream1, java.util.function.BiFunction<? super U,? super R1,? extends java.util.stream.BaseStream<R2,?>> stream2, Fn3<? super U,? super R1,? super R2,? extends java.util.stream.BaseStream<R3,?>> stream3, Fn4<? super U,? super R1,? super R2,? super R3,java.lang.Boolean> filterFunction, Fn4<? super U,? super R1,? super R2,? super R3,? extends R> yieldingFunction)
ReactiveSeq
//ReactiveSeq [1,2,3]
seq.forEach4(a->ReactiveSeq.range(10,13),
(a,b)->Stream.of(""+(a+b),"hello world"),
(a,b,c)->Stream.of(a,b,c),
(a,b,c,d)-> c!=3,
(a,b,c)->c+":"a+":"+b);
forEach4 in interface ReactiveSeq<U>stream1 - Nested Stream to iterate overstream2 - Nested Stream to iterate overstream3 - Nested Stream to iterate overfilterFunction - Filter to applyHKT over elements before passing non-filtered
values to the yielding functionyieldingFunction - Function with pointers to the current element from both
Streams that generates the new elementsdefault <R1,R2,R> FutureStream<R> forEach3(java.util.function.Function<? super U,? extends java.util.stream.BaseStream<R1,?>> stream1, java.util.function.BiFunction<? super U,? super R1,? extends java.util.stream.BaseStream<R2,?>> stream2, Fn3<? super U,? super R1,? super R2,? extends R> yieldingFunction)
ReactiveSeq
ReactiveSeq.of(1,2)
.forEach3(a->IntStream.range(10,13),
(a,b)->Stream.of(""+(a+b),"hello world"),
(a,b,c)->c+":"a+":"+b);
//ReactiveSeq[11:1:2,hello world:1:2,14:1:4,hello world:1:4,12:1:2,hello world:1:2,15:1:5,hello world:1:5]
forEach3 in interface ReactiveSeq<U>stream1 - Nested Stream to iterate overstream2 - Nested Stream to iterate overyieldingFunction - Function with pointers to the current element from both
Streams that generates the new elementsdefault <R1,R2,R> FutureStream<R> forEach3(java.util.function.Function<? super U,? extends java.util.stream.BaseStream<R1,?>> stream1, java.util.function.BiFunction<? super U,? super R1,? extends java.util.stream.BaseStream<R2,?>> stream2, Fn3<? super U,? super R1,? super R2,java.lang.Boolean> filterFunction, Fn3<? super U,? super R1,? super R2,? extends R> yieldingFunction)
ReactiveSeq
ReactiveSeq.of(1,2,3)
.forEach3(a->IntStream.range(10,13),
(a,b)->Stream.of(""+(a+b),"hello world"),
(a,b,c)-> c!=3,
(a,b,c)->c+":"a+":"+b);
//ReactiveSeq[11:1:2,hello world:1:2,14:1:4,hello world:1:4,12:1:2,hello world:1:2,15:1:5,hello world:1:5]
forEach3 in interface ReactiveSeq<U>stream1 - Nested Stream to iterate overstream2 - Nested Stream to iterate overfilterFunction - Filter to applyHKT over elements before passing non-filtered
values to the yielding functionyieldingFunction - Function with pointers to the current element from both
Streams that generates the new elementsdefault <R1,R> FutureStream<R> forEach2(java.util.function.Function<? super U,? extends java.util.stream.BaseStream<R1,?>> stream1, java.util.function.BiFunction<? super U,? super R1,? extends R> yieldingFunction)
ReactiveSeq
ReactiveSeq.of(1,2,3)
.forEach2(a->IntStream.range(10,13),
(a,b)->a+b);
//ReactiveSeq[11,14,12,15,13,16]
forEach2 in interface ReactiveSeq<U>stream1 - Nested Stream to iterate overyieldingFunction - Function with pointers to the current element from both
Streams that generates the new elementsdefault <R1,R> FutureStream<R> forEach2(java.util.function.Function<? super U,? extends java.util.stream.BaseStream<R1,?>> stream1, java.util.function.BiFunction<? super U,? super R1,java.lang.Boolean> filterFunction, java.util.function.BiFunction<? super U,? super R1,? extends R> yieldingFunction)
ReactiveSeq
ReactiveSeq.of(1,2,3)
.forEach2(a->IntStream.range(10,13),
(a,b)-> a<3 && b>10,
(a,b)->a+b);
//ReactiveSeq[14,15]
forEach2 in interface ReactiveSeq<U>stream1 - Nested Stream to iterate overfilterFunction - Filter to applyHKT over elements before passing non-filtered
values to the yielding functionyieldingFunction - Function with pointers to the current element from both
Streams that generates the new elementsdefault FutureStream<U> remove(U t)
FutureStream.of(1,2,3,4,5,1,2,3).remove(1)
//FutureStream[2,3,4,5,2,3]
remove in interface ReactiveSeq<U>t - element to removedefault FutureStream<U> subStream(int start, int end)
FutureStream.of(1,2,3,4,5,6).subStream(1,3);
//FutureStream[2,3]
subStream in interface ReactiveSeq<U>start - index inclusiveend - index exclusivedefault FutureStream<ReactiveSeq<U>> permutations()
permutations in interface ExtendedTraversable<U>permutations in interface ReactiveSeq<U>default FutureStream<ReactiveSeq<U>> combinations()
FutureStream.of(1,2,3).combinations()
//FutureStream[Stream[],Stream[1],Stream[2],Stream[3].Stream[1,2],Stream[1,3],Stream[2,3]
,Stream[1,2,3]]
combinations in interface ExtendedTraversable<U>combinations in interface ReactiveSeq<U>default OperationsOnFutures<U> actOnFutures()
new LazyReact().react(()->slow(),()->fast())
.zipWithIndex();
//[fast,0l],[slow,1l]
The first result will be fast and will have index 0 (the result index)
Operating on futures
new LazyReact().react(()->slow(),()->fast())
.actOnFutures()
.zipWithIndex();
//[fast,1l],[slow,0l]
The first result will still be fast, but the index will now be the skip index 1default CloseableIterator<U> iterator()
iterator in interface BaseSimpleReactStream<U>iterator in interface java.util.stream.BaseStream<U,java.util.stream.Stream<U>>iterator in interface java.lang.Iterable<U>iterator in interface LazyStream<U>Continueable getSubscription()
getSubscription in interface BaseSimpleReactStream<U>getSubscription in interface LazySimpleReactStream<U><R> FutureStream<R> withLastActive(LazyStreamWrapper<R> streamWrapper)
withLastActive in interface LazySimpleReactStream<U>LazyReact getSimpleReact()
getSimpleReact in interface BaseSimpleReactStream<U>getSimpleReact in interface ConfigurableStream<U,FastFuture<U>>getSimpleReact in interface LazySimpleReactStream<U>default void subscribe(org.reactivestreams.Subscriber<? super U> s)
subscribe in interface FutureStreamSynchronousPublisher<U>subscribe in interface org.reactivestreams.Publisher<U>subscribe in interface Traversable<U>default java.util.Iterator<java.util.Collection<U>> chunkLastReadIterator()
default FutureStream<java.util.Collection<U>> chunkSinceLastRead()
default long count()
ReactiveSeq
return mapToLong(e -> 1L).sum();
This is a terminal operation.
count in interface CyclopsCollectable<U>count in interface ReactiveSeq<U>count in interface java.util.stream.Stream<U>FutureStream<U> withTaskExecutor(java.util.concurrent.Executor e)
withTaskExecutor in interface ConfigurableStream<U,FastFuture<U>>withTaskExecutor in interface LazySimpleReactStream<U>FutureStream<U> withLazyCollector(java.util.function.Supplier<LazyResultConsumer<U>> lazy)
FutureStream<U> withQueueFactory(QueueFactory<U> queue)
withQueueFactory in interface BaseSimpleReactStream<U>withQueueFactory in interface ConfigurableStream<U,FastFuture<U>>withQueueFactory in interface LazySimpleReactStream<U>FutureStream<U> withSubscription(Continueable sub)
withSubscription in interface BaseSimpleReactStream<U>withSubscription in interface ConfigurableStream<U,FastFuture<U>>withSubscription in interface LazySimpleReactStream<U>FutureStream<U> withAsync(boolean async)
withAsync in interface ConfigurableStream<U,FastFuture<U>>withAsync in interface LazySimpleReactStream<U>default void forEach(java.util.function.Consumer<? super U> c)
ReactiveSeqReactiveSeq.forEachAsync(Consumer) and forEach overloads
such as FoldableTraversable.forEach(Consumer, Consumer) and FoldableTraversable.forEach(Consumer, Consumer,Runnable)
This method overrides the JDK Stream.forEach(Consumer) and maintains it's blocking
semantics. Other forEach overloads in ReactiveSeq are non-blocking for asynchronously executing Streams.
This is a terminal operation.
The behavior of this operation is explicitly nondeterministic. For parallel reactiveStream pipelines, this operation does not guarantee to respect the encounter order of the reactiveStream, as doing so would sacrifice the benefit of parallelism. For any given element, the action may be performed at whatever time and in whatever thread the library chooses. If the action accesses shared state, it is responsible for providing the required synchronization.
forEach in interface java.lang.Iterable<U>forEach in interface LazyStream<U>forEach in interface ReactiveSeq<U>forEach in interface java.util.stream.Stream<U>c - a
non-interfering action to perform on the elementsdefault Queue<U> toQueue()
LazyToQueuetoQueue in interface BaseSimpleReactStream<U>toQueue in interface FutureStreamSynchronousPublisher<U>toQueue in interface LazyToQueue<U>toQueue in interface ToQueue<U>default <R> FutureStream<R> parallel(java.util.function.Function<? super java.util.stream.Stream<U>,? extends java.util.stream.Stream<? extends R>> fn)
parallel in interface ReactiveSeq<U>default <R> FutureStream<R> jool(java.util.function.Function<? super org.jooq.lambda.Seq<U>,? extends org.jooq.lambda.Seq<R>> mapper)
jool in interface ReactiveSeq<U>default <T> T reduce(T identity,
java.util.function.BiFunction<T,? super U,T> accumulator)
FoldsStream.reduce(Object, BinaryOperator)reduce in interface Folds<U>reduce in interface LazyStream<U>reduce in interface ReactiveSeq<U>identity - Identity value for the combiner function (leaves the input unchanged)accumulator - Combiner functiondefault U reduce(U identity, java.util.function.BinaryOperator<U> accumulator)
FoldsStream.reduce(Object, BinaryOperator)reduce in interface Folds<U>reduce in interface LazyStream<U>reduce in interface ReactiveSeq<U>reduce in interface java.util.stream.Stream<U>accumulator - Combiner functiondefault <T> T reduce(T identity,
java.util.function.BiFunction<T,? super U,T> accumulator,
java.util.function.BinaryOperator<T> combiner)
FoldsStream.reduce(Object, BiFunction, BinaryOperator)reduce in interface Folds<U>reduce in interface LazyStream<U>reduce in interface ReactiveSeq<U>reduce in interface java.util.stream.Stream<U>default java.util.Optional<U> reduce(java.util.function.BinaryOperator<U> accumulator)
FoldsStream.reduce(BinaryOperator)
ReactiveSeq.of(1,2,3,4,5).map(it -> it*100).reduce(
(acc,next) -> acc+next)
//Optional[1500]
reduce in interface Folds<U>reduce in interface LazyStream<U>reduce in interface ReactiveSeq<U>reduce in interface java.util.stream.Stream<U>accumulator - Combiner functiondefault <R> R collect(java.util.function.Supplier<R> supplier,
java.util.function.BiConsumer<R,? super U> accumulator,
java.util.function.BiConsumer<R,R> combiner)
ReactiveSeqArrayList, and elements are incorporated by updating
the state of the result rather than by replacing the result. This
produces a result equivalent to:
R result = supplier.get();
for (T element : this reactiveStream)
accumulator.accept(result, element);
return result;
Like ReactiveSeq.reduce(Object, BinaryOperator), collect operations
can be parallelized without requiring additional synchronization.
This is a terminal operation.
collect in interface LazyStream<U>collect in interface ReactiveSeq<U>collect in interface java.util.stream.Stream<U>supplier - a function that creates a new result container. For a
parallel execution, this function may be called
multiple times and must return a fresh value each time.accumulator - an associative,
non-interfering,
stateless
function for incorporating an additional element into a resultcombiner - an associative,
non-interfering,
stateless
function for combining two values, which must be
compatible with the accumulator functiondefault FutureStream<U> sync()
sync in interface BaseSimpleReactStream<U>default FutureStream<U> async()
async in interface BaseSimpleReactStream<U>default FutureStream<U> unboundedWaitFree()
default FutureStream<U> boundedWaitFree(int size)
FutureStream.of(col)
.boundedWaitFree(110)
.flatMap(Collection::reactiveStream)
.toList();
size - Buffer sizeFutureStream<U> maxActive(int concurrentTasks)
List<String> data = new LazyReact().react(urlFile)
.maxActive(100)
.flatMap(this::loadUrls)
.map(this::callUrls)
.block();
concurrentTasks - Maximum number of active task chainsdefault <R> FutureStream<R> thenSync(java.util.function.Function<? super U,? extends R> fn)
thenSync in interface BaseSimpleReactStream<U>thenSync in interface LazySimpleReactStream<U>thenSync in interface LazyToQueue<U>default FutureStream<U> peekSync(java.util.function.Consumer<? super U> consumer)
LazySimpleReactStreampeekSync in interface BaseSimpleReactStream<U>peekSync in interface LazySimpleReactStream<U>peekSync in interface LazyToQueue<U>consumer - Peek consumerdefault void closeAll()
default HotStream<U> hotStream()
HotStream<Integer> ints = new LazyReact().range(0,Integer.MAX_VALUE)
.hotStream()
ints.connect().forEach(System.out::println);
//print out all the ints
//multiple consumers are possible, so other Streams can connect on different Threads
default java.util.Optional<U> findFirst()
default SimpleReactStream<U> convertToSimpleReact()
LazyReact.parallelCommonBuilder()
.react(()->slow(),()->1,()->2)
.peek(System.out::println)
.convertToSimpleReact()
.allOf(list->list)
.block()
default <R> FutureStream<R> map(java.util.function.Function<? super U,? extends R> mapper)
Transformable
of(1,2,3).map(i->i*2)
//[2,4,6]
map in interface FilterableTransformable<U>map in interface ReactiveSeq<U>map in interface java.util.stream.Stream<U>map in interface Transformable<U>mapper - Transformation functiondefault <K> java.util.Map<K,FutureStream<U>> shard(java.util.Map<K,Queue<U>> shards, java.util.function.Function<? super U,? extends K> sharder)
FutureStream.of(10,20,25,30,41,43).shard(ImmutableMap.of("even",new
Queue(),"odd",new Queue(),element-> element%2==0? "even" : "odd");
results in 2 Streams "even": 10,20,30 "odd" : 25,41,43shards - Map of Queue's keyed by shard identifiersharder - Function to split split incoming elements into shardsdefault FutureStream<U> debounce(long time, java.util.concurrent.TimeUnit unit)
FutureStream.of(1,2,3,4,5,6)
.debounce(1000,TimeUnit.SECONDS).toList();
// [1]
debounce in interface ReactiveSeq<U>time - Time from which to accept only one elementunit - Time unit for specified timedefault FutureStream<U> control(java.util.function.Function<java.util.function.Supplier<U>,java.util.function.Supplier<U>> fn)
fn - Function takes a supplier, which can be used repeatedly to get
the next value from the Stream. If there are no more values, a
ClosedQueueException will be thrown. This function should
return a Supplier which returns the desired result for the
next element (or just the next element).default <C extends java.util.Collection<U>> FutureStream<C> group(java.util.function.Function<java.util.function.Supplier<U>,java.util.function.Supplier<C>> fn)
fn - Function takes a supplier, which can be used repeatedly to get
the next value from the Stream. If there are no more values, a
ClosedQueueException will be thrown. This function should
return a Supplier which creates a toX of the batched
valuesdefault FutureStream<ListX<U>> groupedBySizeAndTime(int size, long time, java.util.concurrent.TimeUnit unit)
ReactiveSeq
ReactiveSeq.of(1,2,3,4,5,6)
.groupedBySizeAndTime(3,10,TimeUnit.SECONDS)
.toList();
//[[1,2,3],[4,5,6]]
groupedBySizeAndTime in interface ReactiveSeq<U>size - Max size of a batchtime - (Max) time period to build a singleUnsafe batch inunit - time unit for batchdefault <C extends java.util.Collection<? super U>,R> FutureStream<R> groupedBySizeAndTime(int size, long time, java.util.concurrent.TimeUnit unit, java.util.function.Supplier<C> factory, java.util.function.Function<? super C,? extends R> finalizer)
groupedBySizeAndTime in interface ReactiveSeq<U>default <C extends java.util.Collection<? super U>> FutureStream<C> grouped(int size, java.util.function.Supplier<C> supplier)
FutureStream.of(1,1,1,1,1,1)
.batchBySize(3,()->new TreeSet<>())
.toList()
//[[1],[1]]
grouped in interface ReactiveSeq<U>grouped in interface Traversable<U>size - Size of batchsupplier - Create the batch holding toXdefault FutureStream<U> jitter(long jitterInNanos)
FutureStream.parallelCommonBuilder()
.of(IntStream.range(0, 100))
.map(it -> it*100)
.jitter(10l)
.peek(System.out::println)
.block();
jitter in interface ReactiveSeq<U>jitterInNanos - Max number of nanos for jitter (random number less than this
will be selected)/default FutureStream<U> fixedDelay(long time, java.util.concurrent.TimeUnit unit)
FutureStream.of(1,2,3,4)
.fixedDelay(1,TimeUnit.hours);
Will emit 1 on skip, applyHKT 2 after an hour, 3 after 2 hours and so on.
However all 4 numbers will be populated in the Stream immediately.
FutureStream.of(1,2,3,4)
.withQueueFactories(QueueFactories.boundedQueue(1))
.fixedDelay(1,TimeUnit.hours);
Will populate each number in the Stream an hour apart.fixedDelay in interface ReactiveSeq<U>fixedDelay in interface Sequential<U>time - amount of time between emissionsunit - TimeUnit for emissionsdefault FutureStream<U> onePer(long time, java.util.concurrent.TimeUnit unit)
FutureStream.of(1,2,3,4,5,6)
.onePer(1000,TimeUnit.NANOSECONDS)
.toList();
onePer in interface ReactiveSeq<U>onePer in interface Sequential<U>time - Frequency period of element emissionunit - Time unit for frequency perioddefault FutureStream<U> xPer(int x, long time, java.util.concurrent.TimeUnit unit)
FutureStream.of(1,2,3,4,5,6)
.xPer(6,100000000,TimeUnit.NANOSECONDS)
.toList();
xPer in interface ReactiveSeq<U>xPer in interface Sequential<U>x - Number of allowable emissions per time periodtime - Frequency time periodunit - Frequency time unitdefault FutureStream<ListX<U>> groupedByTime(long time, java.util.concurrent.TimeUnit unit)
FutureStream.react(()->load1(),()->load2(),()->load3(),()->load4(),()->load5(),()->load6())
.batchByTime(15000,TimeUnit.MICROSECONDS);
groupedByTime in interface ReactiveSeq<U>time - Time period during which all elements should be collectedunit - Time unit during which all elements should be collecteddefault <C extends java.util.Collection<? super U>> FutureStream<C> groupedByTime(long time, java.util.concurrent.TimeUnit unit, java.util.function.Supplier<C> factory)
List <TreeSet<Integer>> set = FutureStream.ofThread(1,1,1,1,1,1)
.batchByTime(1500,TimeUnit.MICROSECONDS,()-> new TreeSet<>())
.block();
assertThat(set.get(0).size(),is(1));
groupedByTime in interface ReactiveSeq<U>time - Time period during which all elements should be collectedunit - Time unit during which all elements should be collectedfactory - Instantiates the collections used in the batchingdefault <C extends java.util.Collection<? super U>,R> FutureStream<R> groupedByTime(long time, java.util.concurrent.TimeUnit unit, java.util.function.Supplier<C> factory, java.util.function.Function<? super C,? extends R> finalizer)
groupedByTime in interface ReactiveSeq<U>default <R> FutureStream<R> then(java.util.function.Function<? super U,? extends R> fn, java.util.concurrent.Executor service)
then in interface BaseSimpleReactStream<U>then in interface LazySimpleReactStream<U>then in interface LazyToQueue<U>default <R> FutureStream<R> then(java.util.function.Function<? super U,? extends R> fn)
LazySimpleReactStream
new SimpleReact().<Integer, Integer> react(() -> 1, () -> 2, () -> 3)
.transform((it) -> it * 100)
.transform((it) -> "*" + it)
React transform allows event reactors to be chained. Unlike React with, which
returns a toX of Future references, React transform is a fluent
interface that returns the React builder - allowing further reactors to
be added to the chain.
React transform does not block.
React with can be called after React transform which gives access to the full
CompleteableFuture api. CompleteableFutures can be passed back into
SimpleReact via SimpleReact.react(streamOfCompleteableFutures);
See this blog post for examples of what can be achieved via
CompleteableFuture :- http://www.nurkiewicz.com/2013/12/promises-and-completablefuture.htm
l then in interface BaseSimpleReactStream<U>then in interface LazySimpleReactStream<U>fn - Function to be applied to the results of the currently active
event tasksdefault java.util.List<FutureStream<U>> copy(int times)
FutureStream.of(1,2,3,4,5,6)
.map(i->i+2)
.copy(5)
.forEach(s -> System.out.println(s.toList()));
times - to copy this Streamdefault <R> FutureStream<R> switchOnNextValue(java.util.stream.Stream<FutureStream> streams)
FutureStream<Integer> fast = ... // [1,2,3,4,5,6,7..]
FutureStream<Integer> slow = ... // [100,200,300,400,500,600..]
FutureStream<Integer> merged = fast.switchOnNextValue(Stream.of(slow)); //[1,2,3,4,5,6,7,8,100,9,10,11,12,13,14,15,16,200..]
streams - default <R> FutureStream<R> mergeLatest(FutureStream<?>... streams)
FutureStream<Integer> fast = ... // [1,2,3,4,5,6,7..]
FutureStream<Integer> slow = ... // [100,200,300,400,500,600..]
FutureStream<Integer> merged = fast.mergeLatest(slow); //[1,2,3,4,5,6,7,8,100,9,10,11,12,13,14,15,16,200..]
streams - default FutureStream<U> onFail(java.util.function.Function<? super SimpleReactFailedStageException,? extends U> fn)
LazySimpleReactStream
List<String> strings = new SimpleReact().<Integer, Integer> react(() -> 100, () -> 2, () -> 3)
.transform(it -> {
if (it == 100)
throw new RuntimeException("boo!");
return it;
})
.onFail(e -> 1)
.transform(it -> "*" + it)
.block();
In this example onFail recovers from the RuntimeException thrown when the
input to the first 'transform' stage is 100.onFail in interface BaseSimpleReactStream<U>onFail in interface LazySimpleReactStream<U>fn - Recovery function, the exception is input, and the recovery
value is emitteddefault FutureStream<U> onFail(java.lang.Class<? extends java.lang.Throwable> exceptionClass, java.util.function.Function<? super SimpleReactFailedStageException,? extends U> fn)
LazySimpleReactStream
onFail(IOException.class, recoveryFunction1)
.onFail(Throwable.class,recovertyFunction2)
For an IOException recoveryFunction1 will be executed
but with the definitions reveresed
onFail(Throwable.class,recovertyFunction2)
.onFail(IOException.class, recoveryFunction1)
recoveryFunction1 will not be calledonFail in interface BaseSimpleReactStream<U>onFail in interface LazySimpleReactStream<U>exceptionClass - Class of exceptions to recover fromfn - Recovery functiondefault FutureStream<U> capture(java.util.function.Consumer<java.lang.Throwable> errorHandler)
LazySimpleReactStream
List<String> strings = new SimpleReact().<Integer, Integer> react(() -> 1, () -> 2, () -> 3)
.transform(it -> it * 100)
.transform(it -> {
if (it == 100)
throw new RuntimeException("boo!");
return it;
})
.onFail(e -> 1)
.transform(it -> "*" + it)
.transform(it -> {
if ("*200".equals(it))
throw new RuntimeException("boo!");
return it;
})
.capture(e -> logger.error(e.getMessage(),e))
.block();
In this case, strings will only contain the two successful results (for
()->1 and ()->3), an exception for the chain starting from Supplier
()->2 will be logged by capture. Capture will not capture the
exception thrown when an Integer value of 100 is found, but will catch
the exception when the String value "*200" is passed along the chain.capture in interface BaseSimpleReactStream<U>capture in interface LazySimpleReactStream<U>errorHandler - A consumer that recieves and deals with an unrecoverable error
in the dataflowdefault FutureStream<U> peek(java.util.function.Consumer<? super U> consumer)
LazySimpleReactStreampeek in interface BaseSimpleReactStream<U>peek in interface LazySimpleReactStream<U>peek in interface ReactiveSeq<U>peek in interface java.util.stream.Stream<U>peek in interface Transformable<U>consumer - That will recieve current resultsdefault FutureStream<U> filter(java.util.function.Predicate<? super U> p)
LazySimpleReactStreamfilter in interface BaseSimpleReactStream<U>filter in interface FilterableTransformable<U>filter in interface Filters<U>filter in interface LazySimpleReactStream<U>filter in interface ReactiveSeq<U>filter in interface java.util.stream.Stream<U>p - Predicate that will be used to filter elements from the
dataflowdefault <R> FutureStream<R> flatMap(java.util.function.Function<? super U,? extends java.util.stream.Stream<? extends R>> flatFn)
LazySimpleReactStreamflatMap in interface BaseSimpleReactStream<U>flatMap in interface LazySimpleReactStream<U>flatMap in interface ReactiveSeq<U>flatMap in interface java.util.stream.Stream<U>flatFn - Function that coverts a value (e.g. a Collection) into a Streamdefault <R> FutureStream<R> flatMapAnyM(java.util.function.Function<? super U,AnyM<Witness.stream,? extends R>> flatFn)
ReactiveSeq
assertThat(ReactiveSeq.of(1,2,3)).flatMapAnyM(i-> fromEither5(CompletableFuture.completedFuture(i+2))).toList(),equalTo(Arrays.asList(3,4,5)));
flatMapAnyM in interface ReactiveSeq<U>flatFn - to be applieddefault <R> FutureStream<R> flatMapCompletableFuture(java.util.function.Function<? super U,java.util.concurrent.CompletableFuture<? extends R>> flatFn)
assertThat( new LazyReact()
.of(1,2,3)
.flatMapCompletableFuture(i->CompletableFuture.completedFuture(i))
.block(),equalTo(Arrays.asList(1,2,3)));
In this example the result of the flatMapCompletableFuture is 'flattened' to the raw integer valuesflatFn - flatMap functiondefault <R> FutureStream<R> retry(java.util.function.Function<? super U,? extends R> fn)
ReactiveSeq
given(serviceMock.applyHKT(anyInt())).willThrow(
new RuntimeException(new SocketException("First")),
new RuntimeException(new IOException("Second"))).willReturn(
"42");
String result = ReactiveSeq.of( 1, 2, 3)
.retry(serviceMock)
.firstValue();
//result = 42
retry in interface ReactiveSeq<U>retry in interface Transformable<U>fn - Function to retry if failsdefault <R> FutureStream<R> fromStream(java.util.stream.Stream<R> stream)
fromStream in interface BaseSimpleReactStream<U>default <R> FutureStream<R> fromStreamOfFutures(java.util.stream.Stream<FastFuture<R>> stream)
default FutureStream<U> concat(java.util.stream.Stream<? extends U> other)
ReactiveSeq.concat(Stream[])default FutureStream<U> concat(U other)
// (1, 2, 3, 4)
FutureStream.of(1, 2, 3).concat(4)
ReactiveSeq.concat(Stream[])default FutureStream<U> concat(U... other)
// (1, 2, 3, 4, 5, 6)
FutureStream.of(1, 2, 3).concat(4, 5, 6)
ReactiveSeq.concat(Stream[])default <U> FutureStream<U> cast(java.lang.Class<? extends U> type)
ReactiveSeqClassCastException.
// ClassCastException ReactiveSeq.of(1, "a", 2, "b", 3).cast(Integer.class)cast in interface BaseSimpleReactStream<U>cast in interface ReactiveSeq<U>cast in interface Transformable<U>default <U> FutureStream<U> ofType(java.lang.Class<? extends U> type)
ofType in interface BaseSimpleReactStream<U>ofType in interface Filters<U>ofType in interface ReactiveSeq<U>default FutureStream<U> intersperse(U value)
// (1, 0, 2, 0, 3, 0, 4)
FutureStream.of(1, 2, 3, 4).intersperse(0)
intersperse in interface BaseSimpleReactStream<U>intersperse in interface ReactiveSeq<U>intersperse in interface Traversable<U>#intersperse(Stream, Object)default FutureStream<U> limit(long maxSize)
ReactiveSeq
assertThat(ReactiveSeq.of(4,3,6,7).limit(2).toList(),equalTo(Arrays.asList(4,3));
limit in interface ReactiveSeq<U>limit in interface java.util.stream.Stream<U>limit in interface Traversable<U>maxSize - Limit element size to numdefault FutureStream<U> drop(long drop)
Traversable
assertThat(ReactiveSeq.of(4,3,6,7).drop(2).toList(),equalTo(Arrays.asList(6,7)));
drop in interface ReactiveSeq<U>drop in interface Traversable<U>drop - Number of elemenets to dropdefault FutureStream<U> take(long take)
Traversable
assertThat(ReactiveSeq.of(4,3,6,7).take(2).toList(),equalTo(Arrays.asList(4,3));
take in interface ReactiveSeq<U>take in interface Traversable<U>take - Limit element size to numdefault FutureStream<U> takeWhile(java.util.function.Predicate<? super U> p)
Traversable
ListX.of(1,2,3).takeWhile(i<3);
//[1,2]
takeWhile in interface ReactiveSeq<U>takeWhile in interface Traversable<U>p - Predicate to determine when values should be takendefault FutureStream<U> dropWhile(java.util.function.Predicate<? super U> p)
Traversable
ListX.of(1,2,3).dropWhile(i<3);
//[3]
dropWhile in interface ReactiveSeq<U>dropWhile in interface Traversable<U>p - Predicate to determine when values should be droppeddefault FutureStream<U> takeUntil(java.util.function.Predicate<? super U> p)
Traversable
ListX.of(1,2,3).takeUntil(i<2);
//[1,2]
takeUntil in interface ReactiveSeq<U>takeUntil in interface Traversable<U>p - Predicate to determine when values should be taken untildefault FutureStream<U> dropUntil(java.util.function.Predicate<? super U> p)
Traversable
ListX.of(1,2,3).dropUntil(i>2);
//[3]
dropUntil in interface ReactiveSeq<U>dropUntil in interface Traversable<U>p - Predicate to determine when values should be droppeddefault FutureStream<U> dropRight(int num)
Traversable
ListX.of(1,2,3).dropRight(2);
//[1]
dropRight in interface ReactiveSeq<U>dropRight in interface Traversable<U>num - Drop this number of elements from the take of this Traversabledefault FutureStream<U> takeRight(int num)
Traversable
ListX.of(1,2,3).takeRight(2);
//[2,3]
takeRight in interface ReactiveSeq<U>takeRight in interface Traversable<U>num - Take this number of elements from the take of this Traversabledefault FutureStream<U> skip(long n)
ReactiveSeq
assertThat(ReactiveSeq.of(4,3,6,7).skip(2).toList(),equalTo(Arrays.asList(6,7)));
skip in interface BaseSimpleReactStream<U>skip in interface ReactiveSeq<U>skip in interface java.util.stream.Stream<U>skip in interface Traversable<U>n - Number of elemenets to skipdefault FutureStream<U> distinct()
distinct in interface ReactiveSeq<U>distinct in interface java.util.stream.Stream<U>distinct in interface Traversable<U>default FutureStream<VectorX<U>> sliding(int size)
//futureStream of [1,2,3,4,5,6]
List<List<Integer>> list = futureStream.sliding(2)
.collect(CyclopsCollectors.toList());
assertThat(list.get(0),hasItems(1,2));
assertThat(list.get(1),hasItems(2,3));
sliding in interface ReactiveSeq<U>sliding in interface Traversable<U>size - Size of sliding windowdefault FutureStream<VectorX<U>> sliding(int size, int increment)
//futureStream of [1,2,3,4,5,6,7,8]
List<List<Integer>> list = futureStream.sliding(3,2)
.collect(CyclopsCollectors.toList());
assertThat(list.get(0),hasItems(1,2,3));
assertThat(list.get(1),hasItems(3,4,5));
sliding in interface ReactiveSeq<U>sliding in interface Traversable<U>size - Size of sliding windowincrement - for each windowdefault org.jooq.lambda.tuple.Tuple2<FutureStream<U>,FutureStream<U>> duplicateFutureStream()
default org.jooq.lambda.tuple.Tuple2<ReactiveSeq<U>,ReactiveSeq<U>> partition(java.util.function.Predicate<? super U> predicate)
// tuple((1, 3, 5), (2, 4, 6))
FutureStream.of(1, 2, 3, 4, 5,6).partition(i -> i % 2 != 0)
partition in interface ReactiveSeq<U>partitionFutureStream(Predicate),
#partition(Stream, Predicate)default org.jooq.lambda.tuple.Tuple2<FutureStream<U>,FutureStream<U>> partitionFutureStream(java.util.function.Predicate<? super U> predicate)
FutureStream.of(1, 2, 3, 4, 5, 6).partition(i -> i % 2 != 0)
results in
tuple((1, 3, 5), (2, 4, 6))
predicate - Predicate to split Streampartition(Predicate)default FutureStream<U> slice(long from, long to)
slice in interface ReactiveSeq<U>slice in interface Traversable<U>default FutureStream<org.jooq.lambda.tuple.Tuple2<U,java.lang.Long>> zipWithIndex()
// (tuple("a", 0), tuple("b", 1), tuple("c", 2))
FutureStream.of("a","b", "c").zipWithIndex()
zipWithIndex in interface ReactiveSeq<U>zipWithIndex in interface Traversable<U>#zipWithIndex(Stream)default <T> FutureStream<org.jooq.lambda.tuple.Tuple2<U,T>> zipS(java.util.stream.Stream<? extends T> other)
ReactiveSeq
List<Tuple2<Integer, String>> list = of(1, 2).zip(of("a", "b", "c", "d")).toList();
// [[1,"a"],[2,"b"]]
zipS in interface ReactiveSeq<U>zipS in interface Traversable<U>zipS in interface Zippable<U>other - Stream to combine withdefault <T> FutureStream<org.jooq.lambda.tuple.Tuple2<U,T>> zip(java.lang.Iterable<? extends T> other)
Zippablezip in interface ReactiveSeq<U>zip in interface Traversable<U>zip in interface Zippable<U>other - Iterable to combine withdefault <T,R> FutureStream<R> zipS(java.util.stream.Stream<? extends T> other, java.util.function.BiFunction<? super U,? super T,? extends R> zipper)
ZippablezipS in interface ReactiveSeq<U>zipS in interface Traversable<U>zipS in interface Zippable<U>other - Stream to combine withzipper - Zip / combining functiondefault <T,R> FutureStream<R> zip(java.lang.Iterable<? extends T> other, java.util.function.BiFunction<? super U,? super T,? extends R> zipper)
Zippablezip in interface ReactiveSeq<U>zip in interface Traversable<U>zip in interface Zippable<U>other - to zip withzipper - Zip functiondefault <T> FutureStream<T> scanLeft(T seed, java.util.function.BiFunction<? super T,? super U,? extends T> function)
scanLeft in interface ReactiveSeq<U>scanLeft in interface Traversable<U>default <R> FutureStream<R> scanRight(R seed, java.util.function.BiFunction<? super U,? super R,? extends R> function)
scanRight in interface ReactiveSeq<U>scanRight in interface Traversable<U>default FutureStream<U> scanRight(Monoid<U> monoid)
ReactiveSeq
assertThat(of("a", "b", "c").scanRight(Monoid.of("", String::concat)).toList().size(),
is(asList("", "c", "bc", "abc").size()));
scanRight in interface ReactiveSeq<U>scanRight in interface Traversable<U>default FutureStream<U> reverse()
reverse in interface ReactiveSeq<U>reverse in interface Traversable<U>default FutureStream<U> shuffle()
shuffle in interface ReactiveSeq<U>shuffle in interface Traversable<U>default FutureStream<U> shuffle(java.util.Random random)
shuffle in interface ReactiveSeq<U>shuffle in interface Traversable<U>default FutureStream<U> skipWhile(java.util.function.Predicate<? super U> predicate)
skipWhile in interface ReactiveSeq<U>skipWhile in interface Traversable<U>predicate - Predicate to skip while true#skipWhile(Stream, Predicate)default FutureStream<U> skipUntil(java.util.function.Predicate<? super U> predicate)
skipUntil in interface ReactiveSeq<U>skipUntil in interface Traversable<U>predicate - Predicate to skip until true#skipUntil(Stream, Predicate)default FutureStream<U> limitWhile(java.util.function.Predicate<? super U> predicate)
limitWhile in interface ReactiveSeq<U>limitWhile in interface Traversable<U>predicate - Limit while predicate is true#limitWhile(Stream, Predicate)default FutureStream<U> limitUntil(java.util.function.Predicate<? super U> predicate)
limitUntil in interface ReactiveSeq<U>limitUntil in interface Traversable<U>predicate - Limit until predicate is true#limitUntil(Stream, Predicate)default <T> FutureStream<org.jooq.lambda.tuple.Tuple2<U,T>> crossJoin(ReactiveSeq<? extends T> other)
// (tuple(1, "a"), tuple(1, "b"), tuple(2, "a"), tuple(2, "b"))
FutureStream.of(1, 2).crossJoin(FutureStream.of("a", "b"))
crossJoin in interface ReactiveSeq<U>other - Stream to crossJoindefault FutureStream<U> onEmpty(U value)
value, in case this reactiveStream is empty.
FutureStream.of().onEmpty(1)
//1
onEmpty in interface OnEmpty<U>onEmpty in interface ReactiveSeq<U>onEmpty in interface Traversable<U>default FutureStream<U> onEmptyGet(java.util.function.Supplier<? extends U> supplier)
supplier, in case this reactiveStream is empty.
FutureStream.of().onEmptyGet(() -> 1)
//1
onEmptyGet in interface OnEmpty<U>onEmptyGet in interface ReactiveSeq<U>onEmptyGet in interface Traversable<U>supplier - to determine new value for containerdefault <X extends java.lang.Throwable> FutureStream<U> onEmptyThrow(java.util.function.Supplier<? extends X> supplier)
supplier, in case this reactiveStream is empty.
FutureStream.of().capture(e -> ex = e).onEmptyThrow(() -> new RuntimeException()).toList();
//throws RuntimeException
onEmptyThrow in interface OnEmpty<U>onEmptyThrow in interface ReactiveSeq<U>onEmptyThrow in interface Traversable<U>supplier - to create exception fromdefault <T> FutureStream<org.jooq.lambda.tuple.Tuple2<U,T>> innerJoin(java.util.stream.Stream<? extends T> other, java.util.function.BiPredicate<? super U,? super T> predicate)
// (tuple(1, 1), tuple(2, 2))
FutureStream.of(1, 2, 3).innerJoin(Seq.of(1, 2), t -> Objects.equals(t.v1, t.v2))
default <T> FutureStream<org.jooq.lambda.tuple.Tuple2<U,T>> leftOuterJoin(java.util.stream.Stream<? extends T> other, java.util.function.BiPredicate<? super U,? super T> predicate)
// (tuple(1, 1), tuple(2, 2), tuple(3, null))
FutureStream.of(1, 2, 3).leftOuterJoin(Seq.of(1, 2), t -> Objects.equals(t.v1, t.v2))
default <T> FutureStream<org.jooq.lambda.tuple.Tuple2<U,T>> rightOuterJoin(java.util.stream.Stream<? extends T> other, java.util.function.BiPredicate<? super U,? super T> predicate)
// (tuple(1, 1), tuple(2, 2), tuple(null, 3))
FutureStream.of(1, 2).rightOuterJoin(Seq.of(1, 2, 3), t -> Objects.equals(t.v1, t.v2))
default FutureStream<U> cycle()
assertThat(FutureStream.of(1,2,2).cycle().limit(6)
.collect(CyclopsCollectors.toList()),
equalTo(Arrays.asList(1,2,2,1,2,2));
cycle in interface ReactiveSeq<U>default FutureStream<U> cycleWhile(java.util.function.Predicate<? super U> predicate)
int count =0;
assertThat(FutureStream.of(1,2,2).cycleWhile(next -> count++<6 )
.collect(CyclopsCollectors.toList()),equalTo(Arrays.asList(1,2,2,1,2,2)));
cycleWhile in interface ReactiveSeq<U>cycleWhile in interface Traversable<U>predicate - repeat while truedefault FutureStream<U> cycleUntil(java.util.function.Predicate<? super U> predicate)
count =0;
assertThat(FutureStream.of(1,2,2,3).cycleUntil(next -> count++>10 )
.collect(CyclopsCollectors.toList()),equalTo(Arrays.asList(1, 2, 2, 3, 1, 2, 2, 3, 1, 2, 2)));
cycleUntil in interface ReactiveSeq<U>cycleUntil in interface Traversable<U>predicate - repeat while truedefault ReactiveSeq<U> stream()
ReactiveSeqstream in interface ExtendedTraversable<U>stream in interface FoldableTraversable<U>stream in interface Folds<U>stream in interface ReactiveSeq<U>stream in interface Sequential<U>stream in interface TransformerTraversable<U>stream in interface Traversable<U>default FutureStream<U> parallel()
parallel in interface java.util.stream.BaseStream<U,java.util.stream.Stream<U>>parallel in interface ReactiveSeq<U>default FutureStream<U> sequential()
sequential in interface java.util.stream.BaseStream<U,java.util.stream.Stream<U>>sequential in interface ReactiveSeq<U>default FutureStream<U> unordered()
unordered in interface java.util.stream.BaseStream<U,java.util.stream.Stream<U>>unordered in interface ReactiveSeq<U>default FutureStream<U> onClose(java.lang.Runnable closeHandler)
onClose in interface java.util.stream.BaseStream<U,java.util.stream.Stream<U>>onClose in interface ReactiveSeq<U>default FutureStream<U> sorted()
ReactiveSeq
assertThat(ReactiveSeq.of(4,3,6,7)).sorted().toList(),equalTo(Arrays.asList(3,4,6,7)));
sorted in interface ReactiveSeq<U>sorted in interface java.util.stream.Stream<U>sorted in interface Traversable<U>default FutureStream<U> sorted(java.util.Comparator<? super U> comparator)
ReactiveSeq
assertThat(ReactiveSeq.of(4,3,6,7).sorted((a,b) -> b-a).toList(),equalTo(Arrays.asList(7,6,4,3)));
sorted in interface ReactiveSeq<U>sorted in interface java.util.stream.Stream<U>sorted in interface Traversable<U>comparator - Compartor to sort withdefault FutureStream<U> self(java.util.function.Consumer<FutureStream<U>> consumer)
consumer - Consumer that will recieve current stagestatic <T1> FutureStream<T1> flatten(ReactiveSeq<? extends java.util.stream.Stream<? extends T1>> nested)
static <T1> FutureStream<T1> narrow(FutureStream<? extends T1> broad)
default java.util.Spliterator<U> spliterator()
ReactiveSeq
This is a terminal operation.
spliterator in interface java.util.stream.BaseStream<U,java.util.stream.Stream<U>>spliterator in interface java.lang.Iterable<U>spliterator in interface ReactiveSeq<U>default boolean isParallel()
default java.util.stream.IntStream mapToInt(java.util.function.ToIntFunction<? super U> mapper)
mapToInt in interface ReactiveSeq<U>mapToInt in interface java.util.stream.Stream<U>mapToInt in interface Traversable<U>default java.util.stream.LongStream mapToLong(java.util.function.ToLongFunction<? super U> mapper)
mapToLong in interface ReactiveSeq<U>mapToLong in interface java.util.stream.Stream<U>mapToLong in interface Traversable<U>default java.util.stream.DoubleStream mapToDouble(java.util.function.ToDoubleFunction<? super U> mapper)
mapToDouble in interface ReactiveSeq<U>mapToDouble in interface java.util.stream.Stream<U>mapToDouble in interface Traversable<U>default java.util.stream.IntStream flatMapToInt(java.util.function.Function<? super U,? extends java.util.stream.IntStream> mapper)
flatMapToInt in interface java.util.stream.Stream<U>default java.util.stream.LongStream flatMapToLong(java.util.function.Function<? super U,? extends java.util.stream.LongStream> mapper)
flatMapToLong in interface java.util.stream.Stream<U>default java.util.stream.DoubleStream flatMapToDouble(java.util.function.Function<? super U,? extends java.util.stream.DoubleStream> mapper)
flatMapToDouble in interface java.util.stream.Stream<U>default void forEachOrdered(java.util.function.Consumer<? super U> action)
forEachOrdered in interface java.util.stream.Stream<U>default java.lang.Object[] toArray()
toArray in interface java.util.stream.Stream<U>default <A> A[] toArray(java.util.function.IntFunction<A[]> generator)
toArray in interface java.util.stream.Stream<U>default java.util.Optional<U> findAny()
findAny in interface Folds<U>findAny in interface ReactiveSeq<U>findAny in interface java.util.stream.Stream<U>
ReactiveSeq.of(1,2,3,4,5).filter(it -> it <3).findAny().get();
//3
(non-deterministic)default java.util.Set<U> toSet()
CyclopsCollectableLinkedHashSet.toSet in interface CyclopsCollectable<U>toSet in interface ReactiveSeq<U>default java.util.List<U> toList()
CyclopsCollectableArrayList.toList in interface CyclopsCollectable<U>toList in interface ReactiveSeq<U>default <R> FutureStream<U> distinct(java.util.function.Function<? super U,? extends R> keyExtractor)
distinct in interface ReactiveSeq<U>default org.jooq.lambda.tuple.Tuple2<ReactiveSeq<U>,ReactiveSeq<U>> duplicate()
ReactiveSeq
Tuple2<ReactiveSeq<Integer>, ReactiveSeq<Integer>> copies = of(1, 2, 3, 4, 5, 6)
.duplicate();
assertTrue(copies.v1.anyMatch(i > i == 2));
assertTrue(copies.v2.anyMatch(i > i == 2));
duplicate in interface ReactiveSeq<U>default org.jooq.lambda.tuple.Tuple2<ReactiveSeq<U>,ReactiveSeq<U>> duplicate(java.util.function.Supplier<java.util.Deque<U>> bufferFactory)
duplicate in interface ReactiveSeq<U>default org.jooq.lambda.tuple.Tuple3<ReactiveSeq<U>,ReactiveSeq<U>,ReactiveSeq<U>> triplicate()
ReactiveSeq
Tuple3<ReactiveSeq<Tuple3<T1, T2, T3>>, ReactiveSeq<Tuple3<T1, T2, T3>>, ReactiveSeq<Tuple3<T1, T2, T3>>> Tuple3 = sequence.triplicate();
triplicate in interface ReactiveSeq<U>default org.jooq.lambda.tuple.Tuple3<ReactiveSeq<U>,ReactiveSeq<U>,ReactiveSeq<U>> triplicate(java.util.function.Supplier<java.util.Deque<U>> bufferFactory)
triplicate in interface ReactiveSeq<U>default org.jooq.lambda.tuple.Tuple4<ReactiveSeq<U>,ReactiveSeq<U>,ReactiveSeq<U>,ReactiveSeq<U>> quadruplicate()
ReactiveSeq
Tuple4<ReactiveSeq<Tuple4<T1, T2, T3, T4>>,
ReactiveSeq<Tuple4<T1, T2, T3, T4>>,
ReactiveSeq<Tuple4<T1, T2, T3, T4>>,
ReactiveSeq<Tuple4<T1, T2, T3, T4>>> quad = sequence.quadruplicate();
quadruplicate in interface ReactiveSeq<U>default org.jooq.lambda.tuple.Tuple4<ReactiveSeq<U>,ReactiveSeq<U>,ReactiveSeq<U>,ReactiveSeq<U>> quadruplicate(java.util.function.Supplier<java.util.Deque<U>> bufferFactory)
quadruplicate in interface ReactiveSeq<U>default org.jooq.lambda.tuple.Tuple2<java.util.Optional<U>,ReactiveSeq<U>> splitAtHead()
ReactiveSeq
ReactiveSeq.of(1,2,3)
.splitAtHead()
//Optional[1], ReactiveSeq[2,3]
splitAtHead in interface ReactiveSeq<U>default org.jooq.lambda.tuple.Tuple2<ReactiveSeq<U>,ReactiveSeq<U>> splitAt(int where)
ReactiveSeq
ReactiveSeq.of(1,2,3)
.splitAt(1)
//ReactiveSeq[1], ReactiveSeq[2,3]
splitAt in interface ReactiveSeq<U>default org.jooq.lambda.tuple.Tuple2<ReactiveSeq<U>,ReactiveSeq<U>> splitBy(java.util.function.Predicate<U> splitter)
ReactiveSeq
ReactiveSeq.of(1, 2, 3, 4, 5, 6).splitBy(i->i<4)
//ReactiveSeq[1,2,3] ReactiveSeq[4,5,6]
splitBy in interface ReactiveSeq<U>default FutureStream<U> cycle(Monoid<U> m, long times)
ReactiveSeq
List<Integer> list = ReactiveSeq.of(1,2,2))
.cycle(Reducers.toCountInt(),3)
.collect(CyclopsCollectors.toList());
//List[3,3,3];
cycle in interface ReactiveSeq<U>cycle in interface Traversable<U>m - Monoid to be used in reductiontimes - Number of times value should be repeateddefault <S,R> FutureStream<org.jooq.lambda.tuple.Tuple3<U,S,R>> zip3(java.lang.Iterable<? extends S> second, java.lang.Iterable<? extends R> third)
ReactiveSeq
List<Tuple3<Integer, Integer, Character>> list = of(1, 2, 3, 4, 5, 6).zip3(of(100, 200, 300, 400), of('a', 'b', 'c')).collect(CyclopsCollectors.toList());
// [[1,100,'a'],[2,200,'b'],[3,300,'c']]
zip3 in interface ReactiveSeq<U>zip3 in interface Traversable<U>zip3 in interface Zippable<U>default <T2,T3,T4> FutureStream<org.jooq.lambda.tuple.Tuple4<U,T2,T3,T4>> zip4(java.lang.Iterable<? extends T2> second, java.lang.Iterable<? extends T3> third, java.lang.Iterable<? extends T4> fourth)
ReactiveSeq
List<Tuple4<Integer, Integer, Character, String>> list = of(1, 2, 3, 4, 5, 6).zip4(of(100, 200, 300, 400), of('a', 'b', 'c'), of("hello", "world"))
.collect(CyclopsCollectors.toList());
// [[1,100,'a',"hello"],[2,200,'b',"world"]]
zip4 in interface ReactiveSeq<U>zip4 in interface Traversable<U>zip4 in interface Zippable<U>default FutureStream<ListX<U>> grouped(int groupSize)
ReactiveSeq
List<List<Integer>> list = ReactiveSeq.of(1, 2, 3, 4, 5, 6).grouped(3).collect(CyclopsCollectors.toList());
assertThat(list.get(0), hasItems(1, 2, 3));
assertThat(list.get(1), hasItems(4, 5, 6));
grouped in interface ReactiveSeq<U>grouped in interface Traversable<U>groupSize - Size of each Groupdefault FutureStream<U> scanLeft(Monoid<U> monoid)
ReactiveSeq
assertEquals(asList("", "a", "ab", "abc"),ReactiveSeq.of("a", "b", "c")
.scanLeft(Reducers.toString("")).toList());
scanLeft in interface ReactiveSeq<U>scanLeft in interface Traversable<U>default boolean startsWithIterable(java.lang.Iterable<U> iterable)
ReactiveSeq
assertTrue(ReactiveSeq.of(1,2,3,4).startsWith(Arrays.asList(1,2,3)));
startsWithIterable in interface Folds<U>startsWithIterable in interface ReactiveSeq<U>default boolean startsWith(java.util.stream.Stream<U> iterator)
ReactiveSeq
assertTrue(ReactiveSeq.of(1,2,3,4).startsWith(Stream.of(1,2,3)))
startsWith in interface Folds<U>startsWith in interface ReactiveSeq<U>iterator - Stream to check if this Folds has the same elements in the same order, at the skipdefault boolean endsWithIterable(java.lang.Iterable<U> iterable)
ReactiveSeq
assertTrue(ReactiveSeq.of(1,2,3,4,5,6)
.endsWith(Arrays.asList(5,6)));
endsWithIterable in interface Folds<U>endsWithIterable in interface ReactiveSeq<U>iterable - Values to checkdefault boolean endsWith(java.util.stream.Stream<U> stream)
ReactiveSeq
assertTrue(ReactiveSeq.of(1,2,3,4,5,6)
.endsWith(Stream.of(5,6)));
default AnyMSeq<Witness.reactiveSeq,U> anyM()
anyM in interface ReactiveSeq<U>default <R> FutureStream<R> flatMapI(java.util.function.Function<? super U,? extends java.lang.Iterable<? extends R>> fn)
ReactiveSeq
ReactiveSeq.of(1,2)
.flatMap(i -> asList(i, -i))
.toList();
//1,-1,2,-2
flatMapI in interface ReactiveSeq<U>default void forEachAsync(java.util.function.Consumer<? super U> action)
ReactiveSeq.forEach(java.util.function.Consumer<? super T>).
For push based reactiveBuffer Stream types (created via Spouts of FutureStream}forEachAsync in interface ReactiveSeq<U>action - a
non-interfering action to perform on the elementsdefault <R> FutureStream<R> flatMapP(java.util.function.Function<? super U,? extends org.reactivestreams.Publisher<? extends R>> fn)
flatMapP in interface ReactiveSeq<U>default <R> FutureStream<R> flatMapP(int maxConcurrency, java.util.function.Function<? super U,? extends org.reactivestreams.Publisher<? extends R>> fn)
flatMapP in interface ReactiveSeq<U>default <R> FutureStream<R> flatMapStream(java.util.function.Function<? super U,java.util.stream.BaseStream<? extends R,?>> fn)
ReactiveSeq
assertThat(ReactiveSeq.of(1,2,3)
.flatMapStream(i->IntStream.of(i))
.toList(),equalTo(Arrays.asList(1,2,3)));
flatMapStream in interface ReactiveSeq<U>fn - to be applieddefault FutureStream<U> append(U... values)
ReactiveSeq
List<String> result = ReactiveSeq.of(1, 2, 3).append(100, 200, 300).map(it -> it + "!!").collect(CyclopsCollectors.toList());
assertThat(result, equalTo(Arrays.asList("1!!", "2!!", "3!!", "100!!", "200!!", "300!!"))); *
append in interface ReactiveSeq<U>append in interface Traversable<U>values - to appenddefault FutureStream<U> prepend(U... values)
ReactiveSeq
List<String> result = ReactiveSeq.of(1,2,3)
.prepend(100,200,300)
.map(it ->it+"!!")
.collect(CyclopsCollectors.toList());
assertThat(result,equalTo(Arrays.asList("100!!","200!!","300!!","1!!","2!!","3!!")));
prepend in interface ReactiveSeq<U>prepend in interface Traversable<U>values - to prependdefault FutureStream<U> insertAt(int pos, U... values)
ReactiveSeq
List<String> result = ReactiveSeq.of(1, 2, 3).insertAt(1, 100, 200, 300).map(it -> it + "!!").collect(CyclopsCollectors.toList());
assertThat(result, equalTo(Arrays.asList("1!!", "100!!", "200!!", "300!!", "2!!", "3!!"))); *
insertAt in interface ReactiveSeq<U>insertAt in interface Traversable<U>pos - to insert data atvalues - to insertdefault FutureStream<U> deleteBetween(int start, int end)
ReactiveSeq
List<String> result = ReactiveSeq.of(1, 2, 3, 4, 5, 6).deleteBetween(2, 4).map(it -> it + "!!").collect(CyclopsCollectors.toList());
assertThat(result, equalTo(Arrays.asList("1!!", "2!!", "5!!", "6!!"))); *
deleteBetween in interface ReactiveSeq<U>deleteBetween in interface Traversable<U>start - indexend - indexdefault FutureStream<U> insertAtS(int pos, java.util.stream.Stream<U> stream)
ReactiveSeq
List<String> result = ReactiveSeq.of(1, 2, 3).insertAtS(1, of(100, 200, 300)).map(it -> it + "!!").collect(CyclopsCollectors.toList());
assertThat(result, equalTo(Arrays.asList("1!!", "100!!", "200!!", "300!!", "2!!", "3!!")));
insertAtS in interface ReactiveSeq<U>insertAtS in interface Traversable<U>pos - to insert Stream atstream - to insertdefault FutureStream<U> skip(long time, java.util.concurrent.TimeUnit unit)
ReactiveSeq
List<Integer> result = ReactiveSeq.of(1, 2, 3, 4, 5, 6).peek(i -> sleep(i * 100)).skip(1000, TimeUnit.MILLISECONDS).toList();
// [4,5,6]
skip in interface ReactiveSeq<U>time - Length of timeunit - Time unitdefault FutureStream<U> limit(long time, java.util.concurrent.TimeUnit unit)
ReactiveSeq
List<Integer> result = ReactiveSeq.of(1, 2, 3, 4, 5, 6).peek(i -> sleep(i * 100)).limit(1000, TimeUnit.MILLISECONDS).toList();
// [1,2,3,4]
limit in interface ReactiveSeq<U>time - Length of timeunit - Time unitdefault FutureStream<U> skipLast(int num)
ReactiveSeqskipLast in interface ReactiveSeq<U>skipLast in interface Traversable<U>default FutureStream<U> limitLast(int num)
ReactiveSeq
assertThat(ReactiveSeq.of(1,2,3,4,5)
.limitLast(2)
.collect(CyclopsCollectors.toList()),equalTo(Arrays.asList(4,5)));
limitLast in interface ReactiveSeq<U>limitLast in interface Traversable<U>num - of elements to return (last elements)default U firstValue()
ReactiveSeq
assertThat(ReactiveSeq.of(1,2,3,4)
.map(u->{throw new RuntimeException();})
.recover(e->"hello")
.firstValue(),equalTo("hello"));
firstValue in interface Folds<U>firstValue in interface ReactiveSeq<U>default <C extends java.util.Collection<? super U>> FutureStream<C> groupedBySizeAndTime(int size, long time, java.util.concurrent.TimeUnit unit, java.util.function.Supplier<C> factory)
ReactiveSeq
ReactiveSeq.of(1,2,3,4,5,6)
.groupedBySizeAndTime(3,10,TimeUnit.SECONDS,()->SetX.empty())
.toList();
//[[1,2,3],[4,5,6]]
groupedBySizeAndTime in interface ReactiveSeq<U>size - Max size of a batchtime - (Max) time period to build a singleUnsafe batch inunit - time unit for batchfactory - Collection factorydefault FutureStream<ListX<U>> groupedStatefullyUntil(java.util.function.BiPredicate<ListX<? super U>,? super U> predicate)
ReactiveSeq
ReactiveSeq.of(1,2,3,4,5,6)
.groupedStatefullyUntil((s,i)-> s.contains(4) ? true : false)
.toList()
.size()
//5
groupedStatefullyUntil in interface ReactiveSeq<U>groupedStatefullyUntil in interface Traversable<U>predicate - Window while truedefault <C extends java.util.Collection<U>,R> FutureStream<R> groupedStatefullyUntil(java.util.function.BiPredicate<C,? super U> predicate, java.util.function.Supplier<C> factory, java.util.function.Function<? super C,? extends R> finalizer)
groupedStatefullyUntil in interface ReactiveSeq<U>default FutureStream<ListX<U>> groupedStatefullyWhile(java.util.function.BiPredicate<ListX<? super U>,? super U> predicate)
groupedStatefullyWhile in interface ReactiveSeq<U>default <C extends java.util.Collection<U>,R> FutureStream<R> groupedStatefullyWhile(java.util.function.BiPredicate<C,? super U> predicate, java.util.function.Supplier<C> factory, java.util.function.Function<? super C,? extends R> finalizer)
groupedStatefullyWhile in interface ReactiveSeq<U>default FutureStream<ListX<U>> groupedUntil(java.util.function.Predicate<? super U> predicate)
ReactiveSeq
assertThat(ReactiveSeq.of(1,2,3,4,5,6)
.batchUntil(i->i%3==0)
.toList()
.size(),equalTo(2));
groupedUntil in interface ReactiveSeq<U>groupedUntil in interface Traversable<U>predicate - Batch until predicate holds, applyHKT open next batchdefault FutureStream<ListX<U>> groupedWhile(java.util.function.Predicate<? super U> predicate)
ReactiveSeq
assertThat(ReactiveSeq.of(1,2,3,4,5,6)
.batchWhile(i->i%3!=0)
.toList().size(),equalTo(2));
groupedWhile in interface ReactiveSeq<U>groupedWhile in interface Traversable<U>predicate - Batch while predicate holds, applyHKT open next batchdefault <C extends java.util.Collection<? super U>> FutureStream<C> groupedWhile(java.util.function.Predicate<? super U> predicate, java.util.function.Supplier<C> factory)
ReactiveSeq
assertThat(ReactiveSeq.of(1,2,3,4,5,6)
.batchWhile(i->i%3!=0)
.toList()
.size(),equalTo(2));
groupedWhile in interface ReactiveSeq<U>groupedWhile in interface Traversable<U>predicate - Batch while predicate holds, applyHKT open next batchfactory - Collection factorydefault <R extends java.lang.Comparable<? super R>> FutureStream<U> sorted(java.util.function.Function<? super U,? extends R> function)
sorted in interface ReactiveSeq<U>sorted in interface Traversable<U>default <C extends java.util.Collection<? super U>> FutureStream<C> groupedUntil(java.util.function.Predicate<? super U> predicate, java.util.function.Supplier<C> factory)
ReactiveSeq
assertThat(ReactiveSeq.of(1,2,3,4,5,6)
.batchUntil(i->i%3!=0)
.toList()
.size(),equalTo(2));
groupedUntil in interface ReactiveSeq<U>groupedUntil in interface Traversable<U>predicate - Batch until predicate holds, applyHKT open next batchfactory - Collection factorydefault FutureStream<U> recover(java.util.function.Function<? super java.lang.Throwable,? extends U> fn)
ReactiveSeq
assertThat(ReactiveSeq.of(1,2,3,4)
.map(i->i+2)
.map(u->{throw new RuntimeException();})
.recover(e->"hello")
.firstValue(),equalTo("hello"));
recover in interface ReactiveSeq<U>recover in interface Traversable<U>fn - Function that accepts a Throwable and returns an alternative
valuedefault <EX extends java.lang.Throwable> FutureStream<U> recover(java.lang.Class<EX> exceptionClass, java.util.function.Function<? super EX,? extends U> fn)
ReactiveSeq
assertThat(ReactiveSeq.of(1,2,3,4)
.map(i->i+2)
.map(u->{ExceptionSoftener.throwSoftenedException( new IOException()); return null;})
.recover(IOException.class,e->"hello")
.firstValue(),equalTo("hello"));
recover in interface ReactiveSeq<U>recover in interface Traversable<U>exceptionClass - Type to recover fromfn - That accepts an error and returns an alternative valuedefault <X extends java.lang.Throwable> org.reactivestreams.Subscription forEach(long numberOfElements,
java.util.function.Consumer<? super U> consumer)
Subscription next = FutureStream.of(1,2,3,4)
.forEach(2,System.out::println);
System.out.println("First batch processed!");
next.request(2);
System.out.println("Second batch processed!");
//prints
1
2
First batch processed!
3
4
Second batch processed!
forEach in interface FoldableTraversable<U>forEach in interface ReactiveStreamsTerminalOperations<U>numberOfElements - To consume from the Stream at this timeconsumer - To accept incoming events from the Streamdefault <X extends java.lang.Throwable> org.reactivestreams.Subscription forEach(long numberOfElements,
java.util.function.Consumer<? super U> consumer,
java.util.function.Consumer<? super java.lang.Throwable> consumerError)
Subscription next = FutureStream.of(()->1,()->2,()->{throw new RuntimeException()},()->4)
.map(Supplier::get)
.forEach(2,System.out::println, e->e.printStackTrace());
System.out.println("First batch processed!");
next.request(2);
System.out.println("Second batch processed!");
//prints
1
2
First batch processed!
RuntimeException Stack Trace on System.err
4
Second batch processed!
forEach in interface FoldableTraversable<U>forEach in interface ReactiveStreamsTerminalOperations<U>numberOfElements - To consume from the Stream at this timeconsumer - To accept incoming elements from the StreamconsumerError - To accept incoming processing errors from the Streamdefault <X extends java.lang.Throwable> org.reactivestreams.Subscription forEach(long numberOfElements,
java.util.function.Consumer<? super U> consumer,
java.util.function.Consumer<? super java.lang.Throwable> consumerError,
java.lang.Runnable onComplete)
Subscription next = LazyFurtureStream.of(()->1,()->2,()->{throw new RuntimeException()},()->4)
.map(Supplier::get)
.forEach(2,System.out::println, e->e.printStackTrace(),()->System.out.println("the take!"));
System.out.println("First batch processed!");
next.request(2);
System.out.println("Second batch processed!");
//prints
1
2
First batch processed!
RuntimeException Stack Trace on System.err
4
Second batch processed!
The take!
forEach in interface FoldableTraversable<U>forEach in interface ReactiveStreamsTerminalOperations<U>numberOfElements - To consume from the Stream at this timeconsumer - To accept incoming elements from the StreamconsumerError - To accept incoming processing errors from the StreamonComplete - To run after an onComplete eventdefault <X extends java.lang.Throwable> void forEach(java.util.function.Consumer<? super U> consumerElement, java.util.function.Consumer<? super java.lang.Throwable> consumerError)
Subscription next = FutureStream.of(()->1,()->2,()->{throw new RuntimeException()},()->4)
.map(Supplier::get)
.forEach(System.out::println, e->e.printStackTrace());
System.out.println("processed!");
//prints
1
2
RuntimeException Stack Trace on System.err
4
processed!
forEach in interface FoldableTraversable<U>forEach in interface ReactiveStreamsTerminalOperations<U>consumerElement - To accept incoming elements from the StreamconsumerError - To accept incoming processing errors from the Streamdefault <X extends java.lang.Throwable> void forEach(java.util.function.Consumer<? super U> consumerElement, java.util.function.Consumer<? super java.lang.Throwable> consumerError, java.lang.Runnable onComplete)
Subscription next = FutureStream.of(()->1,()->2,()->{throw new RuntimeException()},()->4)
.map(Supplier::get)
.forEachEvents(System.out::println, e->e.printStackTrace(),()->System.out.println("the take!"));
System.out.println("processed!");
//prints
1
2
RuntimeException Stack Trace on System.err
4
processed!
forEach in interface FoldableTraversable<U>forEach in interface ReactiveStreamsTerminalOperations<U>consumerElement - To accept incoming elements from the StreamconsumerError - To accept incoming processing errors from the StreamonComplete - To run after an onComplete eventdefault FutureStream<U> complete(java.lang.Runnable fn)
ReactiveSeqcomplete in interface ReactiveSeq<U>fn - Function that accepts and returns completed value