diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index 218dcc3c28..a6c983dece 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -79,10 +79,6 @@ public interface Streamable<@NonNull T> { @NonNull Streamer stream(@NonNull DisposableContainer cancellation); - // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo - // HELPERS - // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo - // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo // Data sources and wrappers // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo @@ -111,6 +107,21 @@ public interface Streamable<@NonNull T> { }, executor); } + /** + * Streams the {@code Streamable}s one after the other from the given iterable sequence + * of {@code Streamable}s. + * @param the element type of the inner and resulting {@code Streamable}s + * @param sources the iterable sequence of {@code Streamable}s. + * @return the new {@code Streamable} source + * @throws NullPointerException if {@code sources} is {@code null} + */ + @CheckReturnValue + @NonNull + static <@NonNull T> Streamable concat(Iterable> sources) { + Objects.requireNonNull(sources, "sources is null"); + return RxJavaPlugins.onAssembly(new StreamableConcatIterable<>(sources, ErrorMode.IMMEDIATE)); // TODO implement + } + /** * Generate a sequence of values via a virtual generator callback (yielder) * which is free to block and is natively backpressured. @@ -514,6 +525,8 @@ static Streamable rangeLong(long start, long count) { * @return the new {@code Streamable} instance * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} */ + @CheckReturnValue + @NonNull static Streamable timer(long delay, TimeUnit unit, Scheduler scheduler) { Objects.requireNonNull(unit, "unit is null"); Objects.requireNonNull(scheduler, "scheduler is null"); @@ -533,12 +546,39 @@ static Streamable timer(long delay, TimeUnit unit, Scheduler scheduler) { * @return the new {@code Streamable} instance * @throws NullPointerException if {@code unit} or {@code executor} is {@code null} */ + @CheckReturnValue + @NonNull static Streamable timer(long delay, TimeUnit unit, ExecutorService executor) { Objects.requireNonNull(unit, "unit is null"); Objects.requireNonNull(executor, "executor is null"); return RxJavaPlugins.onAssembly(new StreamableTimer(delay, unit, null, executor)); } + /** + * For each incoming streamer, this operator creates a resource, then + * uses that resource to create the actual {@code Streamable} instance to + * stream value of and then uses a cleaner callback to dissolve the resource + * once the {@code Streamable} terminated. + * @param the element type of the sequence + * @param the resource type + * @param resourceSupplier supplies a resource object per {@link #stream(DisposableContainer)} call + * @param resourceMapper maps the supplied resource into a {@code Streamable} source + * @param resourceCleaner cleans up the supplied resource + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code resourceSupplier} or {@code resourceMapper} + * or {@code resourceCleaner} is {@code null} + */ + @CheckReturnValue + @NonNull + static Streamable using(Supplier resourceSupplier, + Function> resourceMapper, + Consumer resourceCleaner) { + Objects.requireNonNull(resourceSupplier, "resourceSupplier is null"); + Objects.requireNonNull(resourceMapper, "resourceMapper is null"); + Objects.requireNonNull(resourceCleaner, "resourceCleaner is null"); + return RxJavaPlugins.onAssembly(new StreamableUsing<>(resourceSupplier, resourceMapper, resourceCleaner)); + } + /** * Takes the next element from each source {@code Streamable} and emits them a a single * row of {@link List}. @@ -550,6 +590,8 @@ static Streamable timer(long delay, TimeUnit unit, ExecutorService executo * @return the new {@code Streamable} instance * @throws NullPointerException if {@code sources} is {@&ode null} */ + @CheckReturnValue + @NonNull static Streamable> zip(Iterable> sources) { Objects.requireNonNull(sources, "sources is null"); return RxJavaPlugins.onAssembly(new StreamableZip<>(sources)); @@ -570,7 +612,9 @@ static Streamable> zip(Iterable> s * @return the new {@code Streamable} instance * @throws NullPointerException if {@code collector} is {@code null} */ - default Streamable collect(Collector collector) { + @CheckReturnValue + @NonNull + default Streamable collect(@NonNull Collector collector) { Objects.requireNonNull(collector, "collector is null"); return RxJavaPlugins.onAssembly(new StreamableCollector<>(this, collector)); } @@ -583,7 +627,9 @@ default Streamable collect(Collector collector) { * @return the new {@code Streamable} instance * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} */ - default Streamable delay(long time, TimeUnit unit, Scheduler scheduler) { + @CheckReturnValue + @NonNull + default Streamable delay(long time, @NonNull TimeUnit unit, @NonNull Scheduler scheduler) { Objects.requireNonNull(unit, "unit is null"); Objects.requireNonNull(scheduler, "scheduler is null"); return RxJavaPlugins.onAssembly(new StreamableDelay<>(this, time, unit, scheduler)); @@ -595,7 +641,9 @@ default Streamable delay(long time, TimeUnit unit, Scheduler scheduler) { * @return the new {@code Streamable} instance * @throws NullPointerException if {@code consumer} is {@code null} */ - default Streamable doOnError(Consumer consumer) { + @CheckReturnValue + @NonNull + default Streamable doOnError(@NonNull Consumer consumer) { Objects.requireNonNull(consumer, "consumer is null"); return intercept(StreamableHelper.createOnError(consumer)); } @@ -606,7 +654,9 @@ default Streamable doOnError(Consumer consumer) { * @return the new {@code Streamable} instance * @throws NullPointerException if {@code consumer} is {@code null} */ - default Streamable doOnNext(Consumer consumer) { + @CheckReturnValue + @NonNull + default Streamable doOnNext(@NonNull Consumer consumer) { Objects.requireNonNull(consumer, "consumer is null"); return intercept(new StreamableInterceptConfig<>(v -> { consumer.accept(v); return v; } )); } @@ -638,7 +688,9 @@ default Streamable flatMap(@NonNull Function Streamable> groupBy(Function keySelector) { + @CheckReturnValue + @NonNull + default <@Nullable K> Streamable> groupBy(@NonNull Function keySelector) { Objects.requireNonNull(keySelector, "keySelector is null"); return RxJavaPlugins.onAssembly(new StreamableGroupBy<>(this, keySelector)); } @@ -732,7 +784,9 @@ default Streamable intercept(StreamableInterceptConfig config) { * @return the new {@code Streamable} instance * @throws NullPointerException if {@code fallbackMapper} is {@code null} */ - default Streamable onErrorResumeNext(Function> fallbackMapper) { + @CheckReturnValue + @NonNull + default Streamable onErrorResumeNext(@NonNull Function> fallbackMapper) { Objects.requireNonNull(fallbackMapper, "fallbackMapper is null"); return RxJavaPlugins.onAssembly(new StreamableOnErrorResumeNext<>(this, fallbackMapper)); } @@ -798,7 +852,9 @@ default Streamable takeWhile(@NonNull Predicate predicate) { * @return the new {@code Streamable} instance * @throws NullPointerException if {@code unit} or {@code scheduler} or {@code fallback} is {@code null} */ - default Streamable timeout(long timeout, TimeUnit unit, Scheduler scheduler, Streamable fallback) { + @CheckReturnValue + @NonNull + default Streamable timeout(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Streamable fallback) { Objects.requireNonNull(unit, "unit is null"); Objects.requireNonNull(scheduler, "scheduler is null"); Objects.requireNonNull(fallback, "fallback is null"); @@ -855,6 +911,8 @@ default Flowable toFlowable(@NonNull ExecutorService executor) { * or {@link ExecutorService} on its own. * @return the new {@code Observable} instance */ + @CheckReturnValue + @NonNull default Observable toObservable() { return RxJavaPlugins.onAssembly(new StreamableToObservable<>(this)); } @@ -1003,6 +1061,7 @@ default void subscribe(@NonNull Flow.Subscriber subscriber) { * {@code Streamable} terminates * @throws NullPointerException if {@code consumer} is {@code null} */ + @NonNull default CompletionStage subscribe(@NonNull StreamSink consumer) { return subscribe(consumer, Executors.newVirtualThreadPerTaskExecutor()); } @@ -1016,6 +1075,7 @@ default CompletionStage subscribe(@NonNull StreamSink consumer) * {@code Streamable} terminates * @throws NullPointerException if {@code consumer} or {@code executor} is {@code null} */ + @NonNull default CompletionStage subscribe(@NonNull StreamSink consumer, ExecutorService executor) { Objects.requireNonNull(consumer, "consumer is null"); Objects.requireNonNull(executor, "executor is null"); diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterable.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterable.java new file mode 100644 index 0000000000..a874a920c8 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterable.java @@ -0,0 +1,124 @@ +/* + * 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.internal.operators.streamable; + +import java.io.Serial; +import java.util.Iterator; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.DisposableContainer; + +public record StreamableConcatIterable( + Iterable> sources, + ErrorMode errorMode +) implements Streamable { + + @Override + public @NonNull Streamer<@NonNull T> stream(@NonNull DisposableContainer cancellation) { + return new ConcatIteratorStreamer<>(sources.iterator(), cancellation); + } + + static final class ConcatIteratorStreamer extends AtomicInteger implements Streamer { + + @Serial + private static final long serialVersionUID = -9136569444189652718L; + + final Iterator> iterator; + + final DisposableContainer cancellation; + + DisposableContainer currentCancellation; + + Streamer upstream; + + CompletableFuture nextReady; + + ConcatIteratorStreamer(Iterator> iterator, + DisposableContainer cancellation) { + this.iterator = iterator; + this.cancellation = cancellation; + } + + @Override + public @NonNull CompletionStage next() { + nextReady = new CompletableFuture(); + drain(); + return nextReady; + } + + @Override + public @NonNull T current() { + return upstream.current(); + } + + @Override + public @NonNull CompletionStage finish() { + var localUpstream = upstream; + var localCurrentCancellation = currentCancellation; + upstream = null; + nextReady = null; + currentCancellation = null; + if (localUpstream != null) { + cancellation.delete(localCurrentCancellation); + return localUpstream.finish(); + } + return FINISHED; + } + + void drain() { + if (getAndIncrement() != 0) { + return; + } + + do { + if (upstream == null) { + if (iterator.hasNext()) { + currentCancellation = cancellation.derive(); + var nextStreamable = iterator.next(); + if (nextStreamable == null) { + nextReady.completeExceptionally(new NullPointerException("The iterator returned a null Streamable")); + } else { + upstream = nextStreamable.stream(currentCancellation); + drain(); + } + } else { + nextReady.complete(false); + } + } else { + upstream.next().whenComplete((v, e) -> { + if (e != null) { + nextReady.completeExceptionally(e); + } else + if (v) { + nextReady.complete(true); + } else { + cancellation.delete(currentCancellation); + upstream.finish().whenComplete((_, u) -> { + if (u != null) { + nextReady.completeExceptionally(u); + } else { + upstream = null; + drain(); + } + }); + } + }); + } + } while (decrementAndGet() != 0); + } + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableUsing.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableUsing.java new file mode 100644 index 0000000000..0416a1d448 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableUsing.java @@ -0,0 +1,92 @@ +/* + * 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.internal.operators.streamable; + +import java.util.Objects; +import java.util.concurrent.*; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.DisposableContainer; +import io.reactivex.rxjava4.exceptions.Exceptions; +import io.reactivex.rxjava4.functions.*; + +public record StreamableUsing( + Supplier resourceSupplier, + Function> resourceMapper, + Consumer resourceCleaner +) +implements Streamable { + + @Override + public @NonNull Streamer<@NonNull T> stream(@NonNull DisposableContainer cancellation) { + R resource; + Streamable source; + try { + resource = resourceSupplier.get(); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + return StreamableError.createFailed(ex); + } + + try { + source = Objects.requireNonNull(resourceMapper.apply(resource), "The resourceMapper returned a null Streamable"); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + return new UsingStreamer<>(StreamableError.createFailed(ex), () -> resourceCleaner.accept(resource)); + } + + return new UsingStreamer<>(source.stream(cancellation), () -> resourceCleaner.accept(resource)); + } + + static final class UsingStreamer implements Streamer { + + final Streamer upstream; + + Action cleanup; + + UsingStreamer(Streamer upstream, Action cleanup) { + this.upstream = upstream; + this.cleanup = cleanup; + } + + @Override + public @NonNull CompletionStage next() { + return upstream.next(); + } + + @Override + public @NonNull T current() { + return upstream.current(); + } + + @Override + public @NonNull CompletionStage finish() { + var cf = new CompletableFuture(); + upstream.finish().whenComplete((_, e) -> { + try { + var c = cleanup; + cleanup = null; + c.run(); + } catch (Throwable ex) { + Exceptions.throwIfFatal(e); + cf.completeExceptionally(ex); + return; + } + cf.complete(null); + }); + return cf; + } + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectorTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectorTest.java index c73b80d2a4..b4268986fa 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectorTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectorTest.java @@ -57,6 +57,6 @@ public void finishCrash() throws Throwable { .collect(Collectors.toList()) .test() .awaitDone(5, TimeUnit.SECONDS) - .assertFailure(TestException.class); + .assertFailure(TestException.class, List.of()); } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterableTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterableTest.java new file mode 100644 index 0000000000..bf6d48bb23 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterableTest.java @@ -0,0 +1,124 @@ +/* + * 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.internal.operators.streamable; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.util.*; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.exceptions.TestException; + +public class StreamableConcatIterableTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Streamable.concat(List.of(Streamable.range(1, 5), Streamable.range(6, 5), Streamable.range(11, 5))) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult( + 1, 2, 3, 4, 5, + 6, 7, 8, 9, 10, + 11, 12, 13, 14, 15 + ); + } + + @Test + public void crash() throws Throwable { + Streamable.concat(List.of(Streamable.range(1, 5), Streamable.error(new TestException()), Streamable.range(11, 5))) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class, + 1, 2, 3, 4, 5); + } + + @Test + public void none() throws Throwable { + Streamable.concat(List.of()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void solo() throws Throwable { + Streamable.concat(List.of(Streamable.range(1, 5))) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void nullStreamable() throws Throwable { + Streamable.concat(Arrays.asList(Streamable.range(1, 5), null)) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(NullPointerException.class, 1, 2, 3, 4, 5); + } + + @Test + public void lot() { + var n = 10_000; + List> sources = new ArrayList<>(); + for (int i = 0; i < n; i++) { + sources.add(Streamable.just(i)); + } + + var ts = Streamable.concat(sources) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertValueCount(n) + .assertNoErrors() + .assertComplete(); + + for (int i = 0; i < n; i++) { + assertEquals(i, ts.values().get(i)); + } + } + + @Test + public void finishCrash() throws Throwable { + Streamable.concat(List.of(Streamable.empty(), StreamableFailingFinish.MAIN_COMPLETES)) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void finishCrashDebug() throws Throwable { + withCachedExecutor(exec -> { + Streamable.concat(List.of(Streamable.empty(), StreamableFailingFinish.MAIN_COMPLETES)) + .test(exec) + .awaitDone(5, TimeUnit.MINUTES) + .assertFailure(TestException.class); + }); + } + + @Test + public void lotEmpties() { + var n = 10_000; + List> sources = new ArrayList<>(); + for (int i = 0; i < n; i++) { + sources.add(Streamable.empty()); + } + + Streamable.concat(sources) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFailingFinish.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFailingFinish.java index 6eae228565..4d77d03871 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFailingFinish.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFailingFinish.java @@ -24,7 +24,7 @@ public enum StreamableFailingFinish implements Streamable { NEVER(0), MAIN_FAILS(1), - MAIN_COMPLETES(1) + MAIN_COMPLETES(2) ; private final class StreamableFailingFinishStreamer implements Streamer { diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHideTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHideTest.java new file mode 100644 index 0000000000..720cd8a09b --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHideTest.java @@ -0,0 +1,42 @@ +/* + * 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.internal.operators.streamable; + +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.exceptions.TestException; + +public class StreamableHideTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Streamable.range(1, 5) + .hide() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void crash() throws Throwable { + Streamable.error(new TestException()) + .hide() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableUsingTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableUsingTest.java new file mode 100644 index 0000000000..b4e22b636e --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableUsingTest.java @@ -0,0 +1,90 @@ +/* + * 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.internal.operators.streamable; + +import static org.junit.jupiter.api.Assertions.*; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.exceptions.TestException; + +public class StreamableUsingTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + var resource = new AtomicReference(); + Streamable.using(() -> 2, v -> Streamable.range(v, 5), resource::set) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(2, 3, 4, 5, 6); + + assertEquals(2, resource.get(), "resource cleanup mismatch"); + } + + @Test + public void nulLResourceAllowed() throws Throwable { + var resource = new AtomicReference(3); + Streamable.using(() -> null, v -> Streamable.range(v != null ? v : 0, 5), resource::set) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(0, 1, 2, 3, 4); + + assertNull(resource.get(), "resource cleanup mismatch"); + } + + @Test + public void resourceCrash() throws Throwable { + var resource = new AtomicReference(); + Streamable.using(() -> { throw new TestException(); }, v -> Streamable.range(v, 5), resource::set) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + + assertNull(resource.get(), "resource is not null?"); + } + + @Test + public void resourceMapperNull() throws Throwable { + var resource = new AtomicReference(); + Streamable.using(() -> 3, _ -> null, resource::set) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(NullPointerException.class); + + assertEquals(3, resource.get(), "resource cleanup mismatch"); + } + + @Test + public void resourceMapperCrash() throws Throwable { + var resource = new AtomicReference(); + Streamable.using(() -> 4, _ -> { throw new TestException(); }, resource::set) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + + assertEquals(4, resource.get(), "resource cleanup mismatch"); + } + + @Test + public void resourceCleanerCrash() throws Throwable { + Streamable.using(() -> 5, v -> Streamable.range(v, 5), _ -> { throw new TestException(); }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class, 5, 6, 7, 8, 9); + } +}