Add a finish_subscriptions method

Author: rmosolgo

lib/graphql/execution/interpreter.rb

@@ -88,8 +88,8 @@ def run_all(schema, query_options, context: {}, max_complexity: schema.max_compl
               # Then, find all errors and assign the result to the query object
               results.each_with_index do |data_result, idx|
                 query = queries[idx]
-                if (events = query.context.namespace(:subscriptions)[:events]) && !events.empty?
-                  schema.subscriptions.write_subscription(query, events)
+                if query.subscription?
+                  schema.subscriptions.finish_subscriptions(query)
                 end
                 # Assign the result so that it can be accessed in instrumentation
                 query.result_values = if data_result.equal?(NO_OPERATION)

lib/graphql/execution/next/runner.rb

@@ -176,8 +176,8 @@ def execute
 
             queries.each_with_index.map do |query, idx|
               result = results[idx]
-              if (events = query.context.namespace(:subscriptions)[:events]) && !events.empty?
-                @schema.subscriptions.write_subscription(query, events)
+              if query.subscription?
+                @schema.subscriptions.finish_subscriptions(query)
               end
 
               fin_result = if query.context.errors.empty?

lib/graphql/schema/resolver.rb

@@ -74,6 +74,10 @@ def call
 
         result = if is_authed
           Schema::Validator.validate!(self.class.validators, object, context, @prepared_arguments, as: @field)
+          if q.subscription? && @field.owner == context.schema.subscription
+            # This needs to use arguments without `loads:`
+            @original_arguments = @field_resolve_step.coerce_arguments(@field, @field_resolve_step.ast_node.arguments, false)
+          end
           call_resolve(@prepared_arguments)
         elsif new_return_value.nil?
           err = UnauthorizedFieldError.new(object: object, type: @field_resolve_step.parent_type, context: context, field: @field)
@@ -83,12 +87,6 @@ def call
         end
         q = context.query
         q.current_trace.end_execute_field(field, @prepared_arguments, trace_objs, q, [result])
-        if q.subscription? && @field.owner == context.schema.subscription && !@subscription_written
-          # TODO unify this -- do it in a single pass
-          @original_arguments = @field_resolve_step.coerce_arguments(@field, @field_resolve_step.ast_node.arguments, false)
-          Subscriptions::DefaultSubscriptionResolveExtension.write_subscription(@field, result, @original_arguments, context)
-          @subscription_written = true
-        end
         exec_result[exec_index] = result
       rescue RuntimeError => err
         exec_result[exec_index] = err

lib/graphql/subscriptions.rb

@@ -249,6 +249,16 @@ def initialize_subscriptions(query)
       nil
     end
 
+    # Called during execution when a subscription operation has finished
+    # @param query [GraphQL::Query]
+    # @return [void]
+    def finish_subscriptions(query)
+      if (events = query.context.namespace(:subscriptions)[:events]) && !events.empty?
+        write_subscription(query, events)
+      end
+      nil
+    end
+
     private
 
     # Recursively normalize `args` as belonging to `arg_owner`: