Interface Streamable<T>

Type Parameters:

T - the element type of the Streamable sequence.

All Known Subinterfaces:

StreamProcessor<In,Out>

All Known Implementing Classes:

DispatchStreamProcessor, GroupedStreamable

Functional Interface:

This is a functional interface and can therefore be used as the assignment target for a lambda expression or method reference.

@FunctionalInterface
public interface Streamable<@NonNull T>

Represents a virtual-thread capable, multi-valued (a)synchronous sequence of values that builds upon the Java CompletionStage-based concurrency-coordination model.

The lifecycle of the sequence is as follows:

• consumer calls stream(StreamerCancellation)
• consumer calls Streamer.next() in a loop and consumer checks if what the CompletionStage outcome is:
  • if it succeeded with a boolean true, it is safe to call Streamer.current().
  • if it succeeded with a boolean false, no further values are coming.
  • if it failed with a Throwable, no further values are coming and the error can be propagated further
• consumer calls Streamer.finish()

It is always necessary to have the consumer call finish because that is responsible for cleaning up resources of the upstream.

Downstream cancellations are signaled via the StreamerCancellation, where operators can register their own Disposables that get disposed. Because dispose can happen at any time and asynchronously to the consumption loop, the sensitive sources must complete their waiting CompletionStage returned by next exceptionally via a CancellationException. This will unblock the loops and invoke the finish method of the lifecycle at the consumer thread. Depending on the operator, the CancellationException may not be propagated further.

If a source wishes to fail, it must signal the Throwable via the returned CompletionStage of next. If the finish also throws, its Throwable should be added as suppressed exception to the original Throwable.

The Streamer methods must be invoked sequentially and non-overlappingly, similar to the Reactive Streams rule §1.3.

This reactive type was modeled after the C# IAsyncEnumerable and IAsyncEnumerator interfaces. Unfortunately, Java never added any async/await infrastructure, plus CompletionStage doesn't even have any native way to blockingly join to it. Therefore, when running in a (virtual) blocking fashion, one may use the Streamer.awaitNext() or Streamer.awaitFinish() helper methods.

Since:

4.0.0

Method Summary

