Skip to content

Commit 65819d3

Browse files
committed
Fix finish error propagation in StreamableUsing
Signed-off-by: myetcd <mytech@139.com>
1 parent 3d1247a commit 65819d3

2 files changed

Lines changed: 33 additions & 4 deletions

File tree

‎src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableUsing.java‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import io.reactivex.rxjava4.disposables.StreamerCancellation;
2222
import io.reactivex.rxjava4.exceptions.Exceptions;
2323
import io.reactivex.rxjava4.functions.*;
24+
import io.reactivex.rxjava4.internal.util.ExceptionHelper;
2425

2526
public record StreamableUsing<T, R>(
2627
Supplier<? extends R> resourceSupplier,
@@ -80,11 +81,14 @@ static final class UsingStreamer<T, R> implements Streamer<T> {
8081
cleanup = null;
8182
c.run();
8283
} catch (Throwable ex) {
83-
Exceptions.throwIfFatal(e);
84-
cf.completeExceptionally(ex);
85-
return;
84+
Exceptions.throwIfFatal(ex);
85+
e = ExceptionHelper.unwrapAndCombine(e, ex);
86+
}
87+
if (e != null) {
88+
cf.completeExceptionally(e);
89+
} else {
90+
cf.complete(null);
8691
}
87-
cf.complete(null);
8892
});
8993
return cf;
9094
}

‎src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableUsingTest.java‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -87,4 +87,29 @@ public void resourceCleanerCrash() throws Throwable {
8787
.awaitDone(5, TimeUnit.SECONDS)
8888
.assertFailure(TestException.class, 5, 6, 7, 8, 9);
8989
}
90+
91+
@Test
92+
public void upstreamFinishCrash() {
93+
var resource = new AtomicReference<Integer>();
94+
95+
Streamable.using(() -> 1, _ -> StreamableFailingFinish.MAIN_COMPLETES, resource::set)
96+
.test()
97+
.awaitDone(5, TimeUnit.SECONDS)
98+
.assertFailure(TestException.class)
99+
.assertError(e -> e.getMessage().equals("StreamableFailingFinish.finish()"));
100+
101+
assertEquals(1, resource.get(), "resource cleanup mismatch");
102+
}
103+
104+
@Test
105+
public void upstreamFinishAndResourceCleanerCrash() {
106+
var cleanerError = new TestException("resourceCleaner");
107+
108+
Streamable.using(() -> 1, _ -> StreamableFailingFinish.MAIN_COMPLETES, _ -> { throw cleanerError; })
109+
.test()
110+
.awaitDone(5, TimeUnit.SECONDS)
111+
.assertFailure(TestException.class)
112+
.assertError(e -> e.getMessage().equals("StreamableFailingFinish.finish()"))
113+
.assertError(e -> e.getSuppressed().length == 1 && e.getSuppressed()[0] == cleanerError);
114+
}
90115
}

0 commit comments

Comments
 (0)