Refactor subscriptions

Author: oliemansm

gradle.properties

@@ -10,7 +10,7 @@ PROJECT_DEV_ID = apottere
 PROJECT_DEV_NAME = Andrew Potter
 
 LIB_GRAPHQL_JAVA_VER = 13.0
-LIB_JACKSON_VER = 2.9.9
+LIB_JACKSON_VER = 2.10.0
 
 SOURCE_COMPATIBILITY = 1.8
 TARGET_COMPATIBILITY = 1.8

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

@@ -0,0 +1,27 @@
+package graphql.kickstart.execution.subscriptions;
+
+import java.util.concurrent.atomic.AtomicReference;
+import org.reactivestreams.Subscription;
+
+public class AtomicSubscriptionSubscription {
+
+  private final AtomicReference<Subscription> reference = new AtomicReference<>(null);
+
+  public void set(Subscription subscription) {
+    if (reference.get() != null) {
+      throw new IllegalStateException("Cannot overwrite subscription!");
+    }
+
+    reference.set(subscription);
+  }
+
+  public Subscription get() {
+    Subscription subscription = reference.get();
+    if (subscription == null) {
+      throw new IllegalStateException("Subscription has not been initialized yet!");
+    }
+
+    return subscription;
+  }
+
+}

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

@@ -0,0 +1,102 @@
+package graphql.kickstart.execution.subscriptions;
+
+import graphql.ExecutionResult;
+import graphql.execution.reactive.SingleSubscriberPublisher;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscription;
+
+@Slf4j
+@RequiredArgsConstructor
+public class DefaultSubscriptionSession implements SubscriptionSession {
+
+  @Getter
+  private final GraphQLSubscriptionMapper mapper;
+  private SingleSubscriberPublisher<String> publisher = new SingleSubscriberPublisher<>();
+  private SessionSubscriptions subscriptions = new SessionSubscriptions();
+
+  @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));
+  }
+
+  @Override
+  public void subscribe(String id, Publisher<ExecutionResult> dataPublisher) {
+    dataPublisher.subscribe(new SessionSubscriber(this, id));
+  }
+
+  @Override
+  public void add(String id, Subscription subscription) {
+    subscriptions.add(id, subscription);
+  }
+
+  @Override
+  public void unsubscribe(String id) {
+    subscriptions.cancel(id);
+  }
+
+  @Override
+  public void sendDataMessage(String id, Object payload) {
+    send(mapper.serialize(payload));
+  }
+
+  @Override
+  public void sendErrorMessage(String id) {
+
+  }
+
+  @Override
+  public void sendCompleteMessage(String id) {
+
+  }
+
+  @Override
+  public void close(String reason) {
+    publisher.noMoreData();
+  }
+
+  @Override
+  public Map<String, Object> getUserProperties() {
+    return new HashMap<>();
+  }
+
+  @Override
+  public boolean isOpen() {
+    return true;
+  }
+
+  @Override
+  public String getId() {
+    return null;
+  }
+
+  @Override
+  public SessionSubscriptions getSubscriptions() {
+    return subscriptions;
+  }
+
+  @Override
+  public Object unwrap() {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public Publisher<String> getPublisher() {
+    return publisher;
+  }
+
+}

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

@@ -0,0 +1,10 @@
+package graphql.kickstart.execution.subscriptions;
+
+import graphql.kickstart.execution.GraphQLRequest;
+import graphql.kickstart.execution.input.GraphQLSingleInvocationInput;
+
+public interface GraphQLSubscriptionInvocationInputFactory {
+
+  GraphQLSingleInvocationInput create(GraphQLRequest graphQLRequest, SubscriptionSession session);
+
+}

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

