Refactor subscriptions

Author: oliemansm

graphql-java-kickstart/src/main/java/graphql/kickstart/execution/subscriptions/DefaultSubscriptionSession.java

@@ -23,14 +23,12 @@ public class DefaultSubscriptionSession implements SubscriptionSession {
   @Override
   public void send(String message) {
     Objects.requireNonNull(message, "message is required");
-    log.info("Offer message: {}", message);
     publisher.offer(message);
   }
 
   @Override
   public void sendMessage(Object payload) {
     Objects.requireNonNull(payload, "payload is required");
-    log.info("Send message: {}", payload);
     send(mapper.serialize(payload));
   }
 
@@ -66,6 +64,8 @@ public void sendCompleteMessage(String id) {
 
   @Override
   public void close(String reason) {
+    log.debug("Closing subscription session {}", getId());
+    subscriptions.close();
     publisher.noMoreData();
   }
 
@@ -99,4 +99,9 @@ public Publisher<String> getPublisher() {
     return publisher;
   }
 
+  @Override
+  public String toString() {
+    return getId();
+  }
+
 }

graphql-java-kickstart/src/main/java/graphql/kickstart/execution/subscriptions/SessionSubscriber.java

@@ -18,7 +18,7 @@ class SessionSubscriber implements Subscriber<ExecutionResult> {
 
   @Override
   public void onSubscribe(Subscription subscription) {
-    log.info("Subscribe to execution result: {}", subscription);
+    log.debug("Subscribe to execution result: {}", subscription);
     subscriptionReference.set(subscription);
     subscriptionReference.get().request(1);
 
@@ -27,7 +27,6 @@ public void onSubscribe(Subscription subscription) {
 
   @Override
   public void onNext(ExecutionResult executionResult) {
-    log.info("Next execution result: {}", executionResult);
     Map<String, Object> result = new HashMap<>();
     result.put("data", executionResult.getData());
     session.sendDataMessage(id, result);

graphql-java-kickstart/src/main/java/graphql/kickstart/execution/subscriptions/apollo/SubscriptionConnectionInitCommand.java

@@ -4,14 +4,17 @@
 import graphql.kickstart.execution.subscriptions.apollo.OperationMessage.Type;
 import java.util.Collection;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 
+@Slf4j
 @RequiredArgsConstructor
 class SubscriptionConnectionInitCommand implements SubscriptionCommand {
 
   private final Collection<ApolloSubscriptionConnectionListener> connectionListeners;
 
   @Override
   public void apply(SubscriptionSession session, OperationMessage message) {
+    log.info("Apollo subscription connection init: {}", session);
     try {
       connectionListeners.forEach(it -> it.onConnect(session, message));
       session.sendMessage(new OperationMessage(Type.GQL_CONNECTION_ACK, message.getId(), null));

graphql-java-kickstart/src/main/java/graphql/kickstart/execution/subscriptions/apollo/SubscriptionStartCommand.java

@@ -13,7 +13,9 @@
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 
+@Slf4j
 @RequiredArgsConstructor
 class SubscriptionStartCommand implements SubscriptionCommand {
 
@@ -24,6 +26,7 @@ class SubscriptionStartCommand implements SubscriptionCommand {
 
   @Override
   public void apply(SubscriptionSession session, OperationMessage message) {
+    log.info("Apollo subscription start: {} --> {}", session, message.getPayload());
     connectionListeners.forEach(it -> it.onStart(session, message));
     CompletableFuture<ExecutionResult> executionResult = executeAsync(message.getPayload(), session);
     executionResult.thenAccept(result -> handleSubscriptionStart(session, message.getId(), result));