Modifier and Type	Method	Description
default T	blockingFirst()	Blocks the current thread until this Streamable produces one item, which is then returned.
default T	blockingLast()	Blocks the current thread until this Streamable produces all of its items and the very last is then returned.
default <A,R> @NonNull Streamable<R>	collect(@NonNull Collector<@NonNull T, A, R> collector)	Collects all upstream values via the use of a Collector configuration and emits its resulting value as a single item of the returned Streamable.
static <@NonNull T> @NonNull Streamable<T>	concat(Streamable<? extends Streamable<? extends @NonNull T>> sources, ExecutorService executor)	Emits the elements of each inner sequence produced by the outer sequence.
static <@NonNull T> @NonNull Streamable<T>	concat(Iterable<? extends Streamable<? extends @NonNull T>> sources)	Streams the Streamables one after the other from the given iterable sequence of Streamables.
static <@NonNull T> @NonNull Streamable<T>	create(@NonNull VirtualGenerator<@NonNull T> generator)	Generate a sequence of values via a virtual generator callback (yielder) which is free to block and is natively backpressured.
static <@NonNull T> @NonNull Streamable<T>	create(@NonNull VirtualGenerator<@NonNull T> generator, @NonNull Scheduler scheduler)	Generate a sequence of values via a virtual generator callback (yielder) which is free to block and is natively backpressured.
static <@NonNull T> @NonNull Streamable<T>	create(@NonNull VirtualGenerator<@NonNull T> generator, @NonNull ExecutorService executor)	Generate a sequence of values via a virtual generator callback (yielder) which is free to block and is natively backpressured.
static <@NonNull T> @NonNull Streamable<T>	defer(Supplier<? extends Streamable<? extends @NonNull T>> supplier)	Defers the creation of the actual Streamable, allowing a per streamer state to be created along with it.
default @NonNull Streamable<T>	delay(long time, @NonNull TimeUnit unit, @NonNull Scheduler scheduler)	Delays the delivery of each upstream item by the given time amount.
default @NonNull Streamable<T>	doOnError(@NonNull Consumer<? super Throwable> consumer)	Calls the specific Consumer if there is an error from the upstream.
default @NonNull Streamable<T>	doOnNext(@NonNull Consumer<? super @NonNull T> consumer)	Calls the given consumer whenever an upstream item becomes available.
static <@NonNull T> @NonNull Streamable<T>	empty()	Returns an empty Streamable that never produces an item and just completes.
static <@NonNull T> @NonNull Streamable<T>	error(Throwable throwable)	Creates a Streamable that signals the given Throwable immediately when it begins streaming, ending the sequence.
default @NonNull Streamable<T>	filter(@NonNull Predicate<? super @NonNull T> predicate)	Filters out the upstream items that do not pass the given predicate.
default <R> @NonNull Streamable<R>	flatMap(@NonNull Function<? super @NonNull T, ? extends Streamable<? extends R>> mapper, @NonNull StandardConcurrentConfig config)	Maps each upstream item onto a Streamable and runs them concurrently while relaying inner items as first-come-first-served manner.
default @NonNull CompletionStageDisposable<Void>	forEach(@NonNull BiConsumer<? super @NonNull T, ? super Disposable> consumer, @NonNull DisposableStreamerCancellation canceller, @NonNull ExecutorService executor)	Consumes elements from this Streamable via the provided executor service.
default @NonNull CompletionStageDisposable<Void>	forEach(@NonNull Consumer<? super @NonNull T> consumer)	Consumes elements from this Streamable via the provided executor service.
default @NonNull CompletionStageDisposable<Void>	forEach(@NonNull Consumer<? super @NonNull T> consumer, @NonNull DisposableStreamerCancellation canceller)	Consumes elements from this Streamable via the provided executor service.
default @NonNull CompletionStageDisposable<Void>	forEach(@NonNull Consumer<? super @NonNull T> consumer, @NonNull DisposableStreamerCancellation canceller, @NonNull ExecutorService executor)	Consumes elements from this Streamable via the provided executor service.
default @NonNull CompletionStageDisposable<Void>	forEach(@NonNull Consumer<? super @NonNull T> consumer, @NonNull ExecutorService executor)	Consumes elements from this Streamable via the provided executor service.
static <@NonNull T> @NonNull Streamable<T>	fromArray(@NonNull T... items)	Streams all elements of the given items array.
static <@NonNull T> Streamable<T>	fromCompletable(@NonNull CompletableSource source)	Convert a CompletableSource into a Streamable and relay its terminal events.
static <@NonNull T> @NonNull Streamable<T>	fromIterable(@NonNull Iterable<? extends @NonNull T> items)	Streams all elements of the given Iterable sequence.
static <@NonNull T> Streamable<T>	fromMaybe(@NonNull MaybeSource<@NonNull T> source)	Convert a MaybeSource into a Streamable and relay its terminal events.
static <T> @NonNull Streamable<T>	fromPublisher(Flow.Publisher<T> source)	Convert any Flow.Publisher into a Streamable sequence.
static <T> @NonNull Streamable<T>	fromPublisher(Flow.Publisher<T> source, @NonNull ExecutorService executor)	Convert any Flow.Publisher into a Streamable sequence.
static <@NonNull T> Streamable<T>	fromSingle(@NonNull SingleSource<@NonNull T> source)	Convert a SingleSource into a Streamable and relay its terminal events.
static <@NonNull T> @NonNull Streamable<T>	fromStream(@NonNull Stream<? extends @NonNull T> items)	Streams all elements of the given Stream sequence.
default <@Nullable K> @NonNull Streamable<GroupedStreamable<K,T>>	groupBy(@NonNull Function<? super @NonNull T, ? extends @Nullable K> keySelector)	Maps each upstream item into a GroupedStreamable group, emits those groups and keeps relaying the upstream items into those groups.
default @NonNull Streamable<T>	hide()	Hides the identity of this Streamable and its Streamer.
default @NonNull Streamable<T>	ignoreElements()	Ignores all elements from the current Streamable and completes.
default @NonNull Streamable<T>	intercept(StreamableInterceptConfig<@NonNull T> config)	Intercepts the lifecycle method calls of Streamable and Streamer and allows the modification of them via Function callbacks.
static Streamable<Long>	intervalRange(long start, long count, long initialDelay, long period, TimeUnit unit, Scheduler scheduler)	Constructs a Streamable that after the initial delay, starts emitting an ever increasing numbers from start up to start + count exclusive with the given period.
static Streamable<Long>	intervalRange(long start, long count, long initialDelay, long period, TimeUnit unit, ExecutorService executor)	Constructs a Streamable that after the initial delay, starts emitting an ever increasing numbers from start up to start + count exclusive with the given period.
static <@NonNull T> @NonNull Streamable<T>	just(@NonNull T item)	Returns a single-element Streamable that produces the constant item and completes.
default @NonNull Single<T>	last(@NonNull T defaultItem)	Signals the last item from the upstream Streamable or the provided defaultItem if the upstream is empty, as a Single instance.
default @NonNull Single<T>	lastOrError()	Signals the last item from the upstream Streamable or a NoSuchElementException if the upstream is empty, as a Single instance.
default <@NonNull R> @NonNull Streamable<R>	lift(@NonNull StreamableOperator<? super @NonNull T, ? extends @NonNull R> lifter)	This method requires advanced knowledge about building operators, please consider other standard composition methods first; Returns a Streamable instance which when its stream(StreamerCancellation) is invoked, applies the specified operator callback to the upstream Streamer to produce an actual Streamer instance to be handed downstream.
default <@NonNull R> @NonNull Streamable<R>	map(@NonNull Function<? super @NonNull T, ? extends @NonNull R> mapper)	Maps each upstream item into another item via a mapper function.
default <@NonNull R> @NonNull Streamable<R>	mapOptional(@NonNull Function<? super @NonNull T, ? extends Optional<? extends @NonNull R>> mapper)	Maps each upstream item into another, optional item via a mapper function that skips the empty optionals.
static <@NonNull T> @NonNull Streamable<T>	never()	Returns an Streamable that never produces an item and never terminates.
default @NonNull Streamable<T>	onErrorResumeNext(@NonNull Function<? super Throwable, ? extends Streamable<? extends @NonNull T>> fallbackMapper)	When the upstream fails, the sequence is resumed by the Streamable that is returned for the failure Throwable.
static @NonNull Streamable<Integer>	range(int start, int count)	Emits elements from start up to start + count exclusive.
static @NonNull Streamable<Long>	rangeLong(long start, long count)	Emits elements from start up to start + count exclusive.
default @NonNull Streamable<T>	repeat(long count)	Runs the upstream at most the given count times while it succeeds.
default @NonNull Streamable<T>	repeatWhen(Function<? super Long, ? extends CompletionStage<Boolean>> whenFunction)	Repeats the upstream when the given function signals true via the CompletionStage for the count how many times the upstream was streamed.
default @NonNull Streamable<T>	retry(long count)	Retries at most the given count times the upstream if it fails with any error.
default @NonNull Streamable<T>	retry(BiPredicate<? super Long, ? super Throwable> predicate)	Retries the upstream if the given predicate returns true for the failure count and Throwable of the last streaming of the upstream.
default @NonNull Streamable<T>	retry(Predicate<? super Throwable> predicate)	Retries the upstream if the given predicate returns true for the failure Throwable of the last streaming of the upstream.
default @NonNull Streamable<T>	retryWhen(BiFunction<? super Long, ? super Throwable, ? extends CompletionStage<Boolean>> whenFunction)	Retries the upstream when the given predicate signals true via the CompletionStage for the failure count and Throwable of the last streaming of the upstream.
default @NonNull Streamable<T>	skip(long count)	Skips the first count items and relays the rest to the downstream.
@NonNull Streamer<T>	stream(@NonNull StreamerCancellation cancellation)	Realizes the stream and returns an interface that lets one consume it.
default @NonNull CompletionStage<Void>	subscribe(@NonNull StreamSink<? super @NonNull T> consumer)	Relays the events of the upstream into a StreamSink consumer via the help of the standard Executors.newVirtualThreadPerTaskExecutor() as a mediator for pull-to-push.
default @NonNull CompletionStage<Void>	subscribe(@NonNull StreamSink<? super @NonNull T> consumer, ExecutorService executor)	Relays the events of the upstream into a StreamSink consumer via the help of the given ExecutorService as a mediator for pull-to-push.
default void	subscribe(Flow.Subscriber<? super @NonNull T> subscriber)	Consume this Streamable via the given flow-reactive-streams subscriber on a virtual thread backed executor service.
default void	subscribe(Flow.Subscriber<? super @NonNull T> subscriber, @NonNull ExecutorService executor)	Consume this Streamable via the given flow-reactive-streams subscriber.
default @NonNull Streamable<T>	take(long count)	Takes at most the given number of items from the upstream and relays it to the downstream, then cancels the rest of the sequence.
default <U> @NonNull Streamable<T>	takeUntil(@NonNull Streamable<U> other)	Relays items from this Streamable until the other Streamable signals an item or completes.
default @NonNull Streamable<T>	takeWhile(@NonNull Predicate<? super @NonNull T> predicate)	Relays items from this Streamable while the predicate returns true
default @NonNull TestSubscriber<T>	test()	Creates a new TestSubscriber and subscribes it to this Streamable.
default @NonNull TestSubscriber<T>	test(@NonNull ExecutorService executor)	Creates a new TestSubscriber and subscribes it to this Streamable.
default @NonNull Streamable<T>	timeout(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Streamable<@NonNull T> fallback)	Applies a timeout to each upstream item and switches to the fallback Streamable if the upstream doesn't produce an item within the given timeout period.
static @NonNull Streamable<Long>	timer(long delay, TimeUnit unit, Scheduler scheduler)	Signals a single 0L and completes after the given delay amount of time.
static @NonNull Streamable<Long>	timer(long delay, TimeUnit unit, ExecutorService executor)	Signals a single 0L and completes after the given delay amount of time.
default <@NonNull R> R	to(@NonNull StreamableConverter<@NonNull T, ? extends @NonNull R> converter)	Calls the specified converter function during assembly time and returns its resulting value.
default @NonNull Flowable<T>	toFlowable()	Converts the streamable into a Flowable representation, running on the default Executors.newVirtualThreadPerTaskExecutor() virtual thread.
default @NonNull Flowable<T>	toFlowable(@NonNull ExecutorService executor)	Converts this Streamable into a Flowable representation, running on the provided executor service.
default @NonNull Observable<T>	toObservable()	Converts this Streamable into a Observable representation, emitting items on whatever thread produces items in the current Streamable.
default <@NonNull R> @NonNull Streamable<R>	transform(@NonNull VirtualTransformer<@NonNull T, @NonNull R> transformer)	Transforms the upstream sequence into zero or more elements for the downstream.
default <@NonNull R> @NonNull Streamable<R>	transform(@NonNull VirtualTransformer<@NonNull T, @NonNull R> transformer, @NonNull ExecutorService executor)	Transforms the upstream sequence into zero or more elements for the downstream.
static <T,R> @NonNull Streamable<T>	using(Supplier<? extends R> resourceSupplier, Function<? super R, ? extends Streamable<? extends T>> resourceMapper, Consumer<? super R> resourceCleaner)	For each incoming streamer, this operator creates a resource, then uses that resource to create the actual Streamable instance to stream value of and then uses a cleaner callback to dissolve the resource once the Streamable terminated.
static <T> @NonNull Streamable<List<T>>	zip(Iterable<? extends Streamable<? extends T>> sources)	Takes the next element from each source Streamable and emits them a a single row of List.

Method Details

stream

@CheckReturnValue
@NonNull
@NonNull Streamer<T> stream(@NonNull
 @NonNull StreamerCancellation cancellation)

Realizes the stream and returns an interface that lets one consume it.

Implementations are not meant to dispose the cancellation as it is the privilege of whatever calls the stream method. If you need to talk to other Streamables, use the StreamerCancellation.derive() to gain access to a fully fledged DisposableStreamerCancellation where you can dispose it without disposing the parent cancellation instance by accident.

It is recommened you StreamerCancellation.remove(Disposable) or StreamerCancellation.delete(Disposable) any extra resources you added via StreamerCancellation.add(Disposable) when your operator reaches the Streamer.finish() call.

Parameters:

cancellation - where to register, check and listen for cancellation calls via its StreamerCancellation.isDisposed()

Returns:

the Streamer instance to consume.

concat

@CheckReturnValue
@NonNull
static <@NonNull T> @NonNull Streamable<T> concat(Streamable<? extends Streamable<? extends @NonNull T>> sources,
 ExecutorService executor)

Emits the elements of each inner sequence produced by the outer sequence.

Type Parameters:

T - the common element type

Parameters:

sources - a streamable of inner streamables

executor - the executorservice where to run the virtual wait

Returns:

the new Streamable instance.

Throws:

NullPointerException - if sources or executor is null

concat

@CheckReturnValue
@NonNull
static <@NonNull T> @NonNull Streamable<T> concat(Iterable<? extends Streamable<? extends @NonNull T>> sources)

Streams the Streamables one after the other from the given iterable sequence of Streamables.

Type Parameters:

T - the element type of the inner and resulting Streamables

Parameters:

sources - the iterable sequence of Streamables.

Returns:

the new Streamable source

Throws:

NullPointerException - if sources is null

create

@CheckReturnValue
@NonNull
static <@NonNull T> @NonNull Streamable<T> create(@NonNull
 @NonNull VirtualGenerator<@NonNull T> generator)

Generate a sequence of values via a virtual generator callback (yielder) which is free to block and is natively backpressured.

Runs on the Schedulers.virtual() scheduler.

Example

Streamable.create(emitter -> {
    emitter.emit(1);
    emitter.emit(2);
    emitter.emit(3);
})
.forEach(System.out::println)
;

