Skip to content

Commit 1849d08

Browse files
robertroeseryschimke
robertroeser
authored and
yschimke
committed
Server Connection Closing on Exception (rsocket#442)
* server connection was being closed when the server received an error frame * added test to test handling a request after a server receives an exception
1 parent 3d58344 commit 1849d08

4 files changed

Lines changed: 114 additions & 82 deletions

File tree

‎rsocket-core/src/main/java/io/rsocket/RSocketClient.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -389,7 +389,7 @@ public Frame apply(Payload payload) {
389389
.doOnError(
390390
t -> {
391391
errorConsumer.accept(t);
392-
receiver.cancel();
392+
receiver.dispose();
393393
})
394394
.subscribe();
395395
} else {

‎rsocket-core/src/main/java/io/rsocket/RSocketServer.java‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,14 @@ class RSocketServer implements RSocket {
7373
this.receiveDisposable =
7474
connection
7575
.receive()
76-
.flatMapSequential(this::handleFrame)
76+
.flatMapSequential(
77+
frame ->
78+
handleFrame(frame)
79+
.onErrorResume(
80+
t -> {
81+
errorConsumer.accept(t);
82+
return Mono.empty();
83+
}))
7784
.doOnError(errorConsumer)
7885
.then()
7986
.subscribe();

‎rsocket-core/src/main/java/io/rsocket/internal/SwitchTransform.java‎

Lines changed: 62 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -10,71 +10,71 @@
1010
import reactor.core.publisher.Operators;
1111

1212
public final class SwitchTransform<T, R> extends Flux<R> {
13-
14-
final Publisher<? extends T> source;
15-
final BiFunction<T, Flux<T>, Publisher<? extends R>> transformer;
16-
17-
public SwitchTransform(
18-
Publisher<? extends T> source, BiFunction<T, Flux<T>, Publisher<? extends R>> transformer) {
19-
this.source = Objects.requireNonNull(source, "source");
20-
this.transformer = Objects.requireNonNull(transformer, "transformer");
21-
}
22-
23-
@Override
24-
public void subscribe(CoreSubscriber<? super R> actual) {
25-
Flux.from(source).subscribe(new SwitchTransformSubscriber<>(actual, transformer));
26-
}
27-
28-
static final class SwitchTransformSubscriber<T, R> implements CoreSubscriber<T> {
29-
@SuppressWarnings("rawtypes")
30-
static final AtomicIntegerFieldUpdater<SwitchTransformSubscriber> ONCE =
31-
AtomicIntegerFieldUpdater.newUpdater(SwitchTransformSubscriber.class, "once");
32-
33-
final CoreSubscriber<? super R> actual;
13+
14+
final Publisher<? extends T> source;
3415
final BiFunction<T, Flux<T>, Publisher<? extends R>> transformer;
35-
final UnboundedProcessor<T> processor = new UnboundedProcessor<>();
36-
Subscription s;
37-
volatile int once;
38-
39-
SwitchTransformSubscriber(
40-
CoreSubscriber<? super R> actual,
41-
BiFunction<T, Flux<T>, Publisher<? extends R>> transformer) {
42-
this.actual = actual;
43-
this.transformer = transformer;
16+
17+
public SwitchTransform(
18+
Publisher<? extends T> source, BiFunction<T, Flux<T>, Publisher<? extends R>> transformer) {
19+
this.source = Objects.requireNonNull(source, "source");
20+
this.transformer = Objects.requireNonNull(transformer, "transformer");
4421
}
45-
22+
4623
@Override
47-
public void onSubscribe(Subscription s) {
48-
if (Operators.validate(this.s, s)) {
49-
this.s = s;
50-
processor.onSubscribe(s);
51-
}
24+
public void subscribe(CoreSubscriber<? super R> actual) {
25+
Flux.from(source).subscribe(new SwitchTransformSubscriber<>(actual, transformer));
5226
}
53-
54-
@Override
55-
public void onNext(T t) {
56-
if (once == 0 && ONCE.compareAndSet(this, 0, 1)) {
57-
try {
58-
Publisher<? extends R> result =
59-
Objects.requireNonNull(
60-
transformer.apply(t, processor), "The transformer returned a null value");
61-
Flux.from(result).subscribe(actual);
62-
} catch (Throwable e) {
63-
onError(Operators.onOperatorError(s, e, t, actual.currentContext()));
64-
return;
27+
28+
static final class SwitchTransformSubscriber<T, R> implements CoreSubscriber<T> {
29+
@SuppressWarnings("rawtypes")
30+
static final AtomicIntegerFieldUpdater<SwitchTransformSubscriber> ONCE =
31+
AtomicIntegerFieldUpdater.newUpdater(SwitchTransformSubscriber.class, "once");
32+
33+
final CoreSubscriber<? super R> actual;
34+
final BiFunction<T, Flux<T>, Publisher<? extends R>> transformer;
35+
final UnboundedProcessor<T> processor = new UnboundedProcessor<>();
36+
Subscription s;
37+
volatile int once;
38+
39+
SwitchTransformSubscriber(
40+
CoreSubscriber<? super R> actual,
41+
BiFunction<T, Flux<T>, Publisher<? extends R>> transformer) {
42+
this.actual = actual;
43+
this.transformer = transformer;
44+
}
45+
46+
@Override
47+
public void onSubscribe(Subscription s) {
48+
if (Operators.validate(this.s, s)) {
49+
this.s = s;
50+
processor.onSubscribe(s);
51+
}
52+
}
53+
54+
@Override
55+
public void onNext(T t) {
56+
if (once == 0 && ONCE.compareAndSet(this, 0, 1)) {
57+
try {
58+
Publisher<? extends R> result =
59+
Objects.requireNonNull(
60+
transformer.apply(t, processor), "The transformer returned a null value");
61+
Flux.from(result).subscribe(actual);
62+
} catch (Throwable e) {
63+
onError(Operators.onOperatorError(s, e, t, actual.currentContext()));
64+
return;
65+
}
66+
}
67+
processor.onNext(t);
68+
}
69+
70+
@Override
71+
public void onError(Throwable t) {
72+
processor.onError(t);
73+
}
74+
75+
@Override
76+
public void onComplete() {
77+
processor.onComplete();
6578
}
66-
}
67-
processor.onNext(t);
68-
}
69-
70-
@Override
71-
public void onError(Throwable t) {
72-
processor.onError(t);
73-
}
74-
75-
@Override
76-
public void onComplete() {
77-
processor.onComplete();
7879
}
79-
}
80-
}
80+
}

‎rsocket-examples/src/test/java/io/rsocket/integration/IntegrationTest.java‎

Lines changed: 43 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,6 @@
1616

1717
package io.rsocket.integration;
1818

19-
import static org.hamcrest.Matchers.is;
20-
import static org.junit.Assert.assertThat;
21-
import static org.junit.Assert.assertTrue;
22-
import static org.mockito.ArgumentMatchers.any;
23-
import static org.mockito.Mockito.verify;
24-
import static org.mockito.Mockito.verifyNoMoreInteractions;
25-
2619
import io.rsocket.AbstractRSocket;
2720
import io.rsocket.Payload;
2821
import io.rsocket.RSocket;
@@ -35,28 +28,33 @@
3528
import io.rsocket.transport.netty.server.TcpServerTransport;
3629
import io.rsocket.util.PayloadImpl;
3730
import io.rsocket.util.RSocketProxy;
38-
import java.util.concurrent.CountDownLatch;
39-
import java.util.concurrent.atomic.AtomicInteger;
4031
import org.junit.After;
32+
import org.junit.Assert;
4133
import org.junit.Before;
4234
import org.junit.Test;
35+
import org.reactivestreams.Publisher;
4336
import org.reactivestreams.Subscriber;
4437
import reactor.core.publisher.Flux;
4538
import reactor.core.publisher.Mono;
4639

47-
public class IntegrationTest {
40+
import java.util.concurrent.CountDownLatch;
41+
import java.util.concurrent.atomic.AtomicInteger;
4842

49-
private NettyContextCloseable server;
50-
private RSocket client;
51-
private AtomicInteger requestCount;
52-
private CountDownLatch disconnectionCounter;
53-
public static volatile boolean calledClient = false;
54-
public static volatile boolean calledServer = false;
55-
public static volatile boolean calledFrame = false;
43+
import static org.hamcrest.Matchers.is;
44+
import static org.junit.Assert.assertThat;
45+
import static org.junit.Assert.assertTrue;
46+
import static org.mockito.ArgumentMatchers.any;
47+
import static org.mockito.Mockito.verify;
48+
import static org.mockito.Mockito.verifyNoMoreInteractions;
49+
50+
public class IntegrationTest {
5651

5752
private static final RSocketInterceptor clientPlugin;
5853
private static final RSocketInterceptor serverPlugin;
5954
private static final DuplexConnectionInterceptor connectionPlugin;
55+
public static volatile boolean calledClient = false;
56+
public static volatile boolean calledServer = false;
57+
public static volatile boolean calledFrame = false;
6058

6159
static {
6260
clientPlugin =
@@ -86,8 +84,15 @@ public Mono<Payload> requestResponse(Payload payload) {
8684
};
8785
}
8886

87+
private NettyContextCloseable server;
88+
private RSocket client;
89+
private AtomicInteger requestCount;
90+
private CountDownLatch disconnectionCounter;
91+
private AtomicInteger errorCount;
92+
8993
@Before
9094
public void startup() {
95+
errorCount = new AtomicInteger();
9196
requestCount = new AtomicInteger();
9297
disconnectionCounter = new CountDownLatch(1);
9398

@@ -97,6 +102,9 @@ public void startup() {
97102
RSocketFactory.receive()
98103
.addServerPlugin(serverPlugin)
99104
.addConnectionPlugin(connectionPlugin)
105+
.errorConsumer(t -> {
106+
errorCount.incrementAndGet();
107+
})
100108
.acceptor(
101109
(setup, sendingSocket) -> {
102110
sendingSocket
@@ -116,6 +124,11 @@ public Mono<Payload> requestResponse(Payload payload) {
116124
public Flux<Payload> requestStream(Payload payload) {
117125
return Flux.range(1, 10_000).map(i -> new PayloadImpl("data -> " + i));
118126
}
127+
128+
@Override
129+
public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
130+
return Flux.from(payloads);
131+
}
119132
});
120133
})
121134
.transport(serverTransport)
@@ -145,7 +158,7 @@ public void testRequest() {
145158
assertTrue(calledFrame);
146159
}
147160

148-
@Test
161+
@Test(timeout = 5_000L)
149162
public void testStream() {
150163
Subscriber<Payload> subscriber = TestSubscriber.createCancelling();
151164
client.requestStream(new PayloadImpl("start")).subscribe(subscriber);
@@ -159,4 +172,16 @@ public void testClose() throws InterruptedException {
159172
client.close().block();
160173
disconnectionCounter.await();
161174
}
175+
176+
@Test // (timeout = 5_000L)
177+
public void testCallRequestWithErrorAndThenRequest() {
178+
try {
179+
client.requestChannel(Mono.error(new Throwable())).blockLast();
180+
} catch (Throwable t) {
181+
}
182+
183+
Assert.assertEquals(1, errorCount.incrementAndGet());
184+
185+
testRequest();
186+
}
162187
}

0 commit comments

Comments
 (0)