1 |
|
%% @doc An SSE handler for GraphQL subscriptions. |
2 |
|
%% The graphql request is prepared, and then executed. |
3 |
|
%% |
4 |
|
%% 1. The first execution should return 'null' in 'data', and a new Stream in 'aux'. |
5 |
|
%% 2. Then, whenever an Event is received, the prepared GraphQL request |
6 |
|
%% is executed again, this time with the Stream and the Event in the context. |
7 |
|
%% The resolver should then return either 'null' or the processed Event to send to the client. |
8 |
|
%% 3. Upon termination, the request is executed one last time with the 'terminate' event. |
9 |
|
%% This is an opportunity to clean up all stream resources. |
10 |
|
-module(mongoose_graphql_sse_handler). |
11 |
|
|
12 |
|
-behaviour(lasse_handler). |
13 |
|
-export([init/3, |
14 |
|
handle_notify/2, |
15 |
|
handle_info/2, |
16 |
|
handle_error/3, |
17 |
|
terminate/3]). |
18 |
|
|
19 |
|
-include("mongoose.hrl"). |
20 |
|
|
21 |
|
-type req() :: cowboy_req:req(). |
22 |
|
-type state() :: #{atom() => term()}. |
23 |
|
|
24 |
|
-spec init(state(), any(), cowboy_req:req()) -> |
25 |
|
{ok, req(), state()} | |
26 |
|
{shutdown, cowboy:http_status(), cowboy:http_headers(), iodata(), req(), state()}. |
27 |
|
init(#{sse_idle_timeout := Timeout} = State, _LastEvtId, Req) -> |
28 |
10 |
process_flag(trap_exit, true), % needed for 'terminate' to be called |
29 |
10 |
cowboy_req:cast({set_options, #{ |
30 |
|
idle_timeout => Timeout |
31 |
|
}}, Req), |
32 |
10 |
case cowboy_req:method(Req) of |
33 |
|
<<"GET">> -> |
34 |
9 |
case mongoose_graphql_handler:check_auth_header(Req, State) of |
35 |
|
{ok, State2} -> |
36 |
8 |
case mongoose_graphql_handler:gather(Req) of |
37 |
|
{ok, Req2, Decoded} -> |
38 |
7 |
run_request(Decoded, Req2, State2); |
39 |
|
{error, Reason} -> |
40 |
1 |
reply_error(Reason, Req, State) |
41 |
|
end; |
42 |
|
{error, Reason} -> |
43 |
1 |
reply_error(Reason, Req, State) |
44 |
|
end; |
45 |
|
_ -> |
46 |
1 |
{ok, Req, State} % lasse returns 405: Method Not Allowed |
47 |
|
end. |
48 |
|
|
49 |
|
run_request(#{document := undefined}, Req, State) -> |
50 |
1 |
reply_error(make_error(decode, no_query_supplied), Req, State); |
51 |
|
run_request(GQLReq, Req, #{schema_endpoint := EpName, authorized := AuthStatus} = State) -> |
52 |
6 |
Ep = mongoose_graphql:get_endpoint(EpName), |
53 |
6 |
Ctx = maps:get(schema_ctx, State, #{}), |
54 |
6 |
GQLReq2 = GQLReq#{authorized => AuthStatus, ctx => Ctx#{method => sse}}, |
55 |
6 |
case mongoose_graphql:prepare(Ep, GQLReq2) of |
56 |
|
{error, Reason} -> |
57 |
2 |
reply_error(Reason, Req, State); |
58 |
|
{ok, GQLReq3 = #{ctx := Ctx2}} -> |
59 |
4 |
case mongoose_graphql:execute(Ep, GQLReq3) of |
60 |
|
{ok, #{aux := [{stream, Stream}]}} -> |
61 |
3 |
Ctx3 = Ctx2#{stream => Stream}, |
62 |
3 |
{ok, Req, State#{id => 1, ep => Ep, req => GQLReq3#{ctx := Ctx3}}}; |
63 |
|
{ok, Response} -> |
64 |
1 |
Body = mongoose_graphql_response:term_to_json(Response), |
65 |
1 |
{shutdown, 200, #{}, Body, Req, State} |
66 |
|
end |
67 |
|
end. |
68 |
|
|
69 |
|
-spec handle_notify(term(), state()) -> {nosend, state()}. |
70 |
|
handle_notify(Msg, State) -> |
71 |
:-( |
?UNEXPECTED_INFO(Msg), |
72 |
:-( |
{nosend, State}. |
73 |
|
|
74 |
|
-spec handle_info(term(), state()) -> {nosend, state()} | {send, lasse_handler:event(), state()}. |
75 |
|
handle_info(Event, State = #{ep := Ep, req := Req = #{ctx := Ctx}, id := Id}) -> |
76 |
5 |
Ctx1 = Ctx#{event => Event}, |
77 |
5 |
{ok, #{data := Data} = Response} = mongoose_graphql:execute(Ep, Req#{ctx := Ctx1}), |
78 |
5 |
case has_non_null_value(Data) of |
79 |
|
false -> |
80 |
1 |
{nosend, State}; |
81 |
|
true -> |
82 |
4 |
EventData = mongoose_graphql_response:term_to_json(Response), |
83 |
4 |
SseEvent = #{id => integer_to_binary(Id), data => EventData}, |
84 |
4 |
{send, SseEvent, State#{id := Id + 1}} |
85 |
|
end. |
86 |
|
|
87 |
|
%% Check if there is any value that is non-null. Any list is considered non-null. |
88 |
|
has_non_null_value(M) when is_map(M) -> |
89 |
14 |
lists:any(fun has_non_null_value/1, maps:values(M)); |
90 |
5 |
has_non_null_value(V) -> V =/= null. |
91 |
|
|
92 |
|
-spec handle_error(iodata(), term(), state()) -> ok. |
93 |
|
handle_error(Msg, Reason, _State) -> |
94 |
:-( |
?LOG_ERROR(#{what => mongoose_graphql_sse_handler_failed, |
95 |
:-( |
reason => Reason, text => Msg}). |
96 |
|
|
97 |
|
-spec terminate(term(), req(), state()) -> ok. |
98 |
|
terminate(_Reason, _Req, #{ep := Ep, req := Req = #{ctx := Ctx}}) -> |
99 |
3 |
Ctx1 = Ctx#{event => terminate}, |
100 |
3 |
{ok, #{aux := [{stream, closed}]}} = mongoose_graphql:execute(Ep, Req#{ctx := Ctx1}), |
101 |
3 |
ok; |
102 |
|
terminate(_Reason, _Req, #{}) -> |
103 |
6 |
ok. |
104 |
|
|
105 |
|
make_error(Phase, Term) -> |
106 |
1 |
#{error_term => Term, phase => Phase}. |
107 |
|
|
108 |
|
reply_error(Reason, Req, State) -> |
109 |
5 |
{Code, Error} = mongoose_graphql_errors:format_error(Reason), |
110 |
5 |
Body = jiffy:encode(#{errors => [Error]}), |
111 |
5 |
{shutdown, Code, #{}, Body, Req, State}. |