Add full support for subscription

Implemented according to https://spec.graphql.org/October2021/#sec-Subscription

Author: seriyps

README.md

@@ -351,14 +351,16 @@ inject() ->
       'Faction' => faction_resource,
       ...
       'Query' => query_resource,
-      'Mutation' => mutation_resource
+      'Mutation' => mutation_resource,
+      'Subscription' => subscription_resource
     }
   },
   ok = graphql:load_schema(Map, SchemaData),
       Root = {root,
       #{
         query => 'Query',
         mutation => 'Mutation',
+        subscription => 'Subscription',
         interfaces => []
       }},
   ok = graphql:insert_schema_definition(Root),
@@ -386,10 +388,14 @@ run(Doc, OpName, Vars, Req, State) ->
           ok = graphql:validate(AST2),
           Coerced = graphql:type_check_params(FunEnv, OpName, Vars),
           Ctx = #{ params => Coerced, operation_name => OpName },
-          Response = graphql:execute(Ctx, AST2),
-          Req2 = cowboy_req:set_resp_body(encode_json(Response), Req),
-          {ok, Reply} = cowboy_req:reply(200, Req2),
-          {halt, Reply, State}
+          case graphql:execute(Ctx, AST2) of
+              {subscription, Subscription, SubscriptionContext} ->
+                  {cowboy_loop, Req, store_subscription(Subscription, SubscriptionContext, State)};
+              Response ->
+                  Req2 = cowboy_req:set_resp_body(encode_json(Response), Req),
+                  {ok, Reply} = cowboy_req:reply(200, Req2),
+                  {halt, Reply, State}
+          end
       catch
             throw:Err ->
                 err(400, Err, Req, State)
@@ -579,8 +585,15 @@ handling field requests in that object. The module looks like:
 
 execute(Ctx, SrcObj, <<"f">>, Args) ->
     {ok, 42};