@@ -0,0 +1,39 @@
+package graphql.kickstart.execution.subscriptions;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import graphql.ExecutionResult;
+import graphql.kickstart.execution.GraphQLObjectMapper;
+import graphql.kickstart.execution.GraphQLRequest;
+import java.util.Map;
+import lombok.RequiredArgsConstructor;
+
+@RequiredArgsConstructor
+public class GraphQLSubscriptionMapper {
+
+  private final GraphQLObjectMapper graphQLObjectMapper;
+
+  public GraphQLRequest readGraphQLRequest(Object payload) {
+    return graphQLObjectMapper.getJacksonMapper().convertValue(payload, GraphQLRequest.class);
+  }
+
+  public ExecutionResult sanitizeErrors(ExecutionResult executionResult) {
+    return graphQLObjectMapper.sanitizeErrors(executionResult);
+  }
+
+  public boolean areErrorsPresent(ExecutionResult executionResult) {
+    return graphQLObjectMapper.areErrorsPresent(executionResult);
+  }
+
+  public Map<String, Object> convertSanitizedExecutionResult(ExecutionResult executionResult) {
+    return graphQLObjectMapper.convertSanitizedExecutionResult(executionResult, false);
+  }
+
+  public String serialize(Object payload) {
+    try {
+      return graphQLObjectMapper.getJacksonMapper().writeValueAsString(payload);
+    } catch (JsonProcessingException e) {
+      return e.getMessage();
+    }
+  }
+
+}

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

@@ -0,0 +1,50 @@
+package graphql.kickstart.execution.subscriptions;
+
+import graphql.ExecutionResult;
+import java.util.HashMap;
+import java.util.Map;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+@Slf4j
+@RequiredArgsConstructor
+class SessionSubscriber implements Subscriber<ExecutionResult> {
+
+  private final SubscriptionSession session;
+  private final String id;
+  private AtomicSubscriptionSubscription subscriptionReference = new AtomicSubscriptionSubscription();
+
+  @Override
+  public void onSubscribe(Subscription subscription) {
+    log.info("Subscribe to execution result: {}", subscription);
+    subscriptionReference.set(subscription);
+    subscriptionReference.get().request(1);
+
+    session.add(id, subscriptionReference.get());
+  }
+
+  @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);
+    subscriptionReference.get().request(1);
+  }
+
+  @Override
+  public void onError(Throwable throwable) {
+    log.error("Subscription error", throwable);
+    session.unsubscribe(id);
+    session.sendErrorMessage(id);
+  }
+
+  @Override
+  public void onComplete() {
+    session.unsubscribe(id);
+    session.sendCompleteMessage(id);
+  }
+
+}

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

@@ -0,0 +1,56 @@
+package graphql.kickstart.execution.subscriptions;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.reactivestreams.Subscription;
+
+/**
+ * @author Andrew Potter
+ */
+public class SessionSubscriptions {
+
+  private final Object lock = new Object();
+
+  private boolean closed = false;
+  private Map<String, Subscription> subscriptions = new ConcurrentHashMap<>();
+
+  public void add(Subscription subscription) {
+    add(getImplicitId(subscription), subscription);
+  }
+
+  public void add(String id, Subscription subscription) {
+    synchronized (lock) {
+      if (closed) {
+        throw new IllegalStateException("Websocket was already closed!");
+      }
+      subscriptions.put(id, subscription);
+    }
+  }
+
+  public void cancel(Subscription subscription) {
+    cancel(getImplicitId(subscription));
+  }
+
+  public void cancel(String id) {
+    Subscription subscription = subscriptions.remove(id);
+    if (subscription != null) {
+      subscription.cancel();
+    }
+  }
+
+  public void close() {
+    synchronized (lock) {
+      closed = true;
+      subscriptions.forEach((k, v) -> v.cancel());
+      subscriptions.clear();
+    }
+  }
+
+  private String getImplicitId(Subscription subscription) {
+    return String.valueOf(subscription.hashCode());
+  }
+
+  public int getSubscriptionCount() {
+    return subscriptions.size();
+  }
+}

graphql-java-kickstart/src/main/java/graphql/kickstart/execution/subscription/SubscriptionConnectionListener.java renamed to graphql-java-kickstart/src/main/java/graphql/kickstart/execution/subscriptions/SubscriptionConnectionListener.java

@@ -1,4 +1,4 @@
-package graphql.kickstart.execution.subscription;
+package graphql.kickstart.execution.subscriptions;
 
 /**
  * Marker interface

graphql-java-kickstart/src/main/java/graphql/kickstart/execution/subscription/SubscriptionException.java renamed to graphql-java-kickstart/src/main/java/graphql/kickstart/execution/subscriptions/SubscriptionException.java

@@ -1,4 +1,4 @@
-package graphql.kickstart.execution.subscription;
+package graphql.kickstart.execution.subscriptions;
 
 public class SubscriptionException extends Exception {
 

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

@@ -0,0 +1,5 @@
+package graphql.kickstart.execution.subscriptions;
+
+public class SubscriptionHandler {
+
+}