implement @stream

Author: rmosolgo

lib/graphql/directive.rb

@@ -48,3 +48,4 @@ def on_operation?
 require "graphql/directive/defer_directive"
 require "graphql/directive/include_directive"
 require "graphql/directive/skip_directive"
+require "graphql/directive/stream_directive"

lib/graphql/directive/stream_directive.rb

@@ -0,0 +1,9 @@
+GraphQL::Directive::StreamDirective = GraphQL::Directive.define do
+  name "stream"
+  description "Push items from this list in sequential patches"
+  locations([GraphQL::Directive::FIELD])
+
+  # This doesn't make any sense in this context
+  # But it's required for DirectiveResolution
+  include_proc -> (args) { true }
+end

lib/graphql/execution/deferred_execution.rb

@@ -101,16 +101,41 @@ def execute(ast_operation, root_type, query_object)
           defers = initial_thread.defers
           while defers.any?
             next_defers = []
-            defers.each do |deferred_frame|
+            defers.each do |deferral|
               deferred_thread = ExecThread.new
-              deferred_result = resolve_frame(scope, deferred_thread, deferred_frame)
-              # No use patching for nil, that's there already
+              case deferral
+              when ExecFrame
+                deferred_frame = deferral
+                deferred_result = resolve_frame(scope, deferred_thread, deferred_frame)
+
+              when ExecStream
+                begin
+                  list_frame = deferral.frame
+                  inner_type = deferral.type
+                  item, idx = deferral.enumerator.next
+                  deferred_frame = ExecFrame.new({
+                    node: list_frame.node,
+                    path: list_frame.path + [idx],
+                    type: inner_type,
+                    value: item,
+                  })
+                  deferred_result = resolve_value(scope, deferred_thread, deferred_frame, item, inner_type)
+                  deferred_thread.defers << deferral
+                rescue StopIteration
+                  # The enum is done
+                end
+              else
+                raise("Can't continue deferred #{deferred_frame.class.name}")
+              end
+
+              # TODO: Should I patch nil?
               if !deferred_result.nil?
                 collector.patch(
                   path: ["data"] + deferred_frame.path,
                   value: deferred_result
                 )
               end
+
               deferred_thread.errors.each do |deferred_error|
                 collector.patch(
                   path: ["errors", error_idx],
@@ -183,6 +208,20 @@ def initialize(node:, path:, type:, value:)
         end
       end
 
+      # Contains the list field's ExecFrame
+      # And the enumerator which is being mapped
+      # - {ExecStream#enumerator} is an Enumerator which yields `item, idx`
+      # - {ExecStream#frame} is the {ExecFrame} for the list selection (where `@stream` was present)
+      # - {ExecStream#type} is the inner type of the list (the item's type)
+      class ExecStream
+        attr_reader :enumerator, :frame, :type
+        def initialize(enumerator:, frame:, type:)
+          @enumerator = enumerator
+          @frame = frame
+          @type = type
+        end
+      end
+
       private
 
       # If this `frame` is marked as defer, add it to `defers`
@@ -327,16 +366,27 @@ def resolve_value(scope, thread, frame, value, type_defn)
             resolve_value(scope, thread, frame, value, wrapped_type)
           when GraphQL::TypeKinds::LIST
             wrapped_type = type_defn.of_type
-            resolved_values = value.each_with_index.map do |item, idx|
-              inner_frame = ExecFrame.new({
-                node: frame.node,
-                path: frame.path + [idx],
+            items_enumerator = value.map.with_index
+            if GraphQL::Execution::DirectiveChecks.stream?(frame.node)
+              thread.defers << ExecStream.new(
+                enumerator: items_enumerator,
+                frame: frame,
                 type: wrapped_type,
-                value: item,
-              })
-              resolve_value(scope, thread, inner_frame, item, wrapped_type)
+              )
+              # The streamed list is empty in the initial resolve:
+              []
+            else
+              resolved_values = items_enumerator.each do |item, idx|
+                inner_frame = ExecFrame.new({
+                  node: frame.node,
+                  path: frame.path + [idx],
+                  type: wrapped_type,
+                  value: item,
+                })
+                resolve_value(scope, thread, inner_frame, item, wrapped_type)
+              end
+              resolved_values
             end
-            resolved_values
           when GraphQL::TypeKinds::INTERFACE, GraphQL::TypeKinds::UNION
             resolved_type = type_defn.resolve_type(value, scope)
 

lib/graphql/execution/directive_checks.rb

@@ -6,6 +6,7 @@ module DirectiveChecks
       SKIP = "skip"
       INCLUDE = "include"
       DEFER = "defer"
+      STREAM = "stream"
 
       module_function
 
@@ -14,6 +15,11 @@ def defer?(ast_node)
         ast_node.directives.any? { |dir| dir.name == DEFER }
       end
 
+      # @return [Boolean] Should this AST node be streamed?
+      def stream?(ast_node)
+        ast_node.directives.any? { |dir| dir.name == STREAM }
+      end
+
       # @return [Boolean] Should this AST node be skipped altogether?
       def skip?(ast_node, query)
         ast_node.directives.each do |ast_directive|

lib/graphql/schema.rb

@@ -17,6 +17,7 @@ class Schema
       GraphQL::Directive::SkipDirective,
       GraphQL::Directive::IncludeDirective,
       GraphQL::Directive::DeferDirective,
+      GraphQL::Directive::StreamDirective,
     ]
     DYNAMIC_FIELDS = ["__type", "__typename", "__schema"]
 