+execute(Ctx, SrcObj, <<"heavy">>, Args) ->
+    Token = graphql:token(Ctxt),
+    Worker = spawn_link(fun() ->
+      Result = calculate_heavy_value(Args),
+      graphql:reply_cast(Token, {ok, Result})
+    end),
+    {defer, Token, #{timeout => 5000, worker => Worker}};
 execute(Ctx, SrcObj, Field, Args) ->
-    default
+    default...
 ```
 
 The only function which is needed is the `execute/4` function which is
@@ -599,9 +612,62 @@ called by the system whenever a field is requested in that object. The
   of a backing store and then moving the *cursor* onto that object and
   calling the correct object resource for that type. The `SrcObj` is
   set to point to the object that is currently being operated upon.
+  For `query` and `mutation` the **root** field `SrcObj` is set to `none`, for
+  `subscription` it is set to `{subscription_value(), Event}`.
 * `Field` - The field in the object which is requested.
 * `Args` - A map of field arguments. See the next section.
 
+Possible return values of `execute/4` callback are:
+
+* `{ok, Value}` - the value of the field (either scalar for scalar fields or arbitrarily
+  complex object for object fields); `Value` can be `null` for nullable fields
+* `{ok, [{ok, Value}, ..]}` for array fields
+* `{ok, Val, [Auxilary, ..]}` all the `Auxilary` lists returned during operation execution will be
+  concatenated and returned as a `.aux` field in the result map.
+* `{error, Reason}` for execution errors (`Reason` is either `atom()` or `binary()` or record
+  that will be used to construct the resulting `.errors` field)
+* `{defer, Token, undefined | #{timeout => timeout(), worker => pid()}}` tells graphql execution
+  engine that the value of the field is calculated asynchronously and will be delivered later to it
+  via `graphql:reply_cast/2`.
+  *  `timeout` - tells graphql engine to give-up waiting for the value after this many milliseconds
+  *  `worker` - tells graphql engine that the value is calculated by this Erlang process (worker),
+     so the engine will monitor this process and will report error if the process crashes.
+* `{defer, Token}` same as `{defer, Token, undefined}`
+
+Any value can be either returned from the function or thrown using `graphql:throw/1`.
+
+### Subscription resolution modules
+
+Subscription root object is mapped onto an Erlang module responsible for
+initiation of the subscription and for handling field requests in received events.
+The module is similar to [Output object Resources](#output-object-resources) but
+they have one extra callback.
+
+```erlang
+-module(object_resource).
+
+-export([subscribe/3, execute/4]).
+
+subscribe(Ctx, <<"commentAdded">>, #{<<"topic">> := TopicId}) ->
+    Subscription = #my_subscription{} = topic:subscribe_new_comments(TopicId),
+    {subscription, Subscription}.
+
+execute(Ctx, {#my_subscription{}, Comment}, <<"commentAdded">>, _Args) ->
+    {ok, Comment}.
+
+```
+
+The `subscribe/3` callback is called whenever `subscription` operation is
+executed (it returns `{subscription, Subscription, OpaqueCtx}`). This callback
+should initialize the subscription and return `{subscription, Subscription}`
+tuple, where `Subscription` is whatever could be necessary to properly map
+the events received from pub-sub system to the correct subscription.
+
+Whenever the new event `Event` is received from pubs-sub system,
+`graphql:handle_subscription_event(ExtraCtx, Subscription, OpaqueCtx, Event)`
+must be called which will call `execute/4` callback which will act the same
+way as described in [Output object Resources](#output-object-resources).
+
 #### Field Argument rules
 
 In GraphQL, field arguments follow a specific pattern:

src/graphql.erl

@@ -12,7 +12,8 @@
          type_check/1, type_check_params/3,
          insert_root/1,
          validate/1,
-         execute/1, execute/2
+         execute/1, execute/2,
+         handle_subscription_event/4
         ]).
 
 -export([
@@ -45,9 +46,15 @@
 -type json() :: number() | binary() | true | false | null | #{ binary() | atom() => json() } | [json()] .
 -type param_context() :: json().
 
+-type execute_result() :: #{data => graphql:json(),
+                            errors => [#{message := binary(),
+                                         path => [binary()],
+                                         extensions => map()}],
+                            aux => [any()]}.
+
 -type schema_definition() :: {atom(), #{ atom() => term() }}.
 
--export_type([json/0, param_context/0]).
+-export_type([json/0, param_context/0, execute_result/0]).
 
 -type token() :: {'$graphql_token', pid(), reference(), reference()}.
 -type defer_map() :: #{ worker => pid(),
@@ -57,10 +64,15 @@
 -type name() :: {name, pos_integer(), binary()} | binary().
 -type document() :: #document{}.
 -type directive() :: #directive{}.
--export_type([directive/0,
 
+-type subscription_value() :: any().
+-type subscription_ctx() :: graphql_execute:subscription_ctx().
+
+-export_type([directive/0,
               token/0,
-              schema_field/0]).
+              schema_field/0,
+              subscription_value/0,
+              subscription_ctx/0]).
 
 -define(DEFAULT_TIMEOUT, 3000).
 
@@ -153,12 +165,14 @@ type_check(AST) ->
 type_check_params(FunEnv, OpName, Vars) ->
     graphql_check:check_params(FunEnv, OpName, Vars).
 
--spec execute(document()) -> #{ atom() => json() }.
+-spec execute(document()) -> execute_result()
+              | {subscription, subscription_value(), subscription_ctx()}.
 execute(AST) ->
     Ctx = #{ params => #{}, default_timeout => ?DEFAULT_TIMEOUT },
     execute(Ctx, AST).
 
--spec execute(context(), document()) -> #{ atom() => json() }.
+-spec execute(context(), document()) -> execute_result()
+              | {subscription, subscription_value(), subscription_ctx()}.
 execute(#{default_timeout := _DT } = Ctx, AST) ->
     graphql_execute:x(Ctx, AST);
 execute(Ctx, AST) ->
@@ -168,6 +182,18 @@ execute(Ctx, AST) ->
         Result -> Result
     end.
 
+%% @doc Executes subscription MapSourceToResponseEvent step
+%% `subscription_value()' and `subscription_ctx()' are the values returned as
+%% `{subscription, SubValue, SubCtx} = graphql:execute(...)'
+%% A tuple `{Subscription, Event}' will be passed as 2nd argument to the `execute/4' callback
+%%
+%% @param ExtraCtx additional map fields that needs to be added to execution context
+%% @param Event arbitrary event that was produced by the subscription
+-spec handle_subscription_event(context(), subscription_value(), subscription_ctx(), any()) ->
+          #{ atom() => json() }.
+handle_subscription_event(ExtraCtx, Subscription, SubscriptionCtx, Event) ->
+    graphql_execute:x_subscription_event(ExtraCtx, Subscription, SubscriptionCtx, Event).
+
 %% @doc insert_schema_definition/1 loads a schema definition into the Graph Schema
 %% @end
 -spec insert_schema_definition(schema_definition()) -> ok | {error, Reason}

src/graphql_execute.erl

@@ -8,8 +8,10 @@
 -compile(inline_list_funcs).
 -compile({inline_size, 50}).
 
--export([x/1, x/2]).
+-export([x/1, x/2, x_subscription_event/4]).
 -export([builtin_input_coercer/1]).
+-export_type([subscription_ctx/0]).
+
 -type source() :: reference().
 -type demonitor() :: {reference(), pid()} .
 
@@ -74,27 +76,50 @@
           work = #{} :: #{ source() => defer_closure() },
           timeout :: non_neg_integer() }).
 
--spec x(graphql:ast()) -> #{ atom() => graphql:json() }.
+-record(subscription,
+        { orig_context,
+          op,
+          fragments}).
+
+-opaque subscription_ctx() :: #subscription{}.
+
+-spec x(graphql:ast()) ->
+          graphql:execute_result()
+        | {subscription, graphql:subscription_value(), subscription_ctx()}.
 x(X) -> x(#{ params => #{} }, X).
 
--spec x(term(), graphql:ast()) -> #{ atom() => graphql:json() }.
+-spec x(term(), graphql:ast()) ->
+          graphql:execute_result()
+        | {subscription, graphql:subscription_value(), subscription_ctx()}.
 x(Ctx, X) ->
     Canon = canon_context(Ctx),
     execute_request(Canon, X).
 
+-spec x_subscription_event(map(), graphql:subscription_value(), subscription_ctx(), any()) ->
+          #{atom() => graphql:json()}.
+x_subscription_event(ExtraCtx, Subscription,
+                     #subscription{orig_context = OrigContext,
+                                   op = Op,
+                                   fragments = Frags}, Event) ->
+    CanonCtx = canon_context(maps:merge(OrigContext, ExtraCtx)),
+    Ctx = CanonCtx#ectx{ frags = Frags,
+                         defer_request_id = make_ref() },
+    InitialValue = {Subscription, Event},
+    execute_query(Ctx#ectx{ op_type = subscription }, Op, InitialValue).
+
 execute_request(InitialCtx, #document { definitions = Operations }) ->
     {Frags, Ops} = lists:partition(fun (#frag {}) -> true;(_) -> false end, Operations),
     Ctx = InitialCtx#ectx{ frags = fragments(Frags),
                            defer_request_id = make_ref() },
     case get_operation(Ctx, Ops) of
         {ok, #op { ty = {query, _} } = Op } ->
-            execute_query(Ctx#ectx{ op_type = query }, Op);
-        {ok, #op { ty = {subscription, _} } = Op } ->
-            execute_query(Ctx#ectx{ op_type = subscription }, Op);
+            execute_query(Ctx#ectx{ op_type = query }, Op, none);
         {ok, #op { ty = undefined } = Op } ->
-            execute_query(Ctx#ectx{ op_type = query }, Op);
+            execute_query(Ctx#ectx{ op_type = query }, Op, none);
         {ok, #op { ty = {mutation, _} } = Op } ->
             execute_mutation(Ctx#ectx{ op_type = mutation }, Op);
+        {ok, #op { ty = {subscription, _} } = Op } ->
+            create_source_event_stream(Ctx#ectx{ op_type = subscription }, Op);
         {error, Reason} ->
             {error, Errs} = err(Ctx, Reason),
             complete_top_level(undefined, Errs)
@@ -124,12 +149,13 @@ collect_auxiliary_data() ->
     end.
 
 execute_query(#ectx{ defer_request_id = ReqId } = Ctx,
-              #op { selection_set = SSet, schema = QType }) ->
+              #op { selection_set = SSet, schema = QType },
+              InitialValue) ->
     #object_type{} = QType,
     case execute_sset(Ctx#ectx{ path = [],
                                 defer_process = self(),
                                 defer_target = top_level },
-                      SSet, QType, none) of
+                      SSet, QType, InitialValue) of
         {ok, Res, Errs} ->
             complete_top_level(Res, Errs);
         #work { items = WL, monitor = Ms, demonitors = [] , timeout = TimeOut} ->
@@ -154,6 +180,34 @@ execute_mutation(Ctx, #op { selection_set = SSet,
             complete_top_level(Res, Errs)
     end.
 
+%% 6.2.3.1 https://spec.graphql.org/October2021/#sec-Source-Stream
+create_source_event_stream(#ectx{ frags = Frags, ctx = OrigCtx} = Ctx,
+                           #op { selection_set = SSet,
+                                 schema = QType } = Op) ->
+    #object_type{} = QType,
+    GroupedFieldSet = collect_fields(Ctx, QType, SSet),
+    case orddict:to_list(GroupedFieldSet) of
+        [{Key, [Field]}] ->
+            FieldName = name(Field),
+            #schema_field { directives = Directives } = lookup_field(Field, QType),
+            Args = resolve_args(Ctx, Field),
+            Fun = subscribe_resolver_function(QType),
+            CtxP = add_path(Ctx, Key),
+            case resolve_field_event_stream(CtxP, QType, FieldName, Directives, Fun, Args) of
+                {subscription, Sub} ->
+                    SubCtx = #subscription{orig_context = OrigCtx,
+                                           op = Op,
+                                           fragments = Frags},
+                    {subscription, Sub, SubCtx};
+                {error, Reason} ->
+                    {error, Errs} = err(CtxP, Reason),
+                    complete_top_level(undefined, Errs)
+            end;
+        _Other ->
+            {error, Errs} = err(Ctx, subscription_must_have_one_root_field),
+            complete_top_level(undefined, Errs)
+    end.
+
 execute_sset(#ectx{ defer_target = DeferTarget } = Ctx, SSet, Type, Value) ->
     GroupedFields = collect_fields(Ctx, Type, SSet),
     Self = make_ref(),
@@ -557,6 +611,44 @@ resolve_field_value(#ectx { op_type = OpType,
             {error, {resolver_crash, M}}
     end.
 
+resolve_field_event_stream(#ectx { op_type = OpType,
+                            ctx = CallerContext },
+                    #object_type { id = OID,
+                                   directives = ODirectives} = ObjectType,
+                    Name, FDirectives, Fun, Args) ->
+    AnnotatedCallerCtx =
+        CallerContext#{ op_type => OpType,
+                        field => Name,
+                        field_directives => format_directives(FDirectives),
+                        object_type => OID,
+                        object_directives => format_directives(ODirectives)
+    },
+    try Fun(AnnotatedCallerCtx, Name, Args) of
+        V ->
+            case handle_subscribe_resolver_result(V) of
+                wrong ->
+                    Obj = graphql_schema:id(ObjectType),
+                    report_wrong_return(Obj, Name, Fun, V),
+                    {error, {wrong_resolver_return, {Obj, Name}}};
+                Res -> Res
+            end
+    catch
+        throw:{'$graphql_throw', Msg} ->
+            case handle_subscribe_resolver_result(Msg) of
+                wrong ->
+                    Obj = graphql_schema:id(ObjectType),
+                    report_wrong_return(Obj, Name, Fun, Msg),
+                    {error, {wrong_resolver_return, {Obj, Name}}};
+                Res -> Res
+            end;
+        ?EXCEPTION(Cl, Err, Stacktrace) ->
+            M = #{ type => graphql_schema:id(ObjectType),
+                   field => Name,
+                   stack => ?GET_STACK(Stacktrace),
+                   class => Cl,
+                   error => Err},
+            {error, {resolver_crash, M}}
+    end.
 
 handle_resolver_result({error, Reason}) ->
     {error, {resolver_error, Reason}};
@@ -571,6 +663,12 @@ handle_resolver_result({defer, Token, DeferStateMap}) ->
     {defer, Token, DeferStateMap};
 handle_resolver_result(_Unknown) -> wrong.
 
+handle_subscribe_resolver_result({subscription, V}) ->
+    {subscription, V};
+handle_subscribe_resolver_result({error, Reason}) ->
+    {error, {resolver_error, Reason}};
+handle_subscribe_resolver_result(_Unknown) -> wrong.
+
 complete_value(Ctx, Ty, Fields, {ok, Value}) when is_binary(Ty) ->
     error_logger:warning_msg(
       "Canary: Type lookup during value completion for: ~p~n",
@@ -929,6 +1027,9 @@ resolver_function(#object_type {
 resolver_function(#object_type { resolve_module = M }, undefined) ->
     fun M:execute/4.
 
+subscribe_resolver_function(#object_type { resolve_module = M }) ->
+    fun M:subscribe/3.
+
 %% -- OUTPUT COERCION ------------------------------------
 
 builtin_input_coercer(X) ->