4.x: Streamable operator checklist

Author: akarnokdCreated Aug 27, 2026Updated Aug 27, 2026

LabelsEnhancement4.xStreamableAI

Via Claude Fable 5 High

`Streamable`: missing `Flowable` operators and overloads (4.x parity checklist)

This is a mechanical comparison of the public API of Flowable against Streamable on the 4.x branch (as of commit bc5d8b8). Flowable currently exposes 464 public methods/overloads (206 distinct names); Streamable exposes 75 (55 distinct names).

Signatures below are written with the Streamable equivalents of the Flowable types substituted in:

`Flowable` type	`Streamable` equivalent used below
`Flowable<T>` / `Publisher<T>` (parameters, return types)	`Streamable<T>`
`GroupedFlowable<K, V>`	`GroupedStreamable<K, V>`
`FlowableOperator<R, T>`	`StreamableOperator<T, R>`
`FlowableConverter<T, R>`	`StreamableConverter<T, R>`
`FlowableTransformer<T, R>`	`StreamableTransformer<T, R>` (type does not exist yet)
`ConnectableFlowable<T>`	`ConnectableStreamable<T>` (type does not exist yet)
`FlowableEmitter<T>`	`VirtualEmitter<T>`
`Subscriber<T>` / `FlowableSubscriber<T>`	`Flow.Subscriber<T>`

Things to keep in mind while working through the list:

