subscribe: rewrite async functions

Replicates graphql/graphql-js@3958158

Author: Cito

src/graphql/subscription/subscribe.py

@@ -88,6 +88,7 @@ async def map_source_to_response(payload: Any) -> ExecutionResult:
         )
         return await result if isawaitable(result) else result  # type: ignore
 
+    # Map every source value to a ExecutionResult value as described above.
     return MapAsyncIterator(result_or_stream, map_source_to_response)
 
 
@@ -127,25 +128,35 @@ async def create_source_event_stream(
     # mistake which should throw an early error.
     assert_valid_execution_arguments(schema, document, variable_values)
 
-    # If a valid context cannot be created due to incorrect arguments, this will throw
-    # an error.
-    context = ExecutionContext.build(
-        schema,
-        document,
-        root_value,
-        context_value,
-        variable_values,
-        operation_name,
-        field_resolver,
-    )
+    try:
+        # If a valid context cannot be created due to incorrect arguments,
+        # this will throw an error.
+        context = ExecutionContext.build(
+            schema,
+            document,
+            root_value,
+            context_value,
+            variable_values,
+            operation_name,
+            field_resolver,
+        )
 
-    # Return early errors if execution context failed.
-    if isinstance(context, list):
-        return ExecutionResult(data=None, errors=context)
+        # Return early errors if execution context failed.
+        if isinstance(context, list):
+            return ExecutionResult(data=None, errors=context)
+
+        event_stream = await execute_subscription(context)
+
+        # Assert field returned an event stream, otherwise yield an error.
+        if not isinstance(event_stream, AsyncIterable):
+            raise TypeError(
+                "Subscription field must return AsyncIterable."
+                f" Received: {inspect(event_stream)}."
+            )
+        return event_stream
 
-    try:
-        return await execute_subscription(context)
     except GraphQLError as error:
+        # Report it as an ExecutionResult, containing only errors and no data.
         return ExecutionResult(data=None, errors=[error])
 
 
@@ -171,29 +182,21 @@ async def execute_subscription(context: ExecutionContext) -> AsyncIterable[Any]:
     # Implements the "ResolveFieldEventStream" algorithm from GraphQL specification.
     # It differs from "ResolveFieldValue" due to providing a different `resolveFn`.
 
-    # Build a dictionary of arguments from the field.arguments AST, using the
-    # variables scope to fulfill any variable references.
-    args = get_argument_values(field_def, field_nodes[0], context.variable_values)
+    try:
+        # Build a dictionary of arguments from the field.arguments AST, using the
+        # variables scope to fulfill any variable references.
+        args = get_argument_values(field_def, field_nodes[0], context.variable_values)
 
-    # Call the `subscribe()` resolver or the default resolver to produce an
-    # AsyncIterable yielding raw payloads.
-    resolve_fn = field_def.subscribe or context.field_resolver
+        # Call the `subscribe()` resolver or the default resolver to produce an
+        # AsyncIterable yielding raw payloads.
+        resolve_fn = field_def.subscribe or context.field_resolver
 
-    try:
         event_stream = resolve_fn(context.root_value, info, **args)
         if context.is_awaitable(event_stream):
             event_stream = await event_stream
-    except Exception as error:
-        event_stream = error
-
-    if isinstance(event_stream, Exception):
-        raise located_error(event_stream, field_nodes, path.as_list())
+        if isinstance(event_stream, Exception):
+            raise event_stream
 
-    # Assert field returned an event stream, otherwise yield an error.
-    if not isinstance(event_stream, AsyncIterable):
-        raise TypeError(
-            "Subscription field must return AsyncIterable."
-            f" Received: {inspect(event_stream)}."
-        )
-
-    return event_stream
+        return event_stream
+    except Exception as error:
+        raise located_error(error, field_nodes, path.as_list())