Subscriptions: Respond with error when failing to create source

If a subscribe resolve function throws or returns an error, that typically indicates an issue to be returned to the requesting client. This coerces errors into located GraphQLErrors so they are correctly reported.

src/subscription/__tests__/subscribe-test.js

@@ -65,17 +65,23 @@ describe('Subscribe', () => {
     }
   });
 
-  const SubscriptionType = new GraphQLObjectType({
-    name: 'Subscription',
-    fields: {
-      importantEmail: { type: EmailEventType },
-    }
-  });
+  const emailSchema = emailSchemaWithResolvers();
 
-  const emailSchema = new GraphQLSchema({
-    query: QueryType,
-    subscription: SubscriptionType
-  });
+  function emailSchemaWithResolvers(subscribeFn, resolveFn) {
+    return new GraphQLSchema({
+      query: QueryType,
+      subscription: new GraphQLObjectType({
+        name: 'Subscription',
+        fields: {
+          importantEmail: {
+            type: EmailEventType,
+            resolve: resolveFn,
+            subscribe: subscribeFn,
+          },
+        },
+      })
+    });
+  }
 
   function createSubscription(pubsub, schema = emailSchema, ast) {
     const data = {
@@ -641,40 +647,71 @@ describe('Subscribe', () => {
     expect(caughtError).to.equal(undefined);
   });
 
-  it('should handle error thrown by subscribe method', async () => {
-    const invalidEmailSchema = new GraphQLSchema({
-      query: QueryType,
-      subscription: new GraphQLObjectType({
-        name: 'Subscription',
-        fields: {
-          importantEmail: {
-            type: GraphQLString,
-            subscribe: () => {
-              throw new Error('test error');
-            },
-          },
-        },
-      })
-    });
+  it('should report error thrown by subscribe function', async () => {
+    const erroringEmailSchema = emailSchemaWithResolvers(
+      () => { throw new Error('test error'); }
+    );
 
-    const ast = parse(`
-      subscription {
-        importantEmail
+    const subscription = subscribe(
+      erroringEmailSchema,
+      parse(`
+        subscription {
+          importantEmail
+        }
+      `)
+    );
+
+    const result = await subscription.next();
+
+    expect(result).to.deep.equal({
+      done: false,
+      value: {
+        errors: [
+          {
+            message: 'test error',
+            locations: [ { line: 3, column: 11 } ],
+            path: [ 'importantEmail' ]
+          }
+        ]
       }
-    `);
+    });
+
+    expect(
+      await subscription.next()
+    ).to.deep.equal({ value: undefined, done: true });
+  });
+
+  it('should report error returned by subscribe function', async () => {
+    const erroringEmailSchema = emailSchemaWithResolvers(
+      () => { return new Error('test error'); }
+    );
 
     const subscription = subscribe(
-      invalidEmailSchema,
-      ast
+      erroringEmailSchema,
+      parse(`
+        subscription {
+          importantEmail
+        }
+      `)
     );
 
-    let caughtError;
-    try {
-      await subscription.next();
-    } catch (thrownError) {
-      caughtError = thrownError;
-    }
+    const result = await subscription.next();
 
-    expect(caughtError && caughtError.message).to.equal('test error');
+    expect(result).to.deep.equal({
+      done: false,
+      value: {
+        errors: [
+          {
+            message: 'test error',
+            locations: [ { line: 3, column: 11 } ],
+            path: [ 'importantEmail' ]
+          }
+        ]
+      }
+    });
+
+    expect(
+      await subscription.next()
+    ).to.deep.equal({ value: undefined, done: true });
   });
 });

src/subscription/subscribe.js

@@ -11,6 +11,7 @@
 
 import { isAsyncIterable } from 'iterall';
 import { GraphQLError } from '../error/GraphQLError';
+import { locatedError } from '../error/locatedError';
 import {
   addPath,
   assertValidExecutionArguments,
@@ -21,6 +22,7 @@ import {
   getFieldDef,
   getOperationRootType,
   resolveFieldValueOrError,
+  responsePathAsArray,
 } from '../execution/execute';
 import { GraphQLSchema } from '../type/schema';
 import invariant from '../jsutils/invariant';
@@ -225,12 +227,14 @@ export function createSourceEventStream(
   // AsyncIterable yielding raw payloads.
   const resolveFn = fieldDef.subscribe || exeContext.fieldResolver;
 
+  const path = addPath(undefined, responseName);
+
   const info = buildResolveInfo(
     exeContext,
     fieldDef,
     fieldNodes,
     type,
-    addPath(undefined, responseName)
+    path
   );
 
   // resolveFieldValueOrError implements the "ResolveFieldEventStream"
@@ -246,7 +250,7 @@ export function createSourceEventStream(
   );
 
   if (subscription instanceof Error) {
-    throw subscription;
+    throw locatedError(subscription, fieldNodes, responsePathAsArray(path));
   }
 
   if (!isAsyncIterable(subscription)) {