Flowable's Disposable-returning subscribe(...) overloads probably map to CompletionStageDisposable<Void> on Streamable (as forEach already does).
Overloads that take a Scheduler may want an additional ExecutorService twin, following the existing Streamable.timer / intervalRange / create pattern. Those twins are not listed here because they have no Flowable counterpart.
A few methods already exist on Streamable with a different shape (e.g. flatMap(mapper, StandardConcurrentConfig) vs. Flowable's StandardConcurrentBufferedConfig, retry(BiPredicate<Long, Throwable>) vs. BiPredicate<Integer, Throwable>, repeatWhen/retryWhen taking CompletionStage<Boolean> functions, create(VirtualGenerator) vs. virtualCreate). These are listed in the missing overloads section only where the Flowable overload has no exact-arity/equivalent-type match; adjust or tick them off as appropriate.
Return-type differences on otherwise matching signatures are not listed (e.g. ignoreElements() returns Completable on Flowable but Streamable<T> on Streamable; collect(Collector) returns Single<R> vs. Streamable<R>; forEach returns Disposable vs. CompletionStageDisposable<Void>).

Methods that exist on `Streamable` but are missing overloads (23 methods, 69 overloads)

`blockingFirst`

T blockingFirst(T defaultItem)

`blockingLast`

T blockingLast(T defaultItem)

`collect`

 Single collect(Supplier<? extends U> initialItemSupplier, BiConsumer<? super U, ? super T> collector)

`concat`

static <T> Streamable<T> concat(Streamable<? extends Streamable<? extends T>> sources)
static <T> Streamable<T> concat(Iterable<? extends Streamable<? extends T>> sources, StandardBufferedConfig config)
static <T> Streamable<T> concat(Streamable<? extends Streamable<? extends T>> sources, StandardBufferedConfig config)

`delay`

 Streamable<T> delay(Function<? super T, ? extends Streamable> itemDelayIndicator)
Streamable<T> delay(long time, TimeUnit unit)
<U, V> Streamable<T> delay(Streamable subscriptionIndicator, Function<? super T, ? extends Streamable<V>> itemDelayIndicator)
Streamable<T> delay(long time, TimeUnit unit, Scheduler scheduler, boolean delayError)

`error`

static <T> Streamable<T> error(Supplier<? extends Throwable> supplier)

`flatMap`

<R> Streamable<R> flatMap(Function<? super T, ? extends Streamable<? extends R>> mapper)
<R> Streamable<R> flatMap(Function<? super T, ? extends Streamable<? extends R>> mapper, StandardConcurrentBufferedConfig config)
<U, R> Streamable<R> flatMap(Function<? super T, ? extends Streamable<? extends U>> mapper, BiFunction<? super T, ? super U, ? extends R> combiner)
<R> Streamable<R> flatMap(Function<? super T, ? extends Streamable<? extends R>> onNextMapper, Function<? super Throwable, ? extends Streamable<? extends R>> onErrorMapper, Supplier<? extends Streamable<? extends R>> onCompleteSupplier)
<U, R> Streamable<R> flatMap(Function<? super T, ? extends Streamable<? extends U>> mapper, BiFunction<? super T, ? super U, ? extends R> combiner, StandardConcurrentBufferedConfig config)
<R> Streamable<R> flatMap(Function<? super T, ? extends Streamable<? extends R>> onNextMapper, Function<Throwable, ? extends Streamable<? extends R>> onErrorMapper, Supplier<? extends Streamable<? extends R>> onCompleteSupplier, StandardConcurrentBufferedConfig config)

`groupBy`

<K> Streamable<GroupedStreamable<K, T>> groupBy(Function<? super T, ? extends K> keySelector, StandardBufferedConfig config)
<K, V> Streamable<GroupedStreamable<K, V>> groupBy(Function<? super T, ? extends K> keySelector, Function<? super T, ? extends V> valueSelector)
<K, V> Streamable<GroupedStreamable<K, V>> groupBy(Function<? super T, ? extends K> keySelector, Function<? super T, ? extends V> valueSelector, StandardBufferedConfig config)
<K, V> Streamable<GroupedStreamable<K, V>> groupBy(Function<? super T, ? extends K> keySelector, Function<? super T, ? extends V> valueSelector, Function<? super Consumer<Object>, ? extends Map<K, Object>> evictingMapFactory, StandardBufferedConfig config)

`intervalRange`

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

`just`

static <T> Streamable<T> just(T item1, T item2)
static <T> Streamable<T> just(T item1, T item2, T item3)
static <T> Streamable<T> just(T item1, T item2, T item3, T item4)
static <T> Streamable<T> just(T item1, T item2, T item3, T item4, T item5)
static <T> Streamable<T> just(T item1, T item2, T item3, T item4, T item5, T item6)
static <T> Streamable<T> just(T item1, T item2, T item3, T item4, T item5, T item6, T item7)
static <T> Streamable<T> just(T item1, T item2, T item3, T item4, T item5, T item6, T item7, T item8)
static <T> Streamable<T> just(T item1, T item2, T item3, T item4, T item5, T item6, T item7, T item8, T item9)
static <T> Streamable<T> just(T item1, T item2, T item3, T item4, T item5, T item6, T item7, T item8, T item9, T item10)

`repeat`

Streamable<T> repeat()

`repeatWhen`

Streamable<T> repeatWhen(Function<? super Streamable<Object>, ? extends Streamable<?>> handler)

`retry`

Streamable<T> retry()
Streamable<T> retry(BiPredicate<? super Integer, ? super Throwable> predicate)
Streamable<T> retry(long times, Predicate<? super Throwable> predicate)

`retryWhen`

Streamable<T> retryWhen(Function<? super Streamable<Throwable>, ? extends Streamable<?>> handler)

`skip`

Streamable<T> skip(long time, TimeUnit unit)
Streamable<T> skip(long time, TimeUnit unit, Scheduler scheduler)

`subscribe`

Disposable subscribe()
Disposable subscribe(Consumer<? super T> onNext)
Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError)
Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete)
Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, DisposableContainer container)

`take`

Streamable<T> take(long time, TimeUnit unit)
Streamable<T> take(long time, TimeUnit unit, Scheduler scheduler)

`takeUntil`

Streamable<T> takeUntil(Predicate<? super T> stopPredicate)

`test`

TestSubscriber<T> test(long initialRequest)
TestSubscriber<T> test(long initialRequest, boolean cancel)

`timeout`

