Fix some error handling in subscriptions over websockets

mskacelik · Jan 25, 2023 · 7c05606 · 7c05606
1 parent 8b4322c
commit 7c05606
Show file tree

Hide file tree

Showing 9 changed files with 91 additions and 27 deletions.
diff --git a/...vertx/src/main/java/io/smallrye/graphql/client/vertx/websocket/graphqlws/MessageType.java b/...vertx/src/main/java/io/smallrye/graphql/client/vertx/websocket/graphqlws/MessageType.java
@@ -12,7 +12,7 @@ public enum MessageType {
     GQL_CONNECTION_ERROR("connection_error"),
     GQL_CONNECTION_ACK("connection_ack"),
     GQL_DATA("data"),
-    GQL_ERROR("connection_error"),
+    GQL_ERROR("error"),
     GQL_COMPLETE("complete"),
     GQL_CONNECTION_KEEP_ALIVE("ka");
 

diff --git a/...entation/src/main/java/io/smallrye/graphql/websocket/AbstractGraphQLWebsocketHandler.java b/...entation/src/main/java/io/smallrye/graphql/websocket/AbstractGraphQLWebsocketHandler.java
@@ -115,6 +115,12 @@ public void write(ExecutionResponse executionResponse) {
                                 } else if (data instanceof Publisher) {
                                     // this means the operation is a subscription
                                     sendStreamingMessage(operationId, executionResponse);
+                                } else if (data == null) {
+                                    // if isDataPresent() == true && but data == null,
+                                    // then this is probably a subscription, but the subscription
+                                    // method threw an exception instead of returning
+                                    // a failed Multi
+                                    sendErrorMessage(operationId, executionResponse);
                                 } else {
                                     logUnknownResult(executionResult);
                                 }
@@ -167,8 +173,8 @@ private JsonObject createDataMessage(String operationId, JsonObject payload) {
     }
 
     private void logUnknownResult(ExecutionResult executionResult) {
-        LOG.warn("Unknown execution result of type "
-                + executionResult.getClass());
+        LOG.warn("Unknown data type of execution result: "
+                + executionResult.getData().getClass());
     }
 
     private void sendSingleMessage(String operationId, ExecutionResponse executionResponse) throws IOException {

diff --git a/server/implementation/src/main/java/io/smallrye/graphql/websocket/graphqlws/MessageType.java b/server/implementation/src/main/java/io/smallrye/graphql/websocket/graphqlws/MessageType.java
@@ -12,7 +12,7 @@ public enum MessageType {
     GQL_CONNECTION_ERROR("connection_error"),
     GQL_CONNECTION_ACK("connection_ack"),
     GQL_DATA("data"),
-    GQL_ERROR("connection_error"),
+    GQL_ERROR("error"),
     GQL_COMPLETE("complete"),
     GQL_CONNECTION_KEEP_ALIVE("ka");
 

diff --git a/...lrye/graphql/tests/client/dynamic/subscription/AbstractDynamicClientSubscriptionTest.java b/...lrye/graphql/tests/client/dynamic/subscription/AbstractDynamicClientSubscriptionTest.java
@@ -6,28 +6,25 @@
 import static org.junit.Assert.*;
 
 import java.net.URL;
+import java.time.Duration;
 import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
-import jakarta.json.JsonValue;
-
 import org.jboss.arquillian.container.test.api.Deployment;
 import org.jboss.arquillian.test.api.ArquillianResource;
 import org.jboss.shrinkwrap.api.ShrinkWrap;
 import org.jboss.shrinkwrap.api.asset.EmptyAsset;
 import org.jboss.shrinkwrap.api.spec.WebArchive;
 import org.junit.After;
-import org.junit.Assert;
 import org.junit.Test;
 
+import io.smallrye.graphql.client.GraphQLClientException;
 import io.smallrye.graphql.client.GraphQLError;
 import io.smallrye.graphql.client.Response;
 import io.smallrye.graphql.client.core.Document;
 import io.smallrye.graphql.client.core.OperationType;
 import io.smallrye.graphql.client.vertx.dynamic.VertxDynamicGraphQLClient;
+import io.smallrye.mutiny.helpers.test.AssertSubscriber;
 
 public abstract class AbstractDynamicClientSubscriptionTest {
 
@@ -38,6 +35,8 @@ public static WebArchive deployment() {
                 .addClasses(DynamicClientSubscriptionApi.class);
     }
 
+    static Duration DURATION = Duration.ofSeconds(5);
+
     @ArquillianResource
     URL testingURL;
 
@@ -69,23 +68,28 @@ public void testFailingImmediately() throws InterruptedException {
         Document document = document(
                 operation(OperationType.SUBSCRIPTION,
                         field("failingImmediately")));
-        AtomicReference<Response> response = new AtomicReference<>();
-        CountDownLatch finished = new CountDownLatch(1);
-        client.subscription(document)
-                .subscribe()
-                .with(item -> {
-                    response.set(item);
-                }, throwable -> {
-                    // nothing
-                }, () -> {
-                    finished.countDown();
-                });
-        finished.await(10, TimeUnit.SECONDS);
-        Response actualResponse = response.get();
-        assertNotNull("One response was expected to arrive", actualResponse);
-        Assert.assertEquals(JsonValue.NULL, actualResponse.getData().get("failingImmediately"));
-        // FIXME: add an assertion about the contained error message
-        // right now, there is no error message present, which is a bug
+        AssertSubscriber<Response> subscriber = new AssertSubscriber<>(10);
+        client.subscription(document).subscribe(subscriber);
+        List<Response> messages = subscriber
+                .awaitNextItem(DURATION)
+                .awaitCompletion(DURATION)
+                .assertTerminated()
+                .getItems();
+        assertEquals(1, messages.size());
+        assertEquals("System error", messages.get(0).getErrors().get(0).getMessage());
+    }
+
+    @Test
+    public void testThrowingExceptionDirectly() throws InterruptedException {
+        Document document = document(
+                operation(OperationType.SUBSCRIPTION,
+                        field("throwingExceptionDirectly")));
+        AssertSubscriber<Response> subscriber = new AssertSubscriber<>(10);
+        client.subscription(document).subscribe(subscriber);
+        Throwable failure = subscriber.awaitFailure(DURATION)
+                .getFailure();
+        assertTrue(failure instanceof GraphQLClientException);
+        assertEquals("System error", ((GraphQLClientException) failure).getErrors().get(0).getMessage());
     }
 
     private void assertNoErrors(List<GraphQLError> errors) {

diff --git a/...a/io/smallrye/graphql/tests/client/dynamic/subscription/DynamicClientSubscriptionApi.java b/...a/io/smallrye/graphql/tests/client/dynamic/subscription/DynamicClientSubscriptionApi.java
@@ -25,4 +25,9 @@ public Multi<Integer> failingImmediately() {
         return Multi.createFrom().failure(new RuntimeException("blabla"));
     }
 
+    @Subscription
+    public Multi<Integer> throwingExceptionDirectly() {
+        throw new RuntimeException("blabla");
+    }
+
 }
diff --git a/...ye/graphql/tests/client/typesafe/subscription/AbstractTypesafeClientSubscriptionTest.java b/...ye/graphql/tests/client/typesafe/subscription/AbstractTypesafeClientSubscriptionTest.java
@@ -1,11 +1,13 @@
 package io.smallrye.graphql.tests.client.typesafe.subscription;
 
+import static io.smallrye.graphql.client.core.Field.field;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
 import java.net.URL;
+import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -22,6 +24,7 @@
 
 import io.smallrye.graphql.client.GraphQLClientException;
 import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.helpers.test.AssertSubscriber;
 
 public abstract class AbstractTypesafeClientSubscriptionTest {
 
@@ -37,6 +40,8 @@ public static WebArchive deployment() {
 
     protected SubscriptionClientApi client;
 
+    static Duration DURATION = Duration.ofSeconds(5);
+
     @After
     public void cleanup() {
         try {
@@ -107,4 +112,26 @@ public void failingSubscriptionShouldCloseClient() throws InterruptedException {
                 " to an exception in server-side processing", ended);
     }
 
+    @Test
+    public void testFailingImmediately() {
+        AssertSubscriber<Integer> subscriber = new AssertSubscriber<>(10);
+        client.failingImmediately().subscribe(subscriber);
+        Throwable failure = subscriber
+                .awaitFailure(DURATION)
+                .assertHasNotReceivedAnyItem()
+                .getFailure();
+        assertTrue(failure instanceof GraphQLClientException);
+    }
+
+    @Test
+    public void testThrowingExceptionDirectly() {
+        AssertSubscriber<Integer> subscriber = new AssertSubscriber<>(10);
+        client.failingImmediately().subscribe(subscriber);
+        Throwable failure = subscriber
+                .awaitFailure(DURATION)
+                .assertHasNotReceivedAnyItem()
+                .getFailure();
+        assertTrue(failure instanceof GraphQLClientException);
+    }
+
 }
diff --git a/...src/test/java/io/smallrye/graphql/tests/client/typesafe/subscription/SubscriptionApi.java b/...src/test/java/io/smallrye/graphql/tests/client/typesafe/subscription/SubscriptionApi.java
@@ -29,4 +29,14 @@ public Integer failingSourceField(@Source Dummy dummy) {
         throw new RuntimeException("FAILED SOURCE FIELD");
     }
 
+    @Subscription
+    public Multi<Integer> failingImmediately() {
+        return Multi.createFrom().failure(new RuntimeException("blabla"));
+    }
+
+    @Subscription
+    public Multi<Integer> throwingExceptionDirectly() {
+        throw new RuntimeException("blabla");
+    }
+
 }
diff --git a/...st/java/io/smallrye/graphql/tests/client/typesafe/subscription/SubscriptionClientApi.java b/...st/java/io/smallrye/graphql/tests/client/typesafe/subscription/SubscriptionClientApi.java
@@ -20,4 +20,10 @@ public interface SubscriptionClientApi extends Closeable {
     @Subscription(value = "countToFive")
     Multi<DummyWithErrorOrOnFailingSourceField> countToFiveWithFailingSourceFieldInErrorOr(boolean shouldFail);
 
+    @Subscription
+    Multi<Integer> failingImmediately();
+
+    @Subscription
+    Multi<Integer> throwingExceptionDirectly();
+
 }
diff --git a/ui/graphiql/src/main/webapp/render.js b/ui/graphiql/src/main/webapp/render.js
@@ -153,6 +153,9 @@ function graphQLFetcher(graphQLParams) {
                                 break;
                             case 'pong':
                                 break;
+                            case 'error':
+                                observer.next(data.payload);
+                                webSocket.close();
                             default:
                                 observer.next(data);
                                 break;
@@ -182,6 +185,9 @@ function graphQLFetcher(graphQLParams) {
                                 break;
                             case 'ka':
                                 break;
+                            case 'error':
+                                observer.next(data);
+                                webSocket.close();
                             default:
                                 observer.next(data);
                                 break;