Type Parameters:

T - the element type

Parameters:

generator - the generator to use

Returns:

the streamable instance

create

@CheckReturnValue
@NonNull
static <@NonNull T> @NonNull Streamable<T> create(@NonNull
 @NonNull VirtualGenerator<@NonNull T> generator,
 @NonNull
 @NonNull Scheduler scheduler)

Generate a sequence of values via a virtual generator callback (yielder) which is free to block and is natively backpressured.

Runs on the given scheduler.

Example

Streamable.create(emitter -> {
    emitter.emit(1);
    emitter.emit(2);
    emitter.emit(3);
}, Schedulers.cached())
.forEach(System.out::println)
;

Type Parameters:

T - the element type

Parameters:

generator - the generator to use

scheduler - the scheduler to run the virtual generator on

Returns:

the new Streamable instance

Throws:

NullPointerException - if generator or scheduler is null

create

@CheckReturnValue
@NonNull
static <@NonNull T> @NonNull Streamable<T> create(@NonNull
 @NonNull VirtualGenerator<@NonNull T> generator,
 @NonNull
 @NonNull ExecutorService executor)

Generate a sequence of values via a virtual generator callback (yielder) which is free to block and is natively backpressured.

Runs on the given executor service.

Type Parameters:

T - the element type

Parameters:

generator - the generator to use

executor - the executor to run the virtual generator on

Returns:

the streamable instance

defer

@CheckReturnValue
@NonNull
static <@NonNull T> @NonNull Streamable<T> defer(Supplier<? extends Streamable<? extends @NonNull T>> supplier)

Defers the creation of the actual Streamable, allowing a per streamer state to be created along with it.

Type Parameters:

T - the element type of the Streamable

Parameters:

supplier - the callback that returns the actual Streamable to be streamed

Returns:

the new Streamable instance

Throws:

NullPointerException - if supplier is null

error

@CheckReturnValue
@NonNull
static <@NonNull T> @NonNull Streamable<T> error(Throwable throwable)

Creates a Streamable that signals the given Throwable immediately when it begins streaming, ending the sequence.

Type Parameters:

T - the element type of the sequence

Parameters:

throwable - the Throwable to signal immediately upon streaming

Returns:

the new Streamable instance

Throws:

NullPointerException - if throwable is null

empty

@CheckReturnValue
@NonNull
static <@NonNull T> @NonNull Streamable<T> empty()

Returns an empty Streamable that never produces an item and just completes.

Type Parameters:

T - the element type

Returns:

the Streamable instance

filter

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> filter(@NonNull
 @NonNull Predicate<? super @NonNull T> predicate)

Filters out the upstream items that do not pass the given predicate.

Parameters:

predicate - the callback that should return true to let the upstream value pass or false to ignore it and continue with the next upstream item

Returns:

the new Streamable instance

Throws:

NullPointerException - if predicate is null

fromArray

@SafeVarargs
@CheckReturnValue
@NonNull
static <@NonNull T> @NonNull Streamable<T> fromArray(@NonNull
 @NonNull T... items)

Streams all elements of the given items array.

Type Parameters:

T - the element type of the items

Parameters:

items - the array of items to stream

Returns:

the new Streamable instance

Throws:

NullPointerException - if items is null

fromCompletable

static <@NonNull T> Streamable<T> fromCompletable(@NonNull
 @NonNull CompletableSource source)

Convert a CompletableSource into a Streamable and relay its terminal events.

The resulting Streamable will never produce any items.

Type Parameters:

T - the target type of the sequence

Parameters:

source - the source CompletableSource to convert

Returns:

the new Streamable instance

Throws:

NullPointerException - if source is null

fromIterable

@CheckReturnValue
@NonNull
static <@NonNull T> @NonNull Streamable<T> fromIterable(@NonNull
 @NonNull Iterable<? extends @NonNull T> items)

Streams all elements of the given Iterable sequence.

Type Parameters:

T - the element type of the items

Parameters:

items - the iterable of items to stream

Returns:

the new Streamable instance

Throws:

NullPointerException - if items is null

fromMaybe

static <@NonNull T> Streamable<T> fromMaybe(@NonNull
 @NonNull MaybeSource<@NonNull T> source)

Convert a MaybeSource into a Streamable and relay its terminal events.

The resulting Streamable will never produce any items.

Type Parameters:

T - the target type of the sequence

Parameters:

source - the source MaybeSource to convert

Returns:

the new Streamable instance

Throws:

NullPointerException - if source is null

fromPublisher

@CheckReturnValue
@NonNull
static <T> @NonNull Streamable<T> fromPublisher(@NonNull
 Flow.Publisher<T> source)

Convert any Flow.Publisher into a Streamable sequence.

Type Parameters:

T - the element type

Parameters:

source - Flow.Publisher to convert

Returns:

the new Streamable instance

Throws:

NullPointerException - if source is null

fromPublisher

@CheckReturnValue
@NonNull
static <T> @NonNull Streamable<T> fromPublisher(@NonNull
 Flow.Publisher<T> source,
 @NonNull
 @NonNull ExecutorService executor)

Convert any Flow.Publisher into a Streamable sequence.

Type Parameters:

T - the element type

Parameters:

source - Flow.Publisher to convert

executor - where the conversion will run

Returns:

the new Streamable instance

Throws:

NullPointerException - if source or executor is null

fromSingle

static <@NonNull T> Streamable<T> fromSingle(@NonNull
 @NonNull SingleSource<@NonNull T> source)

Convert a SingleSource into a Streamable and relay its terminal events.

The resulting Streamable will never produce any items.

Type Parameters:

T - the target type of the sequence

Parameters:

source - the source SingleSource to convert

Returns:

the new Streamable instance

Throws:

NullPointerException - if source is null

fromStream

@CheckReturnValue
@NonNull
static <@NonNull T> @NonNull Streamable<T> fromStream(@NonNull
 @NonNull Stream<? extends @NonNull T> items)

Streams all elements of the given Stream sequence.