<V> Streamable<T> timeout(Function<? super T, ? extends Streamable<V>> itemTimeoutIndicator)
<V> Streamable<T> timeout(Function<? super T, ? extends Streamable<V>> itemTimeoutIndicator, Streamable<? extends T> fallback)
Streamable<T> timeout(long timeout, TimeUnit unit)
<U, V> Streamable<T> timeout(Streamable firstTimeoutIndicator, Function<? super T, ? extends Streamable<V>> itemTimeoutIndicator)
Streamable<T> timeout(long timeout, TimeUnit unit, Streamable<? extends T> fallback)
Streamable<T> timeout(long timeout, TimeUnit unit, Scheduler scheduler)
<U, V> Streamable<T> timeout(Streamable firstTimeoutIndicator, Function<? super T, ? extends Streamable<V>> itemTimeoutIndicator, Streamable<? extends T> fallback)

`timer`

static Streamable<Long> timer(long delay, TimeUnit unit)

`using`

static <T, D> Streamable<T> using(Supplier<? extends D> resourceSupplier, Function<? super D, ? extends Streamable<? extends T>> sourceSupplier, Consumer<? super D> resourceCleanup, boolean eager)

`zip`

static <T, R> Streamable<R> zip(Iterable<? extends Streamable<? extends T>> sources, Function<? super Object[], ? extends R> zipper)
static <T, R> Streamable<R> zip(Iterable<? extends Streamable<? extends T>> sources, Function<? super Object[], ? extends R> zipper, StandardBufferedConfig config)
static <T1, T2, R> Streamable<R> zip(Streamable<? extends T1> source1, Streamable<? extends T2> source2, BiFunction<? super T1, ? super T2, ? extends R> zipper)
static <T1, T2, R> Streamable<R> zip(Streamable<? extends T1> source1, Streamable<? extends T2> source2, BiFunction<? super T1, ? super T2, ? extends R> zipper, StandardBufferedConfig config)
static <T1, T2, T3, R> Streamable<R> zip(Streamable<? extends T1> source1, Streamable<? extends T2> source2, Streamable<? extends T3> source3, Function3<? super T1, ? super T2, ? super T3, ? extends R> zipper)
static <T1, T2, T3, T4, R> Streamable<R> zip(Streamable<? extends T1> source1, Streamable<? extends T2> source2, Streamable<? extends T3> source3, Streamable<? extends T4> source4, Function4<? super T1, ? super T2, ? super T3, ? super T4, ? extends R> zipper)
static <T1, T2, T3, T4, T5, R> Streamable<R> zip(Streamable<? extends T1> source1, Streamable<? extends T2> source2, Streamable<? extends T3> source3, Streamable<? extends T4> source4, Streamable<? extends T5> source5, Function5<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? extends R> zipper)
static <T1, T2, T3, T4, T5, T6, R> Streamable<R> zip(Streamable<? extends T1> source1, Streamable<? extends T2> source2, Streamable<? extends T3> source3, Streamable<? extends T4> source4, Streamable<? extends T5> source5, Streamable<? extends T6> source6, Function6<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? extends R> zipper)
static <T1, T2, T3, T4, T5, T6, T7, R> Streamable<R> zip(Streamable<? extends T1> source1, Streamable<? extends T2> source2, Streamable<? extends T3> source3, Streamable<? extends T4> source4, Streamable<? extends T5> source5, Streamable<? extends T6> source6, Streamable<? extends T7> source7, Function7<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? extends R> zipper)
static <T1, T2, T3, T4, T5, T6, T7, T8, R> Streamable<R> zip(Streamable<? extends T1> source1, Streamable<? extends T2> source2, Streamable<? extends T3> source3, Streamable<? extends T4> source4, Streamable<? extends T5> source5, Streamable<? extends T6> source6, Streamable<? extends T7> source7, Streamable<? extends T8> source8, Function8<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? extends R> zipper)
static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Streamable<R> zip(Streamable<? extends T1> source1, Streamable<? extends T2> source2, Streamable<? extends T3> source3, Streamable<? extends T4> source4, Streamable<? extends T5> source5, Streamable<? extends T6> source6, Streamable<? extends T7> source7, Streamable<? extends T8> source8, Streamable<? extends T9> source9, Function9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9, ? extends R> zipper)

