Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fixes onClose behaviour to not error on shutdown #833

Merged
merged 2 commits into from May 11, 2020
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
26 changes: 15 additions & 11 deletions rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java
Expand Up @@ -50,7 +50,6 @@
import io.rsocket.keepalive.KeepAliveSupport;
import io.rsocket.lease.RequesterLeaseHandler;
import java.nio.channels.ClosedChannelException;
import java.util.concurrent.CancellationException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.Consumer;
Expand Down Expand Up @@ -137,10 +136,7 @@ class RSocketRequester implements RSocket {
// DO NOT Change the order here. The Send processor must be subscribed to before receiving
this.sendProcessor = new UnboundedProcessor<>();

connection
.onClose()
.or(onClose)
.subscribe(null, this::tryTerminateOnConnectionError, this::tryTerminateOnConnectionClose);
connection.onClose().subscribe(null, this::tryTerminateOnConnectionError, this::tryShutdown);
connection.send(sendProcessor).subscribe(null, this::handleSendProcessorError);

connection.receive().subscribe(this::handleIncomingFrames, e -> {});
Expand Down Expand Up @@ -188,7 +184,7 @@ public double availability() {

@Override
public void dispose() {
tryTerminate(() -> new CancellationException("Disposed"));
tryShutdown();
}

@Override
Expand Down Expand Up @@ -708,10 +704,6 @@ private void tryTerminateOnConnectionError(Throwable e) {
tryTerminate(() -> e);
}

private void tryTerminateOnConnectionClose() {
tryTerminate(() -> CLOSED_CHANNEL_EXCEPTION);
}

private void tryTerminateOnZeroError(ByteBuf errorFrame) {
tryTerminate(() -> Exceptions.from(0, errorFrame));
}
Expand All @@ -725,6 +717,14 @@ private void tryTerminate(Supplier<Throwable> errorSupplier) {
}
}

private void tryShutdown() {
if (terminationError == null) {
if (TERMINATION_ERROR.compareAndSet(this, null, CLOSED_CHANNEL_EXCEPTION)) {
terminate(CLOSED_CHANNEL_EXCEPTION);
}
}
}

private void terminate(Throwable e) {
connection.dispose();
leaseHandler.dispose();
Expand Down Expand Up @@ -760,7 +760,11 @@ private void terminate(Throwable e) {
senders.clear();
receivers.clear();
sendProcessor.dispose();
onClose.onError(e);
if (e == CLOSED_CHANNEL_EXCEPTION) {
onClose.onComplete();
} else {
onClose.onError(e);
}
}

private void handleSendProcessorError(Throwable t) {
Expand Down
Expand Up @@ -75,7 +75,6 @@ class RSocketResponder implements RSocket {
private final PayloadDecoder payloadDecoder;
private final ResponderLeaseHandler leaseHandler;
private final Disposable leaseHandlerDisposable;
private final MonoProcessor<Void> onClose;

private volatile Throwable terminationError;
private static final AtomicReferenceFieldUpdater<RSocketResponder, Throwable> TERMINATION_ERROR =
Expand Down Expand Up @@ -110,7 +109,6 @@ class RSocketResponder implements RSocket {
this.leaseHandler = leaseHandler;
this.sendingSubscriptions = new SynchronizedIntObjectHashMap<>();
this.channelProcessors = new SynchronizedIntObjectHashMap<>();
this.onClose = MonoProcessor.create();

// DO NOT Change the order here. The Send processor must be subscribed to before receiving
// connections
Expand All @@ -123,7 +121,6 @@ class RSocketResponder implements RSocket {

this.connection
.onClose()
.or(onClose)
.subscribe(null, this::tryTerminateOnConnectionError, this::tryTerminateOnConnectionClose);
}

Expand Down Expand Up @@ -256,12 +253,12 @@ public void dispose() {

@Override
public boolean isDisposed() {
return onClose.isDisposed();
return connection.isDisposed();
}

@Override
public Mono<Void> onClose() {
return onClose;
return connection.onClose();
}

private void cleanup(Throwable e) {
Expand Down
33 changes: 33 additions & 0 deletions rsocket-core/src/test/java/io/rsocket/core/RSocketTest.java
Expand Up @@ -41,6 +41,8 @@
import org.junit.runner.Description;
import org.junit.runners.model.Statement;
import org.reactivestreams.Publisher;
import reactor.core.Disposable;
import reactor.core.Disposables;
import reactor.core.publisher.DirectProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
Expand All @@ -51,6 +53,34 @@ public class RSocketTest {

@Rule public final SocketRule rule = new SocketRule();

@Test
public void rsocketDisposalShouldEndupWithNoErrorsOnClose() {
RSocket requestHandlingRSocket =
new RSocket() {
final Disposable disposable = Disposables.single();

@Override
public void dispose() {
disposable.dispose();
}

@Override
public boolean isDisposed() {
return disposable.isDisposed();
}
};
rule.setRequestAcceptor(requestHandlingRSocket);
rule.crs
.onClose()
.as(StepVerifier::create)
.expectSubscription()
.then(rule.crs::dispose)
.expectComplete()
.verify(Duration.ofMillis(100));

Assertions.assertThat(requestHandlingRSocket.isDisposed()).isTrue();
}

@Test(timeout = 2_000)
public void testRequestReplyNoError() {
StepVerifier.create(rule.crs.requestResponse(DefaultPayload.create("hello")))
Expand Down Expand Up @@ -413,6 +443,9 @@ protected void init() {
LocalDuplexConnection clientConnection =
new LocalDuplexConnection("client", allocator, serverProcessor, clientProcessor);

clientConnection.onClose().doFinally(__ -> serverConnection.dispose()).subscribe();
serverConnection.onClose().doFinally(__ -> clientConnection.dispose()).subscribe();

requestAcceptor =
null != requestAcceptor
? requestAcceptor
Expand Down