Type Parameters:

T - the element type of the items

Parameters:

items - the stream of items to stream

Returns:

the new Streamable instance

Throws:

NullPointerException - if items is null

intervalRange

static Streamable<Long> intervalRange(long start,
 long count,
 long initialDelay,
 long period,
 TimeUnit unit,
 Scheduler scheduler)

Constructs a Streamable that after the initial delay, starts emitting an ever increasing numbers from start up to start + count exclusive with the given period.

If there are processing delays, this source may emit multiple queued up items in a quick succession.

Parameters:

start - the first long value to emit

count - the number of items to emit, use Long.MAX_VALUE for an unlimited range

initialDelay - the time to delay before the start item is emitted

period - the period of how often emit the next item

unit - the time unit for both initialDelay and period

scheduler - the scheduler to use for the timed waiting

Returns:

the new Streamable instance

Throws:

NullPointerException - if unit or scheduler is null

IllegalArgumentException - if count is negative

intervalRange

static Streamable<Long> intervalRange(long start,
 long count,
 long initialDelay,
 long period,
 TimeUnit unit,
 ExecutorService executor)

Constructs a Streamable that after the initial delay, starts emitting an ever increasing numbers from start up to start + count exclusive with the given period.

If the provided ExecutorService is a ScheduledExecutorService, its ScheduledExecutorService.scheduleAtFixedRate(Runnable, long, long, TimeUnit) will be used. Otherwise, a plain ExecutorService will be wrapped via Schedulers.from(Executor).

If there are processing delays, this source may emit multiple queued up items in a quick succession.

Parameters:

start - the first long value to emit

count - the number of items to emit, use Long.MAX_VALUE for an unlimited range

initialDelay - the time to delay before the start item is emitted

period - the period of how often emit the next itme

unit - the time unit for both initialDelay and period

executor - the executor to use

Returns:

the new Streamable instance

Throws:

NullPointerException - if unit or executor is null

IllegalArgumentException - if count is negative

just

@CheckReturnValue
@NonNull
static <@NonNull T> @NonNull Streamable<T> just(@NonNull
 @NonNull T item)

Returns a single-element Streamable that produces the constant item and completes.

Type Parameters:

T - the element type

Parameters:

item - the constant item to produce

Returns:

the Streamable instance

Throws:

NullPointerException - if item is null

never

@CheckReturnValue
@NonNull
static <@NonNull T> @NonNull Streamable<T> never()

Returns an Streamable that never produces an item and never terminates.

Type Parameters:

T - the element type

Returns:

the Streamable instance

range

@CheckReturnValue
@NonNull
static @NonNull Streamable<Integer> range(int start,
 int count)

Emits elements from start up to start + count exclusive.

Parameters:

start - the start element

count - the number of elements to emit

Returns:

the new Streamable instance

Throws:

IllegalArgumentException - if count is negative

rangeLong

@CheckReturnValue
@NonNull
static @NonNull Streamable<Long> rangeLong(long start,
 long count)

Emits elements from start up to start + count exclusive.

Parameters:

start - the start element

count - the number of elements to emit

Returns:

the new Streamable instance

Throws:

IllegalArgumentException - if count is negative

timer

@CheckReturnValue
@NonNull
static @NonNull Streamable<Long> timer(long delay,
 TimeUnit unit,
 Scheduler scheduler)

Signals a single 0L and completes after the given delay amount of time.

Parameters:

delay - the amount to delay the signaling of a single item

unit - the time unit

scheduler - where the timed delay should happen

Returns:

the new Streamable instance

Throws:

NullPointerException - if unit or scheduler is null

timer

@CheckReturnValue
@NonNull
static @NonNull Streamable<Long> timer(long delay,
 TimeUnit unit,
 ExecutorService executor)

Signals a single 0L and completes after the given delay amount of time.

If the executor is a ScheduledExecutorService, the operator will use its ScheduledExecutorService.schedule(Runnable, long, TimeUnit) method. Otherwise, the ExecutorService.submit(Callable) will be invoked with an upfront TimeUnit.sleep(long).

Parameters:

delay - the amount to delay the signaling of a single item

unit - the time unit

executor - where the timed delay should happen

Returns:

the new Streamable instance

Throws:

NullPointerException - if unit or executor is null

using

@CheckReturnValue
@NonNull
static <T,R> @NonNull Streamable<T> using(Supplier<? extends R> resourceSupplier,
 Function<? super R, ? extends Streamable<? extends T>> resourceMapper,
 Consumer<? super R> resourceCleaner)

For each incoming streamer, this operator creates a resource, then uses that resource to create the actual Streamable instance to stream value of and then uses a cleaner callback to dissolve the resource once the Streamable terminated.

Type Parameters:

T - the element type of the sequence

R - the resource type

Parameters:

resourceSupplier - supplies a resource object per stream(StreamerCancellation) call

resourceMapper - maps the supplied resource into a Streamable source

resourceCleaner - cleans up the supplied resource

Returns:

the new Streamable instance

Throws:

NullPointerException - if resourceSupplier or resourceMapper or resourceCleaner is null

zip

@CheckReturnValue
@NonNull
static <T> @NonNull Streamable<List<T>> zip(Iterable<? extends Streamable<? extends T>> sources)

Takes the next element from each source Streamable and emits them a a single row of List.

If any of the sources is shorter than the rest or any of them fails, the sequence is completed early normally or with an exception, respectively.

Type Parameters:

T - the common element type of the sequences

Parameters:

sources - the Iterable sequence of the source Streamables

Returns:

the new Streamable instance

Throws:

NullPointerException - if sources is null

blockingFirst

@CheckReturnValue
@NonNull
default T blockingFirst()

Blocks the current thread until this Streamable produces one item, which is then returned.

Returns:

the first item of this Streamable

Throws:

NoSuchElementException - if the this Streamable is empty

CancellationException - if this Streamable failed with a checked exception

RuntimeException - if this Streamable failed with an unchecked exception

blockingLast

@CheckReturnValue
@NonNull
default T blockingLast()