Methods that are entirely missing from `Streamable` (140 methods, 317 overloads)

`all`

Single<Boolean> all(Predicate<? super T> predicate)

`amb`

static <T> Streamable<T> amb(Iterable<? extends Streamable<? extends T>> sources)

`ambArray`

static <T> Streamable<T> ambArray(Streamable<? extends T>... sources)

`ambWith`

Streamable<T> ambWith(Streamable<? extends T> other)

`any`

Single<Boolean> any(Predicate<? super T> predicate)

`blockingForEach`

void blockingForEach(Consumer<? super T> onNext)
void blockingForEach(Consumer<? super T> onNext, int bufferSize)

`blockingIterable`

Iterable<T> blockingIterable()
Iterable<T> blockingIterable(int bufferSize)

`blockingLatest`

Iterable<T> blockingLatest()

`blockingMostRecent`

Iterable<T> blockingMostRecent(T initialItem)

`blockingNext`

Iterable<T> blockingNext()

`blockingSingle`

T blockingSingle()
T blockingSingle(T defaultItem)

`blockingStream`

Stream<T> blockingStream()
Stream<T> blockingStream(int prefetch)

`blockingSubscribe`

void blockingSubscribe()
void blockingSubscribe(Consumer<? super T> onNext)
void blockingSubscribe(Subscriber<? super T> subscriber)
void blockingSubscribe(Consumer<? super T> onNext, int bufferSize)
void blockingSubscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError)
void blockingSubscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, int bufferSize)
void blockingSubscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete)
void blockingSubscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, int bufferSize)

`buffer`

Streamable<List<T>> buffer(int count)
 Streamable<List<T>> buffer(Streamable boundaryIndicator)
Streamable<List<T>> buffer(int count, int skip)
> Streamable buffer(int count, Supplier bufferSupplier)
Streamable<List<T>> buffer(long timespan, TimeUnit unit)
<TOpening, TClosing> Streamable<List<T>> buffer(Streamable<? extends TOpening> openingIndicator, Function<? super TOpening, ? extends Streamable<? extends TClosing>> closingIndicator)
 Streamable<List<T>> buffer(Streamable boundaryIndicator, int initialCapacity)
<B, U extends Collection<? super T>> Streamable buffer(Streamable boundaryIndicator, Supplier bufferSupplier)
> Streamable buffer(int count, int skip, Supplier bufferSupplier)
Streamable<List<T>> buffer(long timespan, long timeskip, TimeUnit unit)
Streamable<List<T>> buffer(long timespan, TimeUnit unit, int count)
Streamable<List<T>> buffer(long timespan, TimeUnit unit, Scheduler scheduler)
<TOpening, TClosing, U extends Collection<? super T>> Streamable buffer(Streamable<? extends TOpening> openingIndicator, Function<? super TOpening, ? extends Streamable<? extends TClosing>> closingIndicator, Supplier bufferSupplier)
Streamable<List<T>> buffer(long timespan, long timeskip, TimeUnit unit, Scheduler scheduler)
Streamable<List<T>> buffer(long timespan, TimeUnit unit, Scheduler scheduler, int count)
> Streamable buffer(long timespan, long timeskip, TimeUnit unit, Scheduler scheduler, Supplier bufferSupplier)
> Streamable buffer(long timespan, TimeUnit unit, Scheduler scheduler, int count, Supplier bufferSupplier, boolean restartTimerOnMaxSize)

`cache`

Streamable<T> cache()

`cacheWithInitialCapacity`

Streamable<T> cacheWithInitialCapacity(int initialCapacity)

`cast`

 Streamable cast(Class clazz)

`collectInto`

 Single collectInto(U initialItem, BiConsumer<? super U, ? super T> collector)

`combineLatest`

