Merge pull request #5589 from rmosolgo/exec-next-subscriptions

exec-next: Subscriptions

Author: rmosolgo

lib/graphql/execution.rb

@@ -10,10 +10,13 @@
 module GraphQL
   module Execution
     # @api private
-    class Skip < GraphQL::Error; end
+    class Skip < GraphQL::RuntimeError
+      attr_accessor :path
+      def ast_nodes=(_ignored); end
 
-    # Just a singleton for implementing {Query::Context#skip}
-    # @api private
-    SKIP = Skip.new
+      def assign_graphql_result(query, result_data, key)
+        result_data.delete(key)
+      end
+    end
   end
 end

lib/graphql/execution/interpreter.rb

@@ -58,11 +58,6 @@ def run_all(schema, query_options, context: {}, max_complexity: schema.max_compl
               # Do as much eager evaluation of the query as possible
               results = []
               queries.each_with_index do |query, idx|
-                if query.subscription? && !query.subscription_update?
-                  subs_namespace = query.context.namespace(:subscriptions)
-                  subs_namespace[:events] = []
-                  subs_namespace[:subscriptions] = {}
-                end
                 multiplex.dataloader.append_job {
                   operation = query.selected_operation
                   result = if operation.nil? || !query.valid? || !query.context.errors.empty?
@@ -74,7 +69,9 @@ def run_all(schema, query_options, context: {}, max_complexity: schema.max_compl
                       # in particular, assign it here:
                       runtime = Runtime.new(query: query)
                       query.context.namespace(:interpreter_runtime)[:runtime] = runtime
-
+                      if query.subscription? && !query.subscription_update?
+                        schema.subscriptions.initialize_subscriptions(query)
+                      end
                       query.current_trace.execute_query(query: query) do
                         runtime.run_eager
                       end
@@ -91,9 +88,6 @@ 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)
-                end
                 # Assign the result so that it can be accessed in instrumentation
                 query.result_values = if data_result.equal?(NO_OPERATION)
                   if !query.valid? || !query.context.errors.empty?
@@ -103,6 +97,9 @@ def run_all(schema, query_options, context: {}, max_complexity: schema.max_compl
                     data_result
                   end
                 else
+                  if query.subscription?
+                    schema.subscriptions.finish_subscriptions(query)
+                  end
                   result = {}
 
                   if !query.context.errors.empty?

lib/graphql/execution/interpreter/runtime.rb

@@ -603,7 +603,7 @@ def continue_value(value, field, is_non_null, ast_node, result_name, selection_r
                 err
               end
               continue_value(next_value, field, is_non_null, ast_node, result_name, selection_result)
-            elsif GraphQL::Execution::SKIP == value
+            elsif value.is_a?(GraphQL::Execution::Skip)
               # It's possible a lazy was already written here
               case selection_result
               when GraphQLResultHash

lib/graphql/execution/lazy.rb

@@ -38,7 +38,7 @@ def value
         # (fewer clauses in a hot `case` block), but now it requires special handling here.
         # I think it's still worth it for the performance win, but if the number of special
         # cases grows, then maybe it's worth rethinking somehow.
-        if @value.is_a?(StandardError) && @value != GraphQL::Execution::SKIP
+        if @value.is_a?(StandardError) && !@value.is_a?(GraphQL::Execution::Skip)
           raise @value
         else
           @value

lib/graphql/execution/next.rb

@@ -52,14 +52,16 @@ def self.use(schema, authorization: true)
 
       def self.run_all(schema, query_options, context: {}, max_complexity: schema.max_complexity)
         queries = query_options.map do |opts|
-          case opts
+          query = case opts
           when Hash
             schema.query_class.new(schema, nil, **opts)
           when GraphQL::Query, GraphQL::Query::Partial
             opts
           else
             raise "Expected Hash or GraphQL::Query, not #{opts.class} (#{opts.inspect})"
           end
+          query.context[:__graphql_execute_next] = true
+          query
         end
         multiplex = Execution::Multiplex.new(schema: schema, queries: queries, context: context, max_complexity: max_complexity)
         runner = Runner.new(multiplex, **schema.execution_next_options)

lib/graphql/execution/next/field_resolve_step.rb

@@ -48,59 +48,85 @@ def append_selection(ast_node)
           nil
         end
 
-        def coerce_arguments(argument_owner, ast_arguments_or_hash)
+        def coerce_arguments(argument_owner, ast_arguments_or_hash, run_loads = true)
           arg_defns = argument_owner.arguments(@selections_step.query.context)
           if arg_defns.empty?
             return EmptyObjects::EMPTY_HASH
           end
           args_hash = {}
-          if ast_arguments_or_hash.is_a?(Hash)
-            ast_arguments_or_hash.each do |key, value|
-              key_s = nil
-              arg_defn = arg_defns.each_value.find { |a|
-                a.keyword == key || a.graphql_name == (key_s ||= String(key))
-              }
-              coerce_argument_value(args_hash, arg_defn, value)
-            end
-          else
-            ast_arguments_or_hash.each { |arg_node|
-              arg_defn = arg_defns[arg_node.name]
-              coerce_argument_value(args_hash, arg_defn, arg_node.value)
-            }
+          if ast_arguments_or_hash.nil? # This can happen with `.trigger`
+            return args_hash
           end
-          # TODO refactor the loop above into this one
+
+          arg_inputs_are_h = ast_arguments_or_hash.is_a?(Hash)
+
           arg_defns.each do |arg_graphql_name, arg_defn|
-            if arg_defn.default_value? && !args_hash.key?(arg_defn.keyword)
-              coerce_argument_value(args_hash, arg_defn, arg_defn.default_value)
+            arg_value = nil
+            was_found = false
+            if arg_inputs_are_h
+              ast_arguments_or_hash.each do |key, value|
+                if key == arg_defn.keyword || key.to_s == arg_defn.graphql_name
+                  arg_value = value
+                  was_found = true
+                  break
+                end
+              end
+            else
+              ast_arguments_or_hash.each do |arg_node|
+                if arg_node.name == arg_defn.graphql_name
+                  arg_value = arg_node.value
+                  was_found = true
+                  break
+                end
+              end
+            end
+
+            if arg_value.is_a?(Language::Nodes::VariableIdentifier)
+              vars = @selections_step.query.variables
+              arg_value = if vars.key?(arg_value.name)
+                vars[arg_value.name]
+              elsif vars.key?(arg_value.name.to_sym)
+                vars[arg_value.name.to_sym]
+              else
+                was_found = false
+                nil
+              end
+            end
+
+            if !was_found && arg_defn.default_value?
+              was_found = true
+              arg_value = arg_defn.default_value
+            end
+
+            if was_found
+              coerce_argument_value(args_hash, arg_defn, arg_value, run_loads)
             end
           end
 
           args_hash
         end
 
-        def coerce_argument_value(arguments, arg_defn, arg_value, target_keyword: arg_defn.keyword, as_type: nil)
+        def coerce_argument_value(arguments, arg_defn, arg_value, run_loads, target_keyword: run_loads ? arg_defn.keyword : arg_defn.graphql_name, as_type: nil)
           arg_t = as_type || arg_defn.type
           if arg_t.non_null?
             arg_t = arg_t.of_type
           end
 
-          arg_value = if arg_value.is_a?(Language::Nodes::VariableIdentifier)
+          if arg_value.is_a?(Language::Nodes::VariableIdentifier)
             vars = @selections_step.query.variables
-            if vars.key?(arg_value.name)
+            arg_value = if vars.key?(arg_value.name)
               vars[arg_value.name]
             elsif vars.key?(arg_value.name.to_sym)
               vars[arg_value.name.to_sym]
             else
-              return # not present
+              nil
             end
-          elsif arg_value.is_a?(Language::Nodes::NullValue)
-            nil
+          end
+
+          if arg_value.is_a?(Language::Nodes::NullValue)
+            arg_value = nil
           elsif arg_value.is_a?(Language::Nodes::Enum)
-            arg_value.name
-          elsif arg_value.is_a?(Language::Nodes::InputObject)
-            arg_value.arguments # rubocop:disable Development/ContextIsPassedCop
-          else
-            arg_value
+            arg_value = arg_value.name
           end
 
           ctx = @selections_step.query.context
@@ -111,7 +137,7 @@ def coerce_argument_value(arguments, arg_defn, arg_value, target_keyword: arg_de
               arg_value = Array(arg_value)
               inner_t = arg_t.of_type
               result = Array.new(arg_value.size)
-              arg_value.each_with_index { |v, i| coerce_argument_value(result, arg_defn, v, target_keyword: i, as_type: inner_t) }
+              arg_value.each_with_index { |v, i| coerce_argument_value(result, arg_defn, v, run_loads, target_keyword: i, as_type: inner_t) }
               result
             end
           elsif arg_t.kind.leaf?
@@ -125,7 +151,8 @@ def coerce_argument_value(arguments, arg_defn, arg_value, target_keyword: arg_de
               end
             end
           elsif arg_t.kind.input_object?
-            input_obj_args = coerce_arguments(arg_t, arg_value)
+            input_obj_vals = arg_value.is_a?(Language::Nodes::InputObject) ? arg_value.arguments : arg_value # rubocop:disable Development/ContextIsPassedCop
+            input_obj_args = coerce_arguments(arg_t, input_obj_vals)
             arg_t.new(nil, ruby_kwargs: input_obj_args, context: @selections_step.query.context, defaults_used: nil)
           else
             raise "Unsupported argument value: #{arg_t.to_type_signature} / #{arg_value.class} (#{arg_value.inspect})"
@@ -145,7 +172,7 @@ def coerce_argument_value(arguments, arg_defn, arg_value, target_keyword: arg_de
 
           if arg_value.is_a?(GraphQL::Error)
             @arguments = arg_value
-          elsif arg_defn.loads && as_type.nil? && !arg_value.nil?
+          elsif run_loads && arg_defn.loads && as_type.nil? && !arg_value.nil?
             # This is for legacy compat:
             load_receiver = if (r = @field_definition.resolver)
               r.new(field: @field_definition, context: @selections_step.query.context, object: nil)
@@ -263,7 +290,8 @@ def build_arguments
           arguments = coerce_arguments(@field_definition, @ast_node.arguments) # rubocop:disable Development/ContextIsPassedCop
           @arguments ||= arguments # may have already been set to an error
 
-          if @pending_steps.nil? || @pending_steps.size == 0
+          if (@pending_steps.nil? || @pending_steps.size == 0) &&
+              @field_results.nil? # Make sure the arguments flow didn't already call through
             execute_field
           end
         end
@@ -323,6 +351,14 @@ def execute_field
               is_authed = @field_definition.authorized?(o, @arguments, ctx)
               if is_authed
                 authorized_objects << o
+              else
+                begin
+                  err = GraphQL::UnauthorizedFieldError.new(object: o, type: @parent_type, context: ctx, field: @field_definition)
+                  authorized_objects << query.schema.unauthorized_object(err)
+                  is_authed = true
+                rescue GraphQL::ExecutionError => exec_err
+                  add_graphql_error(exec_err)
+                end
               end
               is_authed
             }
@@ -616,29 +652,35 @@ def resolve_batch(objects, context, args_hash)
           method_receiver = @field_definition.dynamic_introspection ? @field_definition.owner : @parent_type
           case @field_definition.execution_next_mode
           when :resolve_batch
-            if args_hash.empty?
-              method_receiver.public_send(@field_definition.execution_next_mode_key, objects, context)
-            else
+            begin
               method_receiver.public_send(@field_definition.execution_next_mode_key, objects, context, **args_hash)
+            rescue GraphQL::ExecutionError => exec_err
+              Array.new(objects.size, exec_err)
             end
           when :resolve_static
-            result = if args_hash.empty?
-              method_receiver.public_send(@field_definition.execution_next_mode_key, context)
-            else
-              method_receiver.public_send(@field_definition.execution_next_mode_key, context, **args_hash)
-            end
+            result = method_receiver.public_send(@field_definition.execution_next_mode_key, context, **args_hash)
             Array.new(objects.size, result)
           when :resolve_each
-            if args_hash.empty?
-              objects.map { |o| method_receiver.public_send(@field_definition.execution_next_mode_key, o, context) }
-            else
-              objects.map { |o| method_receiver.public_send(@field_definition.execution_next_mode_key, o, context, **args_hash) }
+            objects.map do |o|
+              method_receiver.public_send(@field_definition.execution_next_mode_key, o, context, **args_hash)
+            rescue GraphQL::ExecutionError => err
+              err
             end
           when :hash_key
             objects.map { |o| o[@field_definition.execution_next_mode_key] }
           when :direct_send
             if args_hash.empty?
-              objects.map { |o| o.public_send(@field_definition.execution_next_mode_key) }
+              objects.map do |o|
+                o.public_send(@field_definition.execution_next_mode_key)
+              rescue GraphQL::ExecutionError => err
+                err
+              rescue StandardError => stderr
+                begin
+                  @selections_step.query.handle_or_reraise(stderr)
+                rescue GraphQL::ExecutionError => ex_err
+                  ex_err
+                end
+              end
             else
               objects.map { |o| o.public_send(@field_definition.execution_next_mode_key, **args_hash) }
             end
@@ -684,17 +726,12 @@ def resolve_batch(objects, context, args_hash)
               if @field_definition.dynamic_introspection
                 obj_inst = @owner.wrap(obj_inst, context)
               end
-              if args_hash.empty?
-                obj_inst.public_send(@field_definition.execution_next_mode_key)
-              else
-                obj_inst.public_send(@field_definition.execution_next_mode_key, **args_hash)
-              end
+              obj_inst.public_send(@field_definition.execution_next_mode_key, **args_hash)
             end
           else
             raise "Batching execution for #{path} not implemented (execution_next_mode: #{@execution_next_mode.inspect}); provide `resolve_static:`, `resolve_batch:`, `hash_key:`, `method:`, or use a compatibility plug-in"
           end
         end
-
       end
 
       class RawValueFieldResolveStep < FieldResolveStep