Blocks the current thread until this Streamable produces all of its items and the very last is then returned.

Returns:

the very last item of this Streamable

Throws:

NoSuchElementException - if the this Streamable is empty

CancellationException - if this Streamable failed with a checked exception

RuntimeException - if this Streamable failed with an unchecked exception

collect

@CheckReturnValue
@NonNull
default <A,R> @NonNull Streamable<R> collect(@NonNull
 @NonNull Collector<@NonNull T, A, R> collector)

Collects all upstream values via the use of a Collector configuration and emits its resulting value as a single item of the returned Streamable.

See Collectors for the most typical collector standard implementations.

Type Parameters:

A - the accumulator type of the collector

R - the result type of the collector and the returned Streamamble

Parameters:

collector - the Java collector instance to use

Returns:

the new Streamable instance

Throws:

NullPointerException - if collector is null

delay

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> delay(long time,
 @NonNull
 @NonNull TimeUnit unit,
 @NonNull
 @NonNull Scheduler scheduler)

Delays the delivery of each upstream item by the given time amount.

Parameters:

time - the delay time

unit - the time unit

scheduler - the scheduler where the timed wait is happening

Returns:

the new Streamable instance

Throws:

NullPointerException - if unit or scheduler is null

doOnError

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> doOnError(@NonNull
 @NonNull Consumer<? super Throwable> consumer)

Calls the specific Consumer if there is an error from the upstream.

Parameters:

consumer - the consumer to call with the Throwable

Returns:

the new Streamable instance

Throws:

NullPointerException - if consumer is null

doOnNext

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> doOnNext(@NonNull
 @NonNull Consumer<? super @NonNull T> consumer)

Calls the given consumer whenever an upstream item becomes available.

Parameters:

consumer - the callback to invoke with the next item from upstream

Returns:

the new Streamable instance

Throws:

NullPointerException - if consumer is null

flatMap

@CheckReturnValue
@NonNull
default <R> @NonNull Streamable<R> flatMap(@NonNull
 @NonNull Function<? super @NonNull T, ? extends Streamable<? extends R>> mapper,
 @NonNull
 @NonNull StandardConcurrentConfig config)

Maps each upstream item onto a Streamable and runs them concurrently while relaying inner items as first-come-first-served manner.

Type Parameters:

R - the element type of the output sequence

Parameters:

mapper - the function that turns an upstream item into a Streamable inner sequence

config - the configuration record for this operator

Returns:

the new Streamable instance

Throws:

NullPointerException - if mapper or config is null

groupBy

@CheckReturnValue
@NonNull
default <@Nullable K>
@NonNull Streamable<GroupedStreamable<K,T>> groupBy(@NonNull
 @NonNull Function<? super @NonNull T, ? extends @Nullable K> keySelector)

Maps each upstream item into a GroupedStreamable group, emits those groups and keeps relaying the upstream items into those groups.

Type Parameters:

K - the key type, nulls allowed

Parameters:

keySelector - the function that receives the upstream item and returns a key that determines which group the item will go into

Returns:

the new Streamable instance

Throws:

NullPointerException - if keySelector is null

hide

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> hide()

Hides the identity of this Streamable and its Streamer.

Use it to break optimizations or hide concrete implementations.

Returns:

the new Streamable instance

ignoreElements

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> ignoreElements()

Ignores all elements from the current Streamable and completes.

Returns:

the new Streamable instance

intercept

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> intercept(StreamableInterceptConfig<@NonNull T> config)

Intercepts the lifecycle method calls of Streamable and Streamer and allows the modification of them via Function callbacks.

Parameters:

config - the configuration record for this operator

Returns:

the new Streamable instance

Throws:

NullPointerException - if config is null

last

@CheckReturnValue
@NonNull
default @NonNull Single<T> last(@NonNull
 @NonNull T defaultItem)

Signals the last item from the upstream Streamable or the provided defaultItem if the upstream is empty, as a Single instance.

Parameters:

defaultItem - the item to signal if the upstream turns out to be empty

Returns:

the new Single instance

Throws:

NullPointerException - if defaultItem is null

lastOrError

@CheckReturnValue
@NonNull
default @NonNull Single<T> lastOrError()

Signals the last item from the upstream Streamable or a NoSuchElementException if the upstream is empty, as a Single instance.

Returns:

the new Single instance

lift

@CheckReturnValue
@NonNull
default <@NonNull R> @NonNull Streamable<R> lift(@NonNull
 @NonNull StreamableOperator<? super @NonNull T, ? extends @NonNull R> lifter)

This method requires advanced knowledge about building operators, please consider other standard composition methods first; Returns a Streamable instance which when its stream(StreamerCancellation) is invoked, applies the specified operator callback to the upstream Streamer to produce an actual Streamer instance to be handed downstream.

Use it to implement operators without creating the surrounding Streamable class.

If the lifter returns null or throws, the downstream will receive a standard error streamer.

Type Parameters:

R - the downstream type of the sequence

Parameters:

lifter - the callback that will be invoked with the upstream Streamer and is expected to produce a Streamer for the downstream.

Returns:

the new Streamable instance

Throws:

NullPointerException - if lifter is null

map

@CheckReturnValue
@NonNull
default <@NonNull R> @NonNull Streamable<R> map(@NonNull
 @NonNull Function<? super @NonNull T, ? extends @NonNull R> mapper)

Maps each upstream item into another item via a mapper function.

Type Parameters:

R - the element type of the mapping

Parameters:

mapper - the function that takes an upstream item and returns an item to be emitted to the downstream

Returns:

the new Streamable instance

Throws:

NullPointerException - if mapper is null

mapOptional

@CheckReturnValue
@NonNull
default <@NonNull R> @NonNull Streamable<R> mapOptional(@NonNull
 @NonNull Function<? super @NonNull T, ? extends Optional<? extends @NonNull R>> mapper)

Maps each upstream item into another, optional item via a mapper function that skips the empty optionals.

Type Parameters:

R - the element type of the mapping

Parameters:

mapper - the function that takes an upstream item and returns an optional item to be emitted / skipped to the downstream