static <T, R> Streamable<R> combineLatest(Iterable<? extends Streamable<? extends T>> sources, Function<? super Object[], ? extends R> combiner)
static <T, R> Streamable<R> combineLatest(Iterable<? extends Streamable<? extends T>> sources, Function<? super Object[], ? extends R> combiner, StandardBufferedConfig config)
static <T1, T2, R> Streamable<R> combineLatest(Streamable<? extends T1> source1, Streamable<? extends T2> source2, BiFunction<? super T1, ? super T2, ? extends R> combiner)
static <T1, T2, T3, R> Streamable<R> combineLatest(Streamable<? extends T1> source1, Streamable<? extends T2> source2, Streamable<? extends T3> source3, Function3<? super T1, ? super T2, ? super T3, ? extends R> combiner)
static <T1, T2, T3, T4, R> Streamable<R> combineLatest(Streamable<? extends T1> source1, Streamable<? extends T2> source2, Streamable<? extends T3> source3, Streamable<? extends T4> source4, Function4<? super T1, ? super T2, ? super T3, ? super T4, ? extends R> combiner)
static <T1, T2, T3, T4, T5, R> Streamable<R> combineLatest(Streamable<? extends T1> source1, Streamable<? extends T2> source2, Streamable<? extends T3> source3, Streamable<? extends T4> source4, Streamable<? extends T5> source5, Function5<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? extends R> combiner)
static <T1, T2, T3, T4, T5, T6, R> Streamable<R> combineLatest(Streamable<? extends T1> source1, Streamable<? extends T2> source2, Streamable<? extends T3> source3, Streamable<? extends T4> source4, Streamable<? extends T5> source5, Streamable<? extends T6> source6, Function6<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? extends R> combiner)
static <T1, T2, T3, T4, T5, T6, T7, R> Streamable<R> combineLatest(Streamable<? extends T1> source1, Streamable<? extends T2> source2, Streamable<? extends T3> source3, Streamable<? extends T4> source4, Streamable<? extends T5> source5, Streamable<? extends T6> source6, Streamable<? extends T7> source7, Function7<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? extends R> combiner)
static <T1, T2, T3, T4, T5, T6, T7, T8, R> Streamable<R> combineLatest(Streamable<? extends T1> source1, Streamable<? extends T2> source2, Streamable<? extends T3> source3, Streamable<? extends T4> source4, Streamable<? extends T5> source5, Streamable<? extends T6> source6, Streamable<? extends T7> source7, Streamable<? extends T8> source8, Function8<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? extends R> combiner)
static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Streamable<R> combineLatest(Streamable<? extends T1> source1, Streamable<? extends T2> source2, Streamable<? extends T3> source3, Streamable<? extends T4> source4, Streamable<? extends T5> source5, Streamable<? extends T6> source6, Streamable<? extends T7> source7, Streamable<? extends T8> source8, Streamable<? extends T9> source9, Function9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9, ? extends R> combiner)

`combineLatestArray`

static <T, R> Streamable<R> combineLatestArray(Streamable<? extends T>[] sources, Function<? super Object[], ? extends R> combiner)
`static <T, R> Streamable combineLatestArray(Streamable<? extends T>[] sourc

Source: ReactiveX/RxJava

View original on GitHub View discussion on GitHub

4.x: Streamable operator checklist

Streamable: missing Flowable operators and overloads (4.x parity checklist)

Methods that exist on Streamable but are missing overloads (23 methods, 69 overloads)

blockingFirst

blockingLast

collect

concat

delay

error

flatMap

groupBy

intervalRange

just

repeat

repeatWhen

retry

retryWhen

skip

subscribe

take

takeUntil

test

timeout

timer

using

zip

Methods that are entirely missing from Streamable (140 methods, 317 overloads)

all

amb

ambArray

ambWith

any

blockingForEach

blockingIterable

blockingLatest

blockingMostRecent

blockingNext

blockingSingle

blockingStream

blockingSubscribe

buffer

cache

cacheWithInitialCapacity

cast

collectInto

combineLatest

combineLatestArray