Merge pull request #3512 from OpenFeign/graphql-subscriptions

Add GraphQL subscription support over graphql-transport-ws

Author: velo

graphql-apt/src/main/java/feign/graphql/apt/GraphqlSchemaProcessor.java

@@ -358,12 +358,15 @@ private boolean isJavaBuiltIn(String typeName) {
     return JAVA_BUILT_INS.contains(typeName);
   }
 
+  /** Wrappers that carry the operation result rather than being it. */
+  private static final Set<String> RESULT_CONTAINERS = Set.of("List", "Stream", "Publisher");
+
   private String getSimpleTypeName(TypeMirror typeMirror) {
     if (typeMirror instanceof DeclaredType declaredType) {
       var typeElement = declaredType.asElement();
       var simpleName = typeElement.getSimpleName().toString();
 
-      if ("List".equals(simpleName)) {
+      if (RESULT_CONTAINERS.contains(simpleName)) {
         var typeArgs = declaredType.getTypeArguments();
         if (!typeArgs.isEmpty()) {
           return getSimpleTypeName(typeArgs.get(0));
@@ -376,7 +379,7 @@ private String getSimpleTypeName(TypeMirror typeMirror) {
   }
 
   private boolean isExistingExternalType(TypeMirror typeMirror, String targetPackage) {
-    var unwrapped = unwrapListTypeMirror(typeMirror);
+    var unwrapped = unwrapContainerTypeMirror(typeMirror);
     if (unwrapped.getKind() == TypeKind.ERROR) {
       return false;
     }
@@ -392,13 +395,13 @@ private boolean isExistingExternalType(TypeMirror typeMirror, String targetPacka
     return false;
   }
 
-  private TypeMirror unwrapListTypeMirror(TypeMirror typeMirror) {
+  private TypeMirror unwrapContainerTypeMirror(TypeMirror typeMirror) {
     if (typeMirror instanceof DeclaredType declaredType) {
       var simpleName = declaredType.asElement().getSimpleName().toString();
-      if ("List".equals(simpleName)) {
+      if (RESULT_CONTAINERS.contains(simpleName)) {
         var typeArgs = declaredType.getTypeArguments();
         if (!typeArgs.isEmpty()) {
-          return typeArgs.get(0);
+          return unwrapContainerTypeMirror(typeArgs.get(0));
         }
       }
     }

graphql/README.md

@@ -100,6 +100,89 @@ The processor generates a record for the input type as well:
 public record CreateUserInput(String name, String email) {}
 ```
 
+## Subscriptions
+
+`subscription` operations are detected from the query text and executed over the
+[graphql-transport-ws](https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md) WebSocket
+protocol instead of HTTP. The endpoint is the target URL with its scheme swapped to `ws`/`wss`, and
+one connection is opened per call.
+
+Queries, mutations and subscriptions can live on the same interface: only subscriptions are routed
+to a WebSocket, everything else goes over the regular Feign client — including whichever one you
+configured with `.client(...)` — with its own timeouts, retryer and interceptors unchanged.
+
+The return type decides how many events you get and whether the call blocks:
+
+| Return type | Events | Behaviour |
+| --- | --- | --- |
+| `T` | first only | blocks until the first event, then unsubscribes |
+| `Optional<T>` | first only | as above, empty if the server completes without one |
+| `CompletableFuture<T>` | first only | returns immediately, completes with the first event |
+| `Stream<T>` | all | returns once subscribed, then blocks on each element |
+| `Flow.Publisher<T>` | all | returns immediately, elements are pushed to the subscriber |
+
+```java
+@GraphqlSchema("my-schema.graphql")
+interface StockApi {
+
+  @GraphqlQuery("subscription($symbol: String!) { priceChanged(symbol: $symbol) { symbol price } }")
+  Price nextPrice(@Param("symbol") String symbol);
+
+  @GraphqlQuery("subscription($symbol: String!) { priceChanged(symbol: $symbol) { symbol price } }")
+  Stream<Price> onPrice(@Param("symbol") String symbol);
+
+  @GraphqlQuery("subscription($symbol: String!) { priceChanged(symbol: $symbol) { symbol price } }")
+  Flow.Publisher<Price> publishPrice(@Param("symbol") String symbol);
+}
+```
+
+The single-event forms close the subscription as soon as they have their event. The multi-event
+forms hand you the lifecycle: closing the `Stream` — or cancelling the `Flow.Subscription` — sends
+`complete` and closes the WebSocket, so consume a `Stream` with try-with-resources:
+
+```java
+try (var prices = api.onPrice("ACME")) {
+  prices.forEach(System.out::println);
+}
+```
+
+`Stream` here is the ordinary `java.util.stream.Stream`: synchronous and pull-based, with no timeout
+facilities of its own. So the blocking forms — `T`, `Optional<T>` and `Stream<T>` — are bounded by
+an event timeout, which defaults to **60 seconds** and applies to each event rather than to the
+subscription as a whole. Override it when creating the capability:
+
+```java
+Feign.builder()
+    // wait at most 5s for each event; Duration.ZERO waits indefinitely
+    .addCapability(new GraphqlCapability(new JacksonCodec(), Duration.ofSeconds(5)))
+    .target(StockApi.class, "https://example.com/graphql");
+```
+
+Exceeding it raises `SocketTimeoutException` from the blocking call or the stream element. A
+subscription that can legitimately sit idle for longer needs `Duration.ZERO`.
+
+`Flow.Publisher<T>` and `CompletableFuture<T>` are deliberately *not* bounded by it — their caller
+already owns the deadline, via cancelling the subscription or
+`get(timeout, unit)`/`orTimeout(...)`. Cancelling either one closes the underlying WebSocket.
+
+Those two asynchronous forms each need a worker for as long as the subscription is open. They run on
+a bounded daemon pool by default; pass your own to own the lifecycle:
+
+```java
+new GraphqlCapability(new JacksonCodec(), Duration.ofSeconds(5), myExecutor)
+```
+
+Reads are demand-driven — the client asks the socket for another frame only once the consumer has
+taken the previous event — so a slow consumer applies backpressure to the server instead of growing
+a queue in memory.
+
+`Flow.Publisher` is `java.util.concurrent.Flow.Publisher`, so it plugs into Reactor
+(`JdkFlowAdapter.flowPublisherToFlux`) or RxJava (`Flowable.fromPublisher`) without extra
+dependencies here.
+
+A server `error` message, or `errors` inside a payload, is raised as `GraphqlErrorException`.
+Request headers (for example `Authorization`) are forwarded to the WebSocket handshake.
+
 ## Custom Scalars
 
 When your schema defines custom scalars, map them to Java types using `@Scalar` on default methods:

graphql/src/main/java/feign/graphql/GraphqlCapability.java

@@ -16,6 +16,7 @@
 package feign.graphql;
 
 import feign.Capability;
+import feign.Client;
 import feign.Contract;
 import feign.Experimental;
 import feign.RequestInterceptors;
@@ -24,7 +25,11 @@
 import feign.codec.JsonCodec;
 import feign.codec.JsonDecoder;
 import feign.codec.JsonEncoder;
+import java.time.Duration;
 import java.util.ArrayList;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicLong;
 
 @Experimental
 public class GraphqlCapability implements Capability {
@@ -33,15 +38,70 @@ public class GraphqlCapability implements Capability {
   private final GraphqlEncoder graphqlEncoder;
   private final GraphqlDecoder graphqlDecoder;
   private final GraphqlRequestInterceptor interceptor;
+  private final JsonEncoder jsonEncoder;
+  private final JsonDecoder jsonDecoder;
 
   public GraphqlCapability(JsonCodec codec) {
     this(codec.encoder(), codec.decoder());
   }
 
+  /**
+   * @param eventTimeout how long a blocking subscription call waits for an event before failing
+   *     with {@link java.net.SocketTimeoutException}; {@link java.time.Duration#ZERO} waits
+   *     indefinitely. Does not apply to {@code Flow.Publisher} or {@code CompletableFuture}
+   *     subscriptions, whose caller owns the deadline.
+   */
+  public GraphqlCapability(JsonCodec codec, Duration eventTimeout) {
+    this(codec.encoder(), codec.decoder(), eventTimeout);
+  }
+
+  /**
+   * @param executor runs the worker behind each {@code Flow.Publisher} and {@code
+   *     CompletableFuture} subscription, and delivers to their subscribers. Supply your own to own
+   *     the lifecycle; the default is bounded and daemon, and is never shut down.
+   */
+  public GraphqlCapability(JsonCodec codec, Duration eventTimeout, Executor executor) {
+    this(codec.encoder(), codec.decoder(), eventTimeout, executor);
+  }
+
   public GraphqlCapability(JsonEncoder encoder, JsonDecoder decoder) {
+    this(encoder, decoder, GraphqlDecoder.DEFAULT_EVENT_TIMEOUT);
+  }
+
+  /**
+   * @param eventTimeout see {@link #GraphqlCapability(JsonCodec, Duration)}
+   */
+  public GraphqlCapability(JsonEncoder encoder, JsonDecoder decoder, Duration eventTimeout) {
+    this(encoder, decoder, eventTimeout, defaultExecutor());
+  }
+
+  /**
+   * @param executor see {@link #GraphqlCapability(JsonCodec, Duration, Executor)}
+   */
+  public GraphqlCapability(
+      JsonEncoder encoder, JsonDecoder decoder, Duration eventTimeout, Executor executor) {
     this.graphqlEncoder = new GraphqlEncoder(encoder, contract);
-    this.graphqlDecoder = new GraphqlDecoder(decoder);
+    this.graphqlDecoder = new GraphqlDecoder(decoder, eventTimeout, executor);
     this.interceptor = new GraphqlRequestInterceptor(encoder, contract);
+    this.jsonEncoder = encoder;
+    this.jsonDecoder = decoder;
+  }
+
+  /**
+   * Each open {@code Flow.Publisher} or {@code CompletableFuture} subscription holds one worker for
+   * its lifetime, so the default pool grows on demand and reaps idle threads rather than capping
+   * concurrent subscriptions at a guess. Supply a bounded executor to cap them deliberately: the
+   * excess is refused with {@code RejectedExecutionException} rather than left hanging.
+   */
+  private static Executor defaultExecutor() {
+    var threads = new AtomicLong();
+    return Executors.newCachedThreadPool(
+        runnable -> {
+          var thread =
+              new Thread(runnable, "feign-graphql-subscription-" + threads.incrementAndGet());
+          thread.setDaemon(true);
+          return thread;
+        });
   }
 
   @Override
@@ -59,6 +119,11 @@ public Decoder enrich(Decoder decoder) {
     return graphqlDecoder;
   }
 
+  @Override
+  public Client enrich(Client client) {
+    return new GraphqlSubscriptionClient(client, contract, jsonEncoder, jsonDecoder);
+  }
+
   @Override
   public RequestInterceptors enrich(RequestInterceptors requestInterceptors) {
     var enriched = new ArrayList<>(requestInterceptors.interceptors());

graphql/src/main/java/feign/graphql/GraphqlContract.java

@@ -31,6 +31,8 @@ public class GraphqlContract extends DefaultContract {
 
   private static final Pattern VARIABLE_PATTERN = Pattern.compile("\\$\\s*(\\w+)\\s*:");
 
+  private static final Pattern SUBSCRIPTION_PATTERN = Pattern.compile("^\\s*subscription\\b");
+
   private final Map<String, QueryMetadata> metadata = new ConcurrentHashMap<>();
 
   public GraphqlContract() {
@@ -45,7 +47,8 @@ public GraphqlContract() {
           }
 
           var variableName = extractFirstVariable(query);
-          metadata.put(data.configKey(), new QueryMetadata(query, variableName));
+          metadata.put(
+              data.configKey(), new QueryMetadata(query, variableName, isSubscription(query)));
         });
   }
 
@@ -97,13 +100,19 @@ static String extractFirstVariable(String query) {
     return null;
   }
 
+  static boolean isSubscription(String query) {
+    return SUBSCRIPTION_PATTERN.matcher(query).find();
+  }
+
   static class QueryMetadata {
     final String query;
     final String variableName;
+    final boolean subscription;
 
-    QueryMetadata(String query, String variableName) {
+    QueryMetadata(String query, String variableName, boolean subscription) {
       this.query = query;
       this.variableName = variableName;
+      this.subscription = subscription;
     }
   }
 }