From 9e71899c5f08012a28f191ec5d70505276d66e61 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Sat, 18 Jul 2026 15:19:19 +0200 Subject: [PATCH] 4.x: Streamable mapOptional +perf --- .../streamable/StreamableMapOptionalPerf.java | 153 +++++++++++++ .../streamable/StreamableHelper.java | 13 +- .../streamable/StreamableMapOptional.java | 209 +++++++++++++++--- .../streamable/StreamableMapOptionalTest.java | 139 +++++++++++- 4 files changed, 469 insertions(+), 45 deletions(-) create mode 100644 src/jmh/java/io/reactivex/rxjava4/streamable/StreamableMapOptionalPerf.java diff --git a/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableMapOptionalPerf.java b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableMapOptionalPerf.java new file mode 100644 index 0000000000..07f0aabfeb --- /dev/null +++ b/src/jmh/java/io/reactivex/rxjava4/streamable/StreamableMapOptionalPerf.java @@ -0,0 +1,153 @@ +/* + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava4.streamable; + +import java.util.*; +import java.util.concurrent.TimeUnit; +import java.util.stream.*; + +import org.openjdk.jmh.annotations.*; + +import io.reactivex.rxjava4.core.Streamable; + +/// +/// The map is one of the most used operator in the ecosystem so it must be fast and +/// it must support operator fusion across itself. +/// +/// i9 275HX, 32GB LPDDR5 6400MT CL52, Windows 25H2, JDK 26.0.1 +/// +/// # 0. Baseline +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableMapOptionalPerf.basic 1 thrpt 5 25824335,224 ┬▒ 307158,457 ops/s +/// StreamableMapOptionalPerf.basic 10 thrpt 5 4182696,220 ┬▒ 41200,808 ops/s +/// StreamableMapOptionalPerf.basic 100 thrpt 5 475284,966 ┬▒ 10753,634 ops/s +/// StreamableMapOptionalPerf.basic 1000 thrpt 5 46365,872 ┬▒ 283,392 ops/s +/// StreamableMapOptionalPerf.basic 10000 thrpt 5 4635,538 ┬▒ 54,545 ops/s +/// StreamableMapOptionalPerf.basic 100000 thrpt 5 455,119 ┬▒ 7,502 ops/s +/// StreamableMapOptionalPerf.basic 1000000 thrpt 5 50,573 ┬▒ 4,533 ops/s +/// StreamableMapOptionalPerf.enumerated 1 thrpt 5 13233507,814 ┬▒ 150935,433 ops/s +/// StreamableMapOptionalPerf.enumerated 10 thrpt 5 3433233,629 ┬▒ 83102,643 ops/s +/// StreamableMapOptionalPerf.enumerated 100 thrpt 5 417478,655 ┬▒ 19913,394 ops/s +/// StreamableMapOptionalPerf.enumerated 1000 thrpt 5 42681,071 ┬▒ 399,061 ops/s +/// StreamableMapOptionalPerf.enumerated 10000 thrpt 5 4662,471 ┬▒ 246,930 ops/s +/// StreamableMapOptionalPerf.enumerated 100000 thrpt 5 470,937 ┬▒ 6,526 ops/s +/// StreamableMapOptionalPerf.enumerated 1000000 thrpt 5 41,139 ┬▒ 1,678 ops/s +/// StreamableMapOptionalPerf.indexed 1 thrpt 5 17011123,156 ┬▒ 430025,426 ops/s +/// StreamableMapOptionalPerf.indexed 10 thrpt 5 3775634,947 ┬▒ 36893,569 ops/s +/// StreamableMapOptionalPerf.indexed 100 thrpt 5 448558,858 ┬▒ 9594,675 ops/s +/// StreamableMapOptionalPerf.indexed 1000 thrpt 5 49462,107 ┬▒ 436,808 ops/s +/// StreamableMapOptionalPerf.indexed 10000 thrpt 5 4979,096 ┬▒ 80,306 ops/s +/// StreamableMapOptionalPerf.indexed 100000 thrpt 5 499,161 ┬▒ 2,866 ops/s +/// StreamableMapOptionalPerf.indexed 1000000 thrpt 5 47,332 ┬▒ 1,081 ops/s +/// ``` +/// +/// # 1. Use the Claude Fable state machine for avoiding whenComplete and atomics +/// +/// + 72% on times 1 vs baseline, +212% on times million vs bsaeline +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableMapOptionalPerf.basic 1 thrpt 5 44420406,456 ┬▒ 467265,689 ops/s +/// StreamableMapOptionalPerf.basic 10 thrpt 5 14548571,112 ┬▒ 204608,524 ops/s +/// StreamableMapOptionalPerf.basic 100 thrpt 5 1696277,847 ┬▒ 26481,422 ops/s +/// StreamableMapOptionalPerf.basic 1000 thrpt 5 171547,633 ┬▒ 5148,566 ops/s +/// StreamableMapOptionalPerf.basic 10000 thrpt 5 16837,603 ┬▒ 237,627 ops/s +/// StreamableMapOptionalPerf.basic 100000 thrpt 5 1556,065 ┬▒ 18,238 ops/s +/// StreamableMapOptionalPerf.basic 1000000 thrpt 5 157,966 ┬▒ 3,666 ops/s +/// StreamableMapOptionalPerf.enumerated 1 thrpt 5 20465896,376 ┬▒ 338492,857 ops/s +/// StreamableMapOptionalPerf.enumerated 10 thrpt 5 10045416,375 ┬▒ 119490,078 ops/s +/// StreamableMapOptionalPerf.enumerated 100 thrpt 5 1403277,356 ┬▒ 26248,074 ops/s +/// StreamableMapOptionalPerf.enumerated 1000 thrpt 5 126858,293 ┬▒ 1749,556 ops/s +/// StreamableMapOptionalPerf.enumerated 10000 thrpt 5 12249,540 ┬▒ 173,441 ops/s +/// StreamableMapOptionalPerf.enumerated 100000 thrpt 5 1248,849 ┬▒ 72,192 ops/s +/// StreamableMapOptionalPerf.enumerated 1000000 thrpt 5 125,087 ┬▒ 2,269 ops/s +/// StreamableMapOptionalPerf.indexed 1 thrpt 5 23867521,911 ┬▒ 794423,143 ops/s +/// StreamableMapOptionalPerf.indexed 10 thrpt 5 10788609,481 ┬▒ 266403,589 ops/s +/// StreamableMapOptionalPerf.indexed 100 thrpt 5 1595466,285 ┬▒ 31921,714 ops/s +/// StreamableMapOptionalPerf.indexed 1000 thrpt 5 159969,124 ┬▒ 2228,065 ops/s +/// StreamableMapOptionalPerf.indexed 10000 thrpt 5 16042,879 ┬▒ 268,651 ops/s +/// StreamableMapOptionalPerf.indexed 100000 thrpt 5 1586,144 ┬▒ 20,156 ops/s +/// StreamableMapOptionalPerf.indexed 1000000 thrpt 5 152,721 ┬▒ 5,313 ops/s +/// ``` +/// +/// # 2. add Deferred and direct EnumerableSource paths +/// +/// Basic has too big of a variance compared to 1. +/// Enumerated: +46% over 1, +126% over baseline for times 1 +/// +68 over 1, +413% over baseline for times million +/// +/// ``` +/// Benchmark (times) Mode Cnt Score Error Units +/// StreamableMapOptionalPerf.basic 1 thrpt 5 61437824,388 ┬▒ 19274231,452 ops/s +/// StreamableMapOptionalPerf.basic 10 thrpt 5 16686712,361 ┬▒ 1317978,227 ops/s +/// StreamableMapOptionalPerf.basic 100 thrpt 5 2037714,671 ┬▒ 59005,537 ops/s +/// StreamableMapOptionalPerf.basic 1000 thrpt 5 185415,957 ┬▒ 26578,676 ops/s +/// StreamableMapOptionalPerf.basic 10000 thrpt 5 18706,795 ┬▒ 384,281 ops/s +/// StreamableMapOptionalPerf.basic 100000 thrpt 5 1522,088 ┬▒ 43,043 ops/s +/// StreamableMapOptionalPerf.basic 1000000 thrpt 5 170,135 ┬▒ 11,148 ops/s +/// StreamableMapOptionalPerf.enumerated 1 thrpt 5 29930443,412 ┬▒ 1169887,938 ops/s +/// StreamableMapOptionalPerf.enumerated 10 thrpt 5 52029667,470 ┬▒ 2723918,189 ops/s +/// StreamableMapOptionalPerf.enumerated 100 thrpt 5 6716969,521 ┬▒ 1999419,008 ops/s +/// StreamableMapOptionalPerf.enumerated 1000 thrpt 5 234588,596 ┬▒ 7713,319 ops/s +/// StreamableMapOptionalPerf.enumerated 10000 thrpt 5 24133,205 ┬▒ 725,742 ops/s +/// StreamableMapOptionalPerf.enumerated 100000 thrpt 5 2209,969 ┬▒ 240,498 ops/s +/// StreamableMapOptionalPerf.enumerated 1000000 thrpt 5 211,364 ┬▒ 24,358 ops/s +/// StreamableMapOptionalPerf.indexed 1 thrpt 5 91243822,427 ┬▒ 2337628,568 ops/s +/// StreamableMapOptionalPerf.indexed 10 thrpt 5 51131100,389 ┬▒ 5848224,433 ops/s +/// StreamableMapOptionalPerf.indexed 100 thrpt 5 4407887,744 ┬▒ 240068,362 ops/s +/// StreamableMapOptionalPerf.indexed 1000 thrpt 5 452836,278 ┬▒ 113450,062 ops/s +/// StreamableMapOptionalPerf.indexed 10000 thrpt 5 44535,947 ┬▒ 10809,809 ops/s +/// StreamableMapOptionalPerf.indexed 100000 thrpt 5 4528,836 ┬▒ 509,763 ops/s +/// StreamableMapOptionalPerf.indexed 1000000 thrpt 5 411,513 ┬▒ 10,556 ops/s +/// ``` +@BenchmarkMode(Mode.Throughput) +@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) +@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) +@OutputTimeUnit(TimeUnit.SECONDS) +@Fork(value = 1) +@State(Scope.Thread) +public class StreamableMapOptionalPerf { + @Param({ "1", "10", "100", "1000", "10000", "100000", "1000000" }) + public int times; + + Streamable result; + Streamable> indexedMax; + Streamable> enumeratedMax; + + @Setup + public void setup() { + result = Streamable.range(1, times) + .mapOptional(v -> (v & 1) != 0 ? Optional.of(v + 1) : Optional.empty()) + ; + indexedMax = result.collect(Collectors.maxBy(Comparator.naturalOrder())); + enumeratedMax = Streamable.fromIterable(() -> IntStream.range(1, 1 + times).iterator()) + .mapOptional(v -> (v & 1) != 0 ? Optional.of(v + 1) : Optional.empty()) + .collect(Collectors.maxBy(Comparator.naturalOrder())); + } + + @Benchmark + public Object basic() { + return result.blockingLast(); + } + + @Benchmark + public Object indexed() { + return indexedMax.blockingLast(); + } + + @Benchmark + public Object enumerated() { + return enumeratedMax.blockingLast(); + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java index ecb876fedf..fddb267544 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java @@ -410,14 +410,11 @@ public static StreamableInterceptConfig createOnError(Consumer void whenComplete(CompletionStage stage, java.util.function.BiConsumer consumer) { - if (stage instanceof CompletableFuture cf) { - switch (cf.state()) { - case SUCCESS -> consumer.accept(cf.getNow(null), null); - case CANCELLED, FAILED -> consumer.accept(null, cf.exceptionNow()); - default -> stage.whenComplete(consumer); - } - } else { - stage.whenComplete(consumer); + var cf = stage.toCompletableFuture(); + switch (cf.state()) { + case SUCCESS -> consumer.accept(cf.getNow(null), null); + case CANCELLED, FAILED -> consumer.accept(null, cf.exceptionNow()); + default -> stage.whenComplete(consumer); } } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMapOptional.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMapOptional.java index 52bb9344bb..7b7802ffcb 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMapOptional.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMapOptional.java @@ -13,9 +13,12 @@ package io.reactivex.rxjava4.internal.operators.streamable; +import java.io.Serial; import java.util.*; import java.util.concurrent.*; +import java.util.concurrent.Future.State; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; import io.reactivex.rxjava4.annotations.NonNull; import io.reactivex.rxjava4.core.*; @@ -23,34 +26,122 @@ import io.reactivex.rxjava4.exceptions.Exceptions; import io.reactivex.rxjava4.functions.Function; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; +import io.reactivex.rxjava4.operators.*; public record StreamableMapOptional( @NonNull Streamable source, @NonNull Function> mapper) implements Streamable, HasUpstreamStreamableSource { + @SuppressWarnings("unchecked") @Override public @NonNull Streamer<@NonNull R> stream(@NonNull StreamerCancellation cancellation) { - return new MapStreamer<>(source.stream(cancellation), mapper); + var upstream = source.stream(cancellation); + if (upstream instanceof DeferredEnumerableSource dsrc) { + return new MapOptionalStreamerDeferred(upstream, (DeferredEnumerableSource)dsrc, mapper); + } else + if (upstream instanceof EnumerableSource esrc) { + return new MapOptionalStreamerEnumerable(upstream, (EnumerableSource)esrc, mapper); + } + return new MapOptionalStreamerBasic<>(upstream, mapper); } - static final class MapStreamer implements Streamer { + static abstract class MapOptionalStreamer extends AtomicInteger + implements Streamer, BiConsumer { + + @Serial + private static final long serialVersionUID = 6360266262761297472L; + final Streamer upstream; + final Function> mapper; - volatile R current; - final AtomicInteger wip = new AtomicInteger(); + R current; + + CompletableFuture nextReady; - MapStreamer(Streamer upstream, Function> mapper) { + MapOptionalStreamer(Streamer upstream, Function> mapper) { this.upstream = upstream; this.mapper = mapper; } @Override public @NonNull CompletionStage next() { - var cf = new CompletableFuture(); - drain(cf); - return cf; + nextReady = new CompletableFuture<>(); + drain(); + return nextReady; + } + + void drain() { + for (;;) { + var upstreamNext = upstream.next().toCompletableFuture(); + var state = upstreamNext.state(); + + if (state == State.RUNNING) { + set(1); + upstreamNext.whenComplete(this); + if (compareAndSet(1, 0)) { + return; + } + state = upstreamNext.state(); + } + if (state == State.SUCCESS) { + if (upstreamNext.getNow(false)) { + boolean pass; + T value = upstream.current(); + Optional result = null; + try { + result = Objects.requireNonNull(mapper.apply(value), "The mapper returned a null item"); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + nextReady.completeExceptionally(ex); + return; + } + pass = result.isEmpty(); + if (!pass) { + current = result.get(); + nextReady.complete(true); + return; + } + } else { + nextReady.complete(false); + return; + } + } else { + nextReady.completeExceptionally(upstreamNext.exceptionNow()); + return; + } + } + } + + @Override + public void accept(Boolean t, Throwable u) { + if (!compareAndSet(1, 2)) { + if (u != null) { + nextReady.completeExceptionally(u); + } else + if (t) { + boolean pass; + T value = upstream.current(); + Optional result = null; + try { + result = Objects.requireNonNull(mapper.apply(value), "The mapper returned a null item"); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + nextReady.completeExceptionally(ex); + return; + } + pass = result.isEmpty(); + if (!pass) { + current = result.get(); + nextReady.complete(true); + return; + } + drain(); + } else { + nextReady.complete(false); + } + } } @Override @@ -63,36 +154,84 @@ static final class MapStreamer implements Streamer { current = null; return upstream.finish(); } + } + + static final class MapOptionalStreamerBasic extends MapOptionalStreamer { + + @Serial + private static final long serialVersionUID = -1776499708262136887L; + + MapOptionalStreamerBasic(Streamer upstream, Function> mapper) { + super(upstream, mapper); + } + } + + static final class MapOptionalStreamerDeferred extends MapOptionalStreamer + implements DeferredEnumerableSource { + + @Serial + private static final long serialVersionUID = -1776499708262136887L; + + final DeferredEnumerableSource deferred; + + MapOptionalStreamerDeferred(Streamer upstream, + DeferredEnumerableSource deferred, + Function> mapper) { + super(upstream, mapper); + this.deferred = deferred; + } - void drain(CompletableFuture cf) { - if (wip.getAndIncrement() != 0) { - return; + @Override + public boolean nextSync() throws Throwable { + for (;;) { + if (deferred.nextSync()) { + var value = deferred.current(); + var result = Objects.requireNonNull(mapper.apply(value), "The mapper returned a null item"); + if (!result.isEmpty()) { + current = result.get(); + return true; + } + } else { + return false; + } } - do { - upstream.next() - .whenComplete((v, e) -> { - if (e != null) { - cf.completeExceptionally(e); - } else { - if (v) { - try { - var w = Objects.requireNonNull(mapper.apply(upstream.current()), "The mapper returned a null value"); - if (w.isPresent()) { - current = w.get(); - cf.complete(true); - } else { - drain(cf); - } - } catch (Throwable ex) { - Exceptions.throwIfFatal(ex); - cf.completeExceptionally(ex); - } - } else { - cf.complete(false); - } + } + + @Override + public CompletionStage enumerableReady() { + return deferred.enumerableReady(); + } + } + + static final class MapOptionalStreamerEnumerable extends MapOptionalStreamer + implements EnumerableSource { + + @Serial + private static final long serialVersionUID = -1776499708262136887L; + + final EnumerableSource deferred; + + MapOptionalStreamerEnumerable(Streamer upstream, + EnumerableSource deferred, + Function> mapper) { + super(upstream, mapper); + this.deferred = deferred; + } + + @Override + public boolean nextSync() throws Throwable { + for (;;) { + if (deferred.nextSync()) { + var value = deferred.current(); + var result = Objects.requireNonNull(mapper.apply(value), "The mapper returned a null item"); + if (!result.isEmpty()) { + current = result.get(); + return true; } - }); - } while (wip.decrementAndGet() != 0); + } else { + return false; + } + } } } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMapOptionalTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMapOptionalTest.java index 4c81a5dae3..4d2b618104 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMapOptionalTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableMapOptionalTest.java @@ -13,12 +13,16 @@ package io.reactivex.rxjava4.internal.operators.streamable; -import java.util.Optional; +import java.util.*; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; import org.junit.jupiter.api.Test; -import io.reactivex.rxjava4.core.Streamable; + +import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.exceptions.TestException; +import io.reactivex.rxjava4.processors.DispatchStreamProcessor; +import io.reactivex.rxjava4.schedulers.Schedulers; public class StreamableMapOptionalTest extends StreamableBaseTest { @@ -96,4 +100,135 @@ public void sourceError() { .assertFailure(TestException.class); } + public void intervalRange() { + Streamable.intervalRange(1, 5, 1, 1, TimeUnit.MILLISECONDS, Schedulers.single()) + .mapOptional(v -> Optional.of(v.toString())) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult("1", "2", "3", "4", "5"); + } + + @Test + public void delay() { + Streamable.range(1, 5) + .delay(1, TimeUnit.MILLISECONDS, Schedulers.single()) + .mapOptional(v -> Optional.of(v.toString())) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of("1", "2", "3", "4", "5")); + } + + @Test + public void delayNull() { + Streamable.range(1, 5) + .delay(1, TimeUnit.MILLISECONDS, Schedulers.single()) + .mapOptional(_ -> null) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(NullPointerException.class); + } + + @Test + public void delayCrash() { + Streamable.range(1, 5) + .delay(1, TimeUnit.MILLISECONDS, Schedulers.single()) + .mapOptional(_ -> { throw new TestException(); }) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void delayMIxed() { + Streamable.range(1, 5) + .delay(1, TimeUnit.MILLISECONDS, Schedulers.single()) + .mapOptional(v -> v % 2 == 0 ? Optional.of(v.toString()) : Optional.empty()) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of("2", "4")); + } + + @Test + public void asyncError() throws Throwable { + var dsp = new DispatchStreamProcessor(); + + var ts = dsp + .mapOptional(v -> v % 2 == 0 ? Optional.of(v.toString()) : Optional.empty()) + .test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + awaitStreamers(dsp, 1000); + + dsp.next(1).toCompletableFuture().join(); + dsp.finish(new TestException()).toCompletableFuture().join(); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void asyncEmpty() throws Throwable { + var dsp = new DispatchStreamProcessor(); + + var ts = dsp + .mapOptional(v -> v % 2 == 0 ? Optional.of(v.toString()) : Optional.empty()) + .test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + awaitStreamers(dsp, 1000); + + dsp.next(1).toCompletableFuture().join(); + dsp.finish(null).toCompletableFuture().join(); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void enumerable() { + Streamable.range(1, 5) + .mapOptional(v -> Optional.of(v.toString())) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of("1", "2", "3", "4", "5")); + } + + @Test + public void enumerable2() { + Streamable.fromIterable(List.of(1, 2, 3, 4, 5)) + .mapOptional(v -> Optional.of(v.toString())) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of("1", "2", "3", "4", "5")); + } + + @Test + public void enumerable3() { + Streamable.fromIterable(List.of(1, 2, 3, 4, 5)) + .mapOptional(v -> v % 2 == 0 ? Optional.of(v.toString()) : Optional.empty()) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of("2", "4")); + } + + @Test + public void deferredEnumerable() { + Single.just(1) + .flattenAsStreamable(v -> List.of(v, 2, 3, 4, 5)) + .mapOptional(v -> v % 2 == 0 ? Optional.of(v.toString()) : Optional.empty()) + .collect(Collectors.toList()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(List.of("2", "4")); + } + }