Returns:

the new Streamable instance

Throws:

NullPointerException - if mapper is null

onErrorResumeNext

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> onErrorResumeNext(@NonNull
 @NonNull Function<? super Throwable, ? extends Streamable<? extends @NonNull T>> fallbackMapper)

When the upstream fails, the sequence is resumed by the Streamable that is returned for the failure Throwable.

Parameters:

fallbackMapper - the function that receives the upstream error and should return a Streamable to resume the sequence with

Returns:

the new Streamable instance

Throws:

NullPointerException - if fallbackMapper is null

repeat

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> repeat(long count)

Runs the upstream at most the given count times while it succeeds.

So a repeat(1) will try to consume the upstream once.

Parameters:

count - the number of retries if the upstream fails

Returns:

the new Streamable instance

Throws:

IllegalArgumentException - if count is negative

repeatWhen

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> repeatWhen(Function<? super Long, ? extends CompletionStage<Boolean>> whenFunction)

Repeats the upstream when the given function signals true via the CompletionStage for the count how many times the upstream was streamed.

The first repeat run will present 0 for the function.

Parameters:

whenFunction - the function to call with the run index, it should signal true to repeat the source, false to complete without error or complete exceptionally to become the failure result of the sequence.

Returns:

the new Streamable instance

Throws:

NullPointerException - if whenFunction is null

retry

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> retry(long count)

Retries at most the given count times the upstream if it fails with any error.

So a retry(1) will try to consume the upstream twice.

Parameters:

count - the number of retries if the upstream fails

Returns:

the new Streamable instance

Throws:

IllegalArgumentException - if count is negative

retry

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> retry(Predicate<? super Throwable> predicate)

Retries the upstream if the given predicate returns true for the failure Throwable of the last streaming of the upstream.

Parameters:

predicate - the p

Returns:

the new Streamable instance

Throws:

NullPointerException - if predicate is null

retry

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> retry(BiPredicate<? super Long, ? super Throwable> predicate)

Retries the upstream if the given predicate returns true for the failure count and Throwable of the last streaming of the upstream.

The first failure run will present 0 for the predicate.

Parameters:

predicate - the function to call with the failure index and Throwable

Returns:

the new Streamable instance

Throws:

NullPointerException - if predicate is null

retryWhen

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> retryWhen(BiFunction<? super Long, ? super Throwable, ? extends CompletionStage<Boolean>> whenFunction)

Retries the upstream when the given predicate signals true via the CompletionStage for the failure count and Throwable of the last streaming of the upstream.

The first failure run will present 0 for the predicate.

Parameters:

whenFunction - the function to call with the failure index and Throwable, it should signal true to retry the source, false to complete without error or complete exceptionally to become the failure result of the sequence.

Returns:

the new Streamable instance

Throws:

NullPointerException - if whenFunction is null

skip

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> skip(long count)

Skips the first count items and relays the rest to the downstream.

Parameters:

count - the number of items to skip

Returns:

the new Streamable instance

Throws:

IllegalArgumentException - if count is negative

take

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> take(long count)

Takes at most the given number of items from the upstream and relays it to the downstream, then cancels the rest of the sequence.

Note that cancellation of the upstream happens when the downstream calls Streamer.next() because unlike the push-based take implementations, the current upstream value has to remain accessible until the downstream calls next or Streamer.finish().

Parameters:

count - the maximum number of items to relay

Returns:

the new Streamable instance

Throws:

IllegalArgumentException - if count is negative

takeUntil

@CheckReturnValue
@NonNull
default <U> @NonNull Streamable<T> takeUntil(@NonNull
 @NonNull Streamable<U> other)

Relays items from this Streamable until the other Streamable signals an item or completes.

Type Parameters:

U - the element type of the other Streamable

Parameters:

other - the Streamable expected to signal when to stop taking items from this Streamable

Returns:

the new Streamable instance

Throws:

NullPointerException - if other is null

takeWhile

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> takeWhile(@NonNull
 @NonNull Predicate<? super @NonNull T> predicate)

Relays items from this Streamable while the predicate returns true

Parameters:

predicate - the predicate to test if the sequence should keep going

Returns:

the new Streamable instance

Throws:

NullPointerException - if predicate is null

timeout

@CheckReturnValue
@NonNull
default @NonNull Streamable<T> timeout(long timeout,
 @NonNull
 @NonNull TimeUnit unit,
 @NonNull
 @NonNull Scheduler scheduler,
 @NonNull
 @NonNull Streamable<@NonNull T> fallback)

Applies a timeout to each upstream item and switches to the fallback Streamable if the upstream doesn't produce an item within the given timeout period.

Parameters:

timeout - the time to wait for each upstream item

unit - the time unit

scheduler - the scheduler where to wait for the next upstream item

fallback - the Streamable to switch to if the upstream doesn't produce an item within the time limit

Returns:

the new Streamable instance

Throws:

NullPointerException - if unit or scheduler or fallback is null

to

@CheckReturnValue
@NonNull
default <@NonNull R> R to(@NonNull
 @NonNull StreamableConverter<@NonNull T, ? extends @NonNull R> converter)

Calls the specified converter function during assembly time and returns its resulting value.

This allows fluent conversion to any other type.

Type Parameters:

R - the resulting object type

Parameters:

converter - the function that receives the current Observable instance and returns a value

Returns:

the converted value

Throws:

NullPointerException - if converter is null

toFlowable

@CheckReturnValue
@NonNull
default @NonNull Flowable<T> toFlowable()

Converts the streamable into a Flowable representation, running on the default Executors.newVirtualThreadPerTaskExecutor() virtual thread.

Returns:

the new Flowable instance

toFlowable

@CheckReturnValue
@NonNull
default @NonNull Flowable<T> toFlowable(@NonNull
 @NonNull ExecutorService executor)

Converts this Streamable into a Flowable representation, running on the provided executor service.

Parameters:

executor - the executor to use

Returns:

the new Flowable instance

Throws:

NullPointerException - if executor is null

toObservable

@CheckReturnValue
@NonNull
default @NonNull Observable<T> toObservable()

Converts this Streamable into a Observable representation, emitting items on whatever thread produces items in the current Streamable.

Unlike toFlowable(ExecutorService), the lack of backpressure doesn't require any blocking thus any need to run the conversion on any particular Scheduler or ExecutorService on its own.

Returns:

the new Observable instance

transform

@CheckReturnValue
@NonNull
default <@NonNull R> @NonNull Streamable<R> transform(@NonNull
 @NonNull VirtualTransformer<@NonNull T, @NonNull R> transformer)

Transforms the upstream sequence into zero or more elements for the downstream.

Type Parameters:

R - the result element type

Parameters:

transformer - the interface to implement the transforming logic

Returns:

the new Streamable instance

transform

@CheckReturnValue
@NonNull
default <@NonNull R> @NonNull Streamable<R> transform(@NonNull
 @NonNull VirtualTransformer<@NonNull T, @NonNull R> transformer,
 @NonNull
 @NonNull ExecutorService executor)

Transforms the upstream sequence into zero or more elements for the downstream.

Type Parameters:

R - the result element type

Parameters:

transformer - the interface to implement the transforming logic

executor - where to run the transform and blocking operations

Returns:

the new Streamable instance

Throws:

NullPointerException - if transformer or executor is null

forEach

@CheckReturnValue
@NonNull
default @NonNull CompletionStageDisposable<Void> forEach(@NonNull
 @NonNull Consumer<? super @NonNull T> consumer)

Consumes elements from this Streamable via the provided executor service.

Parameters:

consumer - the callback that gets the elements until completion

Returns:

a Disposable that lets one cancel the sequence asynchronously.

forEach

@CheckReturnValue
@NonNull
default @NonNull CompletionStageDisposable<Void> forEach(@NonNull
 @NonNull Consumer<? super @NonNull T> consumer,
 @NonNull
 @NonNull DisposableStreamerCancellation canceller)

Consumes elements from this Streamable via the provided executor service.

Parameters:

consumer - the callback that gets the elements until completion

canceller - the container to trigger cancellation of the sequence

Returns:

the CompletionStage that gets notified when the sequence ends

forEach

@CheckReturnValue
@NonNull
default @NonNull CompletionStageDisposable<Void> forEach(@NonNull
 @NonNull Consumer<? super @NonNull T> consumer,
 @NonNull
 @NonNull ExecutorService executor)

Consumes elements from this Streamable via the provided executor service.

Parameters:

consumer - the callback that gets the elements until completion

executor - the service that hosts the blocking waits.

Returns:

a Disposable that lets one cancel the sequence asynchronously.

forEach

@CheckReturnValue
@NonNull
default @NonNull CompletionStageDisposable<Void> forEach(@NonNull
 @NonNull Consumer<? super @NonNull T> consumer,
 @NonNull
 @NonNull DisposableStreamerCancellation canceller,
 @NonNull
 @NonNull ExecutorService executor)

Consumes elements from this Streamable via the provided executor service.

Parameters:

consumer - the callback that gets the elements until completion

canceller - the container to trigger cancellation of the sequence

executor - the service that hosts the blocking waits.

Returns:

the new CompletionStageDisposable that gets notified when the sequence ends

Throws:

NullPointerException - if consumer or canceller or executor is null

forEach

@CheckReturnValue
@NonNull
default @NonNull CompletionStageDisposable<Void> forEach(@NonNull
 @NonNull BiConsumer<? super @NonNull T, ? super Disposable> consumer,
 @NonNull
 @NonNull DisposableStreamerCancellation canceller,
 @NonNull
 @NonNull ExecutorService executor)

Consumes elements from this Streamable via the provided executor service.

Parameters:

consumer - the callback that gets the elements until completion

canceller - the container to trigger cancellation of the sequence

executor - the service that hosts the blocking waits.

Returns:

the CompletionStage that gets notified when the sequence ends

Throws:

NullPointerException - if consumer or canceller or executor is null

subscribe

default void subscribe(@NonNull
 Flow.Subscriber<? super @NonNull T> subscriber,
 @NonNull
 @NonNull ExecutorService executor)

Consume this Streamable via the given flow-reactive-streams subscriber.

Parameters:

subscriber - the subscriber to consume with.

executor - the service that hosts the blocking waits.

subscribe

default void subscribe(@NonNull
 Flow.Subscriber<? super @NonNull T> subscriber)

Consume this Streamable via the given flow-reactive-streams subscriber on a virtual thread backed executor service.

Parameters:

subscriber - the subscriber to consume with.

subscribe

@NonNull
default @NonNull CompletionStage<Void> subscribe(@NonNull
 @NonNull StreamSink<? super @NonNull T> consumer)

Relays the events of the upstream into a StreamSink consumer via the help of the standard Executors.newVirtualThreadPerTaskExecutor() as a mediator for pull-to-push.

Parameters:

consumer - the consumer to relay events into

Returns:

the stage that gets completed normally or with an exception when this Streamable terminates

Throws:

NullPointerException - if consumer is null

subscribe

@NonNull
default @NonNull CompletionStage<Void> subscribe(@NonNull
 @NonNull StreamSink<? super @NonNull T> consumer,
 ExecutorService executor)

Relays the events of the upstream into a StreamSink consumer via the help of the given ExecutorService as a mediator for pull-to-push.

Parameters:

consumer - the consumer to relay events into

executor - the ExecutorService to run the blocking consume and emissions

Returns:

the stage that gets completed normally or with an exception when this Streamable terminates

Throws:

NullPointerException - if consumer or executor is null

test

@CheckReturnValue
@NonNull
default @NonNull TestSubscriber<T> test()

Creates a new TestSubscriber and subscribes it to this Streamable.

Returns:

the created test subscriber

test

@CheckReturnValue
@NonNull
default @NonNull TestSubscriber<T> test(@NonNull
 @NonNull ExecutorService executor)

Creates a new TestSubscriber and subscribes it to this Streamable.

Parameters:

executor - the executor to use

Returns:

the created test subscriber