lib/graphql/static_validation/rules/directives_are_defined.rb

@@ -15,6 +15,7 @@ def validate(context)
       def validate_directive(ast_directive, directive_names, errors)
         if !directive_names.include?(ast_directive.name)
           errors << message("Directive @#{ast_directive.name} is not defined", ast_directive)
+          GraphQL::Language::Visitor::SKIP
         end
       end
     end

spec/graphql/execution/deferred_execution_spec.rb

@@ -40,23 +40,23 @@ def merged_result
     DummySchema.query_execution_strategy = @prev_execution_strategy
   end
 
-  let(:query_string) {%|
-    {
-      cheese(id: 1) {
-        id
-        flavor
-        origin @defer
-        cheeseSource: source @defer
-      }
-    }
-  |}
-
   let(:collector) { ArrayCollector.new }
   let(:result) {
     DummySchema.execute(query_string, context: {collector: collector})
   }
 
   describe "@defer-ed fields" do
+    let(:query_string) {%|
+      {
+        cheese(id: 1) {
+          id
+          flavor
+          origin @defer
+          cheeseSource: source @defer
+        }
+      }
+    |}
+
     it "emits them later" do
       result
       assert_equal 3, collector.patches.length
@@ -137,7 +137,7 @@ def merged_result
         }
       }
       |}
-    it "patches the list, then the members" do
+    it "patches the whole list, then the member fields" do
       result
       assert_equal 8, collector.patches.length
       expected_patches = [
@@ -239,4 +239,31 @@ def merged_result
       end
     end
   end
+
+  describe "@stream-ed fields" do
+    let(:query_string) {%|
+    {
+      cheeses @stream {
+        id
+      }
+    }
+    |}
+
+    it "pushes an empty list, then each item" do
+      result
+
+      assert_equal(4, collector.patches.length)
+      assert_equal([], collector.patches[0][:path])
+      assert_equal({"data" => { "cheeses" => [] } }, collector.patches[0][:value])
+      assert_equal(["data", "cheeses", 0], collector.patches[1][:path])
+      assert_equal({"id"=>1}, collector.patches[1][:value])
+      assert_equal(["data", "cheeses", 1], collector.patches[2][:path])
+      assert_equal({"id"=>2}, collector.patches[2][:value])
+      assert_equal(["data", "cheeses", 2], collector.patches[3][:path])
+      assert_equal({"id"=>3}, collector.patches[3][:value])
+    end
+
+    it "supports lazy enumeration"
+    it "defers nested defers"
+  end
 end

spec/graphql/introspection/directive_type_spec.rb

@@ -50,6 +50,14 @@
             "onFragment"=>true,
             "onOperation"=>false,
           },
+          {
+            "name"=>"stream",
+            "args"=>[],
+            "locations"=>["FIELD"],
+            "onField"=>true,
+            "onFragment"=>false,
+            "onOperation"=>false,
+          },
         ]
       }
     }}