Skip to content
This repository was archived by the owner on Jun 21, 2023. It is now read-only.

Commit 578f112

Browse files
artem-vartem-v
authored
Merge pull request #174 from scalecube/get-rid-of-connection-closed-errors
Renamings, fixed WebsocketGatewayAcceptor (added filtering of reactor…
2 parents 1e1e463 + e16bd15 commit 578f112

File tree

8 files changed

+29
-18
lines changed

8 files changed

+29
-18
lines changed

‎services-gateway-netty/src/main/java/io/scalecube/services/gateway/ws/WebsocketGatewayAcceptor.java‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
import static io.scalecube.services.gateway.ws.GatewayMessages.validateSidOnSession;
1111

1212
import io.netty.buffer.ByteBuf;
13+
import io.netty.buffer.Unpooled;
1314
import io.netty.handler.codec.http.HttpHeaders;
1415
import io.scalecube.services.ServiceCall;
1516
import io.scalecube.services.api.ServiceMessage;
@@ -35,6 +36,7 @@
3536
import reactor.core.publisher.Flux;
3637
import reactor.core.publisher.Mono;
3738
import reactor.netty.DisposableChannel;
39+
import reactor.netty.channel.AbortedException;
3840
import reactor.netty.http.server.HttpServerRequest;
3941
import reactor.netty.http.server.HttpServerResponse;
4042
import reactor.netty.http.websocket.WebsocketInbound;
@@ -127,16 +129,25 @@ private Mono<Void> onConnect(WebsocketGatewaySession session) {
127129

128130
session
129131
.receive()
130-
.doOnError(th -> gatewayHandler.onSessionError(session, th))
131132
.subscribe(
132133
byteBuf -> {
134+
if (byteBuf == Unpooled.EMPTY_BUFFER) {
135+
return;
136+
}
137+
133138
if (!byteBuf.isReadable()) {
134139
ReferenceCountUtil.safestRelease(byteBuf);
135140
return;
136141
}
142+
137143
Mono.deferContextual(context -> onRequest(session, byteBuf, (Context) context))
138144
.contextWrite(context -> gatewayHandler.onRequest(session, byteBuf, context))
139145
.subscribe();
146+
},
147+
th -> {
148+
if (!(th instanceof AbortedException)) {
149+
gatewayHandler.onSessionError(session, th);
150+
}
140151
});
141152

142153
return session.onClose(() -> gatewayHandler.onSessionClose(session));

services-gateway-tests/src/test/java/io/scalecube/services/gateway/rsocket/RsocketClientConnectionTest.java renamed to services-gateway-tests/src/test/java/io/scalecube/services/gateway/rsocket/RSocketClientConnectionTest.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@
3535
import reactor.core.publisher.Mono;
3636
import reactor.test.StepVerifier;
3737

38-
class RsocketClientConnectionTest extends BaseTest {
38+
class RSocketClientConnectionTest extends BaseTest {
3939

4040
public static final GatewayClientCodec<Payload> CLIENT_CODEC =
4141
GatewayClientTransports.RSOCKET_CLIENT_CODEC;

services-gateway-tests/src/test/java/io/scalecube/services/gateway/rsocket/RsocketClientErrorMapperTest.java renamed to services-gateway-tests/src/test/java/io/scalecube/services/gateway/rsocket/RSocketClientErrorMapperTest.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,11 +13,11 @@
1313
import org.junit.jupiter.api.extension.RegisterExtension;
1414
import reactor.test.StepVerifier;
1515

16-
class RsocketClientErrorMapperTest extends BaseTest {
16+
class RSocketClientErrorMapperTest extends BaseTest {
1717

1818
@RegisterExtension
19-
static RsocketGatewayExtension extension =
20-
new RsocketGatewayExtension(
19+
static RSocketGatewayExtension extension =
20+
new RSocketGatewayExtension(
2121
ServiceInfo.fromServiceInstance(new ErrorServiceImpl())
2222
.errorMapper(ERROR_MAPPER)
2323
.build());

services-gateway-tests/src/test/java/io/scalecube/services/gateway/rsocket/RsocketGatewayExtension.java renamed to services-gateway-tests/src/test/java/io/scalecube/services/gateway/rsocket/RSocketGatewayExtension.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,15 +4,15 @@
44
import io.scalecube.services.gateway.AbstractGatewayExtension;
55
import io.scalecube.services.gateway.transport.GatewayClientTransports;
66

7-
class RsocketGatewayExtension extends AbstractGatewayExtension {
7+
class RSocketGatewayExtension extends AbstractGatewayExtension {
88

99
private static final String GATEWAY_ALIAS_NAME = "rsws";
1010

11-
RsocketGatewayExtension(Object serviceInstance) {
11+
RSocketGatewayExtension(Object serviceInstance) {
1212
this(ServiceInfo.fromServiceInstance(serviceInstance).build());
1313
}
1414

15-
RsocketGatewayExtension(ServiceInfo serviceInfo) {
15+
RSocketGatewayExtension(ServiceInfo serviceInfo) {
1616
super(
1717
serviceInfo,
1818
opts -> new RSocketGateway(opts.id(GATEWAY_ALIAS_NAME)),

‎services-gateway-tests/src/test/java/io/scalecube/services/gateway/rsocket/RSocketGatewayTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ class RSocketGatewayTest extends BaseTest {
2626
private static final Duration TIMEOUT = Duration.ofSeconds(3);
2727

2828
@RegisterExtension
29-
static RsocketGatewayExtension extension = new RsocketGatewayExtension(new GreetingServiceImpl());
29+
static RSocketGatewayExtension extension = new RSocketGatewayExtension(new GreetingServiceImpl());
3030

3131
private GreetingService service;
3232

services-gateway-tests/src/test/java/io/scalecube/services/gateway/rsocket/RsocketLocalGatewayErrorMapperTest.java renamed to services-gateway-tests/src/test/java/io/scalecube/services/gateway/rsocket/RSocketLocalGatewayErrorMapperTest.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,11 +13,11 @@
1313
import org.junit.jupiter.api.extension.RegisterExtension;
1414
import reactor.test.StepVerifier;
1515

16-
class RsocketLocalGatewayErrorMapperTest extends BaseTest {
16+
class RSocketLocalGatewayErrorMapperTest extends BaseTest {
1717

1818
@RegisterExtension
19-
static RsocketLocalGatewayExtension extension =
20-
new RsocketLocalGatewayExtension(
19+
static RSocketLocalGatewayExtension extension =
20+
new RSocketLocalGatewayExtension(
2121
ServiceInfo.fromServiceInstance(new ErrorServiceImpl()).errorMapper(ERROR_MAPPER).build(),
2222
opts ->
2323
new RSocketGateway(opts.call(opts.call().errorMapper(ERROR_MAPPER)), ERROR_MAPPER));

services-gateway-tests/src/test/java/io/scalecube/services/gateway/rsocket/RsocketLocalGatewayExtension.java renamed to services-gateway-tests/src/test/java/io/scalecube/services/gateway/rsocket/RSocketLocalGatewayExtension.java

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -6,19 +6,19 @@
66
import io.scalecube.services.gateway.transport.GatewayClientTransports;
77
import java.util.function.Function;
88

9-
class RsocketLocalGatewayExtension extends AbstractLocalGatewayExtension {
9+
class RSocketLocalGatewayExtension extends AbstractLocalGatewayExtension {
1010

1111
private static final String GATEWAY_ALIAS_NAME = "rsws";
1212

13-
RsocketLocalGatewayExtension(Object serviceInstance) {
13+
RSocketLocalGatewayExtension(Object serviceInstance) {
1414
this(ServiceInfo.fromServiceInstance(serviceInstance).build());
1515
}
1616

17-
RsocketLocalGatewayExtension(ServiceInfo serviceInfo) {
17+
RSocketLocalGatewayExtension(ServiceInfo serviceInfo) {
1818
this(serviceInfo, RSocketGateway::new);
1919
}
2020

21-
RsocketLocalGatewayExtension(
21+
RSocketLocalGatewayExtension(
2222
ServiceInfo serviceInfo, Function<GatewayOptions, RSocketGateway> gatewaySupplier) {
2323
super(
2424
serviceInfo,

‎services-gateway-tests/src/test/java/io/scalecube/services/gateway/rsocket/RSocketLocalGatewayTest.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,8 @@ class RSocketLocalGatewayTest extends BaseTest {
2525
private static final Duration TIMEOUT = Duration.ofSeconds(3);
2626

2727
@RegisterExtension
28-
static RsocketLocalGatewayExtension extension =
29-
new RsocketLocalGatewayExtension(new GreetingServiceImpl());
28+
static RSocketLocalGatewayExtension extension =
29+
new RSocketLocalGatewayExtension(new GreetingServiceImpl());
3030

3131
private GreetingService service;
3232

0 commit comments

Comments
 (0)