In 1.0.0-RC7 it seems like for onClose() to be subscribable, we have to do something with the "error" resulting from the dispose() that previously was not required. This is a passing test written in RSocketTest that demonstrates what seems to be required:
@Test
public void disposeAndOnClose() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
rule.crs.onClose()
.map(v -> 1)
.onErrorReturn(1) // I only care to know that it was closed, not why
.subscribe(closed -> latch.countDown());
rule.crs.dispose();
assertThat(latch.await(1, TimeUnit.SECONDS)).isTrue();
}Otherwise you get the exception:
reactor.core.Exceptions$ErrorCallbackNotImplemented: java.util.concurrent.CancellationException: Disposed
Caused by: java.util.concurrent.CancellationException: Disposed
at io.rsocket.core.RSocketRequester.lambda$dispose$1(RSocketRequester.java:188)
at io.rsocket.core.RSocketRequester.tryTerminate(RSocketRequester.java:690)
at io.rsocket.core.RSocketRequester.dispose(RSocketRequester.java:188)
I did the weird map(v -> 1).onErrorReturn(1) thing because for the latch to count down we need a non-empty Mono<Void>, but paradoxically cannot create a Mono<Void> with null, the only valid value for Void. I'm probably just not seeing how to do this more idiomatically.
In 1.0.0-RC7 it seems like for
onClose()to be subscribable, we have to do something with the "error" resulting from thedispose()that previously was not required. This is a passing test written inRSocketTestthat demonstrates what seems to be required:Otherwise you get the exception:
I did the weird
map(v -> 1).onErrorReturn(1)thing because for the latch to count down we need a non-emptyMono<Void>, but paradoxically cannot create aMono<Void>with null, the only valid value forVoid. I'm probably just not seeing how to do this more idiomatically.