Make error handling consistent in createSourceEventStream (#1467)

src/subscription/__tests__/subscribe-test.js

@@ -4,7 +4,8 @@ import { expect } from 'chai';
 import { describe, it } from 'mocha';
 import EventEmitter from 'events';
 import eventEmitterAsyncIterator from './eventEmitterAsyncIterator';
-import { subscribe } from '../subscribe';
+import { createSourceEventStream, subscribe } from '../subscribe';
+import { GraphQLError } from '../../error';
 import { parse } from '../../language';
 import {
   GraphQLSchema,
@@ -431,6 +432,58 @@ describe('Subscription Initialization Phase', () => {
     }
   });
 
+  it('resolves to an error for source event stream resolver errors', async () => {
+    // Returning an error
+    const subscriptionReturningErrorSchema = emailSchemaWithResolvers(() => {
+      return new Error('test error');
+    });
+    await testReportsError(subscriptionReturningErrorSchema);
+
+    // Throwing an error
+    const subscriptionThrowingErrorSchema = emailSchemaWithResolvers(() => {
+      throw new Error('test error');
+    });
+    await testReportsError(subscriptionThrowingErrorSchema);
+
+    // Resolving to an error
+    const subscriptionResolvingErrorSchema = emailSchemaWithResolvers(
+      async () => {
+        return new Error('test error');
+      },
+    );
+    await testReportsError(subscriptionResolvingErrorSchema);
+
+    // Rejecting with an error
+    const subscriptionRejectingErrorSchema = emailSchemaWithResolvers(
+      async () => {
+        throw new Error('test error');
+      },
+    );
+    await testReportsError(subscriptionRejectingErrorSchema);
+
+    async function testReportsError(schema) {
+      // Promise<AsyncIterable<ExecutionResult> | ExecutionResult>
+      const result = await createSourceEventStream(
+        schema,
+        parse(`
+          subscription {
+            importantEmail
+          }
+        `),
+      );
+
+      expect(result).to.deep.equal({
+        errors: [
+          {
+            message: 'test error',
+            locations: [{ line: 3, column: 13 }],
+            path: ['importantEmail'],
+          },
+        ],
+      });
+    }
+  });
+
   it('resolves to an error if variables were wrong type', async () => {
     // If we receive variables that cannot be coerced correctly, subscribe()
     // will resolve to an ExecutionResult that contains an informative error
@@ -937,4 +990,59 @@ describe('Subscription Publish Phase', () => {
       value: undefined,
     });
   });
+
+  it('should resolve GraphQL error from source event stream', async () => {
+    const erroringEmailSchema = emailSchemaWithResolvers(
+      async function*() {
+        yield { email: { subject: 'Hello' } };
+        throw new GraphQLError('test error');
+      },
+      email => email,
+    );
+
+    const subscription = await subscribe(
+      erroringEmailSchema,
+      parse(`
+        subscription {
+          importantEmail {
+            email {
+              subject
+            }
+          }
+        }
+      `),
+    );
+
+    const payload1 = await subscription.next();
+    expect(payload1).to.deep.equal({
+      done: false,
+      value: {
+        data: {
+          importantEmail: {
+            email: {
+              subject: 'Hello',
+            },
+          },
+        },
+      },
+    });
+
+    const payload2 = await subscription.next();
+    expect(payload2).to.deep.equal({
+      done: false,
+      value: {
+        errors: [
+          {
+            message: 'test error',
+          },
+        ],
+      },
+    });
+
+    const payload3 = await subscription.next();
+    expect(payload3).to.deep.equal({
+      done: true,
+      value: undefined,
+    });
+  });
 });

src/subscription/subscribe.js

@@ -38,8 +38,9 @@ export type SubscriptionArgs = {|
  * Implements the "Subscribe" algorithm described in the GraphQL specification.
  *
  * Returns a Promise which resolves to either an AsyncIterator (if successful)
- * or an ExecutionResult (client error). The promise will be rejected if a
- * server error occurs.
+ * or an ExecutionResult (error). The promise will be rejected if the schema or
+ * other arguments to this function are invalid, or if the resolved event stream
+ * is not an async iterable.
  *
  * If the client-provided arguments to this function do not result in a
  * compliant subscription, a GraphQL Response (ExecutionResult) with
@@ -160,19 +161,28 @@ function subscribeImpl(
             reportGraphQLError,
           )
         : ((resultOrStream: any): ExecutionResult),
-    reportGraphQLError,
   );
 }
 
 /**
  * Implements the "CreateSourceEventStream" algorithm described in the
  * GraphQL specification, resolving the subscription source event stream.
  *
- * Returns a Promise<AsyncIterable>.
+ * Returns a Promise which resolves to either an AsyncIterable (if successful)
+ * or an ExecutionResult (error). The promise will be rejected if the schema or
+ * other arguments to this function are invalid, or if the resolved event stream
+ * is not an async iterable.
  *
- * If the client-provided invalid arguments, the source stream could not be
- * created, or the resolver did not return an AsyncIterable, this function will
- * will throw an error, which should be caught and handled by the caller.
+ * If the client-provided arguments to this function do not result in a
+ * compliant subscription, a GraphQL Response (ExecutionResult) with
+ * descriptive errors and no data will be returned.
+ *
+ * If the the source stream could not be created due to faulty subscription
+ * resolver logic or underlying systems, the promise will resolve to a single
+ * ExecutionResult containing `errors` and no `data`.
+ *
+ * If the operation succeeded, the promise resolves to the AsyncIterable for the
+ * event stream returned by the resolver.
  *
  * A Source Event Stream represents a sequence of events, each of which triggers
  * a GraphQL execution for that event.
@@ -259,7 +269,11 @@ export function createSourceEventStream(
     return Promise.resolve(result).then(eventStream => {
       // If eventStream is an Error, rethrow a located error.
       if (eventStream instanceof Error) {
-        throw locatedError(eventStream, fieldNodes, responsePathAsArray(path));
+        return {
+          errors: [
+            locatedError(eventStream, fieldNodes, responsePathAsArray(path)),
+          ],
+        };
       }
 
       // Assert field returned an event stream, otherwise yield an error.
@@ -273,6 +287,11 @@ export function createSourceEventStream(
       );
     });
   } catch (error) {
-    return Promise.reject(error);
+    // As with reportGraphQLError above, if the error is a GraphQLError, report
+    // it as an ExecutionResult; otherwise treat it as a system-class error and
+    // re-throw it.
+    return error instanceof GraphQLError
+      ? Promise.resolve({ errors: [error] })
+      : Promise.reject(error);
   }
 }