Skip to content

Commit 9d7d692

Browse files
osiosi
committed
Address comments I made on the PR
 - Create KeepAliveRunner once - Don't stop kee-alive on GQL_STOP
1 parent ceba707 commit 9d7d692

File tree

6 files changed

+53
-39
lines changed

6 files changed

+53
-39
lines changed

‎src/main/java/graphql/servlet/GraphQLWebsocketServlet.java‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -30,15 +30,9 @@ public class GraphQLWebsocketServlet extends Endpoint {
3030
private static final CloseReason ERROR_CLOSE_REASON = new CloseReason(CloseReason.CloseCodes.UNEXPECTED_CONDITION, "Internal Server Error");
3131
private static final CloseReason SHUTDOWN_CLOSE_REASON = new CloseReason(CloseReason.CloseCodes.UNEXPECTED_CONDITION, "Server Shut Down");
3232

33-
private static final List<SubscriptionProtocolFactory> subscriptionProtocolFactories = Collections.singletonList(new ApolloSubscriptionProtocolFactory());
34-
private static final SubscriptionProtocolFactory fallbackSubscriptionProtocolFactory = new FallbackSubscriptionProtocolFactory();
35-
private static final List<String> allSubscriptionProtocols;
36-
37-
static {
38-
allSubscriptionProtocols = Stream.concat(subscriptionProtocolFactories.stream(), Stream.of(fallbackSubscriptionProtocolFactory))
39-
.map(SubscriptionProtocolFactory::getProtocol)
40-
.collect(Collectors.toList());
41-
}
33+
private final List<SubscriptionProtocolFactory> subscriptionProtocolFactories;
34+
private final SubscriptionProtocolFactory fallbackSubscriptionProtocolFactory;
35+
private final List<String> allSubscriptionProtocols;
4236

4337
private final Map<Session, WsSessionSubscriptions> sessionSubscriptionCache = new HashMap<>();
4438
private final SubscriptionHandlerInput subscriptionHandlerInput;
@@ -52,6 +46,12 @@ public GraphQLWebsocketServlet(GraphQLQueryInvoker queryInvoker, GraphQLInvocati
5246

5347
public GraphQLWebsocketServlet(GraphQLQueryInvoker queryInvoker, GraphQLInvocationInputFactory invocationInputFactory, GraphQLObjectMapper graphQLObjectMapper, SubscriptionConnectionListener subscriptionConnectionListener) {
5448
this.subscriptionHandlerInput = new SubscriptionHandlerInput(invocationInputFactory, queryInvoker, graphQLObjectMapper, subscriptionConnectionListener);
49+
50+
subscriptionProtocolFactories = Collections.singletonList(new ApolloSubscriptionProtocolFactory(subscriptionHandlerInput));
51+
fallbackSubscriptionProtocolFactory = new FallbackSubscriptionProtocolFactory(subscriptionHandlerInput);
52+
allSubscriptionProtocols = Stream.concat(subscriptionProtocolFactories.stream(), Stream.of(fallbackSubscriptionProtocolFactory))
53+
.map(SubscriptionProtocolFactory::getProtocol)
54+
.collect(Collectors.toList());
5555
}
5656

5757
@Override
@@ -119,7 +119,7 @@ public void modifyHandshake(ServerEndpointConfig sec, HandshakeRequest request,
119119
}
120120

121121
SubscriptionProtocolFactory subscriptionProtocolFactory = getSubscriptionProtocolFactory(protocol);
122-
sec.getUserProperties().put(PROTOCOL_HANDLER_REQUEST_KEY, subscriptionProtocolFactory.createHandler(subscriptionHandlerInput));
122+
sec.getUserProperties().put(PROTOCOL_HANDLER_REQUEST_KEY, subscriptionProtocolFactory.createHandler());
123123

124124
if (request.getHeaders().get(HandshakeResponse.SEC_WEBSOCKET_ACCEPT) != null) {
125125
response.getHeaders().put(HandshakeResponse.SEC_WEBSOCKET_ACCEPT, allSubscriptionProtocols);
@@ -165,7 +165,7 @@ public boolean isShutDown() {
165165
return isShutDown.get();
166166
}
167167

168-
private static SubscriptionProtocolFactory getSubscriptionProtocolFactory(List<String> accept) {
168+
private SubscriptionProtocolFactory getSubscriptionProtocolFactory(List<String> accept) {
169169
for (String protocol : accept) {
170170
for (SubscriptionProtocolFactory subscriptionProtocolFactory : subscriptionProtocolFactories) {
171171
if (subscriptionProtocolFactory.getProtocol().equals(protocol)) {

‎src/main/java/graphql/servlet/internal/ApolloSubscriptionKeepAliveRunner.java‎

Lines changed: 20 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
import java.util.concurrent.Executors;
1010
import java.util.concurrent.Future;
1111
import java.util.concurrent.ScheduledExecutorService;
12+
import java.util.concurrent.ScheduledFuture;
1213
import java.util.concurrent.TimeUnit;
1314

1415
class ApolloSubscriptionKeepAliveRunner {
@@ -18,10 +19,10 @@ class ApolloSubscriptionKeepAliveRunner {
1819
private static final long KEEP_ALIVE_INTERVAL_SEC = 15;
1920
private static final int EXECUTOR_POOL_SIZE = 10;
2021

21-
private ScheduledExecutorService executor;
22-
private SubscriptionSender sender;
23-
private ApolloSubscriptionProtocolHandler.OperationMessage keepAliveMessage;
24-
private Map<Session, Future<?>> futures;
22+
private final ScheduledExecutorService executor;
23+
private final SubscriptionSender sender;
24+
private final ApolloSubscriptionProtocolHandler.OperationMessage keepAliveMessage;
25+
private final Map<Session, Future<?>> futures;
2526

2627
ApolloSubscriptionKeepAliveRunner(SubscriptionSender sender) {
2728
this.sender = sender;
@@ -31,22 +32,23 @@ class ApolloSubscriptionKeepAliveRunner {
3132
}
3233

3334
void keepAlive(Session session) {
34-
if (!futures.containsKey(session)) {
35-
Future<?> future = executor.scheduleAtFixedRate(() -> {
36-
try {
37-
if (session.isOpen()) {
38-
sender.send(session, keepAliveMessage);
39-
} else {
40-
LOG.warn("Session appears to be closed. Aborting keep alive");
41-
abort(session);
42-
}
43-
} catch (Throwable t) {
44-
LOG.error("Cannot send keep alive message. Aborting keep alive", t);
35+
futures.computeIfAbsent(session, this::startKeepAlive);
36+
}
37+
38+
private ScheduledFuture<?> startKeepAlive(Session session) {
39+
return executor.scheduleAtFixedRate(() -> {
40+
try {
41+
if (session.isOpen()) {
42+
sender.send(session, keepAliveMessage);
43+
} else {
44+
LOG.warn("Session appears to be closed. Aborting keep alive");
4545
abort(session);
4646
}
47-
}, 0, KEEP_ALIVE_INTERVAL_SEC, TimeUnit.SECONDS);
48-
futures.put(session, future);
49-
}
47+
} catch (Throwable t) {
48+
LOG.error("Cannot send keep alive message. Aborting keep alive", t);
49+
abort(session);
50+
}
51+
}, 0, KEEP_ALIVE_INTERVAL_SEC, TimeUnit.SECONDS);
5052
}
5153

5254
void abort(Session session) {

‎src/main/java/graphql/servlet/internal/ApolloSubscriptionProtocolFactory.java‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,12 +4,20 @@
44
* @author Andrew Potter
55
*/
66
public class ApolloSubscriptionProtocolFactory extends SubscriptionProtocolFactory {
7-
public ApolloSubscriptionProtocolFactory() {
7+
private final SubscriptionHandlerInput subscriptionHandlerInput;
8+
private final SubscriptionSender subscriptionSender;
9+
private final ApolloSubscriptionKeepAliveRunner keepAliveRunner;
10+
11+
public ApolloSubscriptionProtocolFactory(SubscriptionHandlerInput subscriptionHandlerInput) {
812
super("graphql-ws");
13+
this.subscriptionHandlerInput = subscriptionHandlerInput;
14+
subscriptionSender =
15+
new SubscriptionSender(subscriptionHandlerInput.getGraphQLObjectMapper().getJacksonMapper());
16+
keepAliveRunner = new ApolloSubscriptionKeepAliveRunner(subscriptionSender);
917
}
1018

1119
@Override
12-
public SubscriptionProtocolHandler createHandler(SubscriptionHandlerInput subscriptionHandlerInput) {
13-
return new ApolloSubscriptionProtocolHandler(subscriptionHandlerInput);
20+
public SubscriptionProtocolHandler createHandler() {
21+
return new ApolloSubscriptionProtocolHandler(subscriptionHandlerInput, subscriptionSender, keepAliveRunner);
1422
}
1523
}

‎src/main/java/graphql/servlet/internal/ApolloSubscriptionProtocolHandler.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -38,14 +38,16 @@ public class ApolloSubscriptionProtocolHandler extends SubscriptionProtocolHandl
3838
private final ApolloSubscriptionKeepAliveRunner keepAliveRunner;
3939
private final ApolloSubscriptionConnectionListener connectionListener;
4040

41-
public ApolloSubscriptionProtocolHandler(SubscriptionHandlerInput subscriptionHandlerInput) {
41+
public ApolloSubscriptionProtocolHandler(SubscriptionHandlerInput subscriptionHandlerInput,
42+
SubscriptionSender subscriptionSender,
43+
ApolloSubscriptionKeepAliveRunner keepAliveRunner) {
4244
this.input = subscriptionHandlerInput;
4345
this.connectionListener = subscriptionHandlerInput.getSubscriptionConnectionListener()
4446
.filter(ApolloSubscriptionConnectionListener.class::isInstance)
4547
.map(ApolloSubscriptionConnectionListener.class::cast)
4648
.orElse(new ApolloSubscriptionConnectionListener() {});
47-
this.sender = new SubscriptionSender(this.input.getGraphQLObjectMapper().getJacksonMapper());
48-
this.keepAliveRunner = new ApolloSubscriptionKeepAliveRunner(this.sender);
49+
this.sender = subscriptionSender;
50+
this.keepAliveRunner = keepAliveRunner;
4951
}
5052

5153
@Override
@@ -87,7 +89,6 @@ public void onMessage(HandshakeRequest request, Session session, WsSessionSubscr
8789
break;
8890

8991
case GQL_STOP:
90-
keepAliveRunner.abort(session);
9192
unsubscribe(subscriptions, message.id);
9293
break;
9394

‎src/main/java/graphql/servlet/internal/FallbackSubscriptionProtocolFactory.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,12 +4,15 @@
44
* @author Andrew Potter
55
*/
66
public class FallbackSubscriptionProtocolFactory extends SubscriptionProtocolFactory {
7-
public FallbackSubscriptionProtocolFactory() {
7+
private final SubscriptionHandlerInput subscriptionHandlerInput;
8+
9+
public FallbackSubscriptionProtocolFactory(SubscriptionHandlerInput subscriptionHandlerInput) {
810
super("");
11+
this.subscriptionHandlerInput = subscriptionHandlerInput;
912
}
1013

1114
@Override
12-
public SubscriptionProtocolHandler createHandler(SubscriptionHandlerInput subscriptionHandlerInput) {
15+
public SubscriptionProtocolHandler createHandler() {
1316
return new FallbackSubscriptionProtocolHandler(subscriptionHandlerInput);
1417
}
1518
}

‎src/main/java/graphql/servlet/internal/SubscriptionProtocolFactory.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,5 +14,5 @@ public String getProtocol() {
1414
return protocol;
1515
}
1616

17-
public abstract SubscriptionProtocolHandler createHandler(SubscriptionHandlerInput subscriptionHandlerInput);
17+
public abstract SubscriptionProtocolHandler createHandler();
1818
}

0 commit comments

Comments
 (0)