_site/cover/resp_connection.COVER.html

1 %% Copyright (c) 2023 Peter Morgan <peter.james.morgan@gmail.com>
2 %%
3 %% Licensed under the Apache License, Version 2.0 (the "License");
4 %% you may not use this file except in compliance with the License.
5 %% You may obtain a copy of the License at
6 %%
7 %% http://www.apache.org/licenses/LICENSE-2.0
8 %%
9 %% Unless required by applicable law or agreed to in writing, software
10 %% distributed under the License is distributed on an "AS IS" BASIS,
11 %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 %% See the License for the specific language governing permissions and
13 %% limitations under the License.
14
15
16 -module(resp_connection).
17
18
19 -export([callback_mode/0]).
20 -export([handle_event/4]).
21 -export([init/1]).
22 -export([start/0]).
23 -export([start/1]).
24 -export([start_link/0]).
25 -export([start_link/1]).
26 -export([terminate/3]).
27 -import(resp_statem, [nei/1]).
28 -include_lib("kernel/include/inet.hrl").
29 -include_lib("kernel/include/logger.hrl").
30
31
32 start() ->
33 
:-(
 ?FUNCTION_NAME(#{}).
34
35
36 start(Arg) ->
37 
:-(
 gen_statem:start(?MODULE, [Arg], envy_gen:options(?MODULE)).
38
39
40 start_link() ->
41 
:-(
 ?FUNCTION_NAME(#{}).
42
43
44 start_link(Arg) ->
45 
:-(
 gen_statem:start_link(?MODULE, [Arg], envy_gen:options(?MODULE)).
46
47
48 init([Arg]) ->
49 
:-(
 process_flag(trap_exit, true),
50 
:-(
 {ok,
51 ready,
52 #{arg => Arg,
53 partial => <<>>,
54 requests => gen_statem:reqids_new()},
55 nei(recv)}.
56
57
58 callback_mode() ->
59 
:-(
 handle_event_function.
60
61
62 handle_event(
63 internal,
64 {callback, F, A},
65 _,
66 #{arg := #{callback := #{module := M} = Callback} = Arg} = Data) ->
67 
:-(
 try apply(M, F, A) of
68 {continue, CallbackData, Actions} when is_list(Actions) ->
69 
:-(
 {keep_state,
70 Data#{arg := Arg#{callback := Callback#{data := CallbackData}}},
71 
:-(
 [nei(Action) || Action <- Actions]};
72
73 {continue, CallbackData, Action} ->
74 
:-(
 {keep_state,
75 Data#{arg := Arg#{callback := Callback#{data := CallbackData}}},
76 nei(Action)};
77
78 {continue, Actions} when is_list(Actions) ->
79 
:-(
 {keep_state_and_data, [nei(Action) || Action <- Actions]};
80
81 {continue, Action} ->
82 
:-(
 {keep_state_and_data, nei(Action)};
83
84 continue ->
85 
:-(
 keep_state_and_data;
86
87 stop ->
88 
:-(
 stop;
89
90 {stop, Reason} ->
91 
:-(
 {stop, Reason}
92 catch
93 Class:Exception:Stacktrace ->
94 
:-(
 ?LOG_ERROR(#{class => Class,
95 exception => Exception,
96 
:-(
 stacktrace => Stacktrace}),
97 
:-(
 {keep_state_and_data,
98 nei({encode, {error, "not implemented"}})}
99 end;
100
101 handle_event(info,
102 {'$socket', Socket, select, Handle},
103 _,
104 #{arg := #{socket := Socket},
105 partial := Partial} = Data) ->
106 
:-(
 case socket:recv(Socket, 0, Handle) of
107 {ok, Received} ->
108 
:-(
 {keep_state,
109 Data#{partial := <<>>},
110 [nei({telemetry,
111 recv,
112 #{bytes => iolist_size(Received)}}),
113 nei({recv, iolist_to_binary([Partial, Received])}),
114 nei(recv)]};
115
116 {select, {select_info, _, _}} ->
117 
:-(
 keep_state_and_data;
118
119 {error, econnreset} ->
120 
:-(
 stop;
121
122 {error, closed} ->
123 
:-(
 stop;
124
125 {error, Reason} ->
126 
:-(
 {stop, Reason}
127 end;
128
129 handle_event(internal,
130 {recv, <<_:8, _/bytes>>= Encoded},
131 _,
132 #{arg := #{callback := #{data := CallbackData}}}) ->
133 
:-(
 case resp_codec:decode(Encoded) of
134 {{array, [{bulk, Command} | _]} = Decoded, Remainder} ->
135 
:-(
 try
136 
:-(
 {keep_state_and_data,
137 [nei({callback,
138 recv,
139 [#{message => Decoded,
140 command => command(string:casefold(Command)),
141 data => CallbackData}]}),
142
143 nei({telemetry,
144 decode,
145 #{count => 1},
146 #{message => Decoded}}),
147
148 nei({recv, Remainder})]}
149
150 catch
151 error:badarg ->
152 
:-(
 {keep_state_and_data,
153 [nei({encode, {error, ["unknown command '", Command, "'"]}}),
154 nei({recv, Remainder})]}
155 end;
156
157 {_, Remainder} ->
158 
:-(
 {keep_state_and_data,
159 [nei({encode, {error, ["syntax"]}}), nei({recv, Remainder})]}
160 end;
161
162 handle_event(internal, {recv, <<>>}, _, _) ->
163 
:-(
 keep_state_and_data;
164
165 handle_event(internal, {encode = EventName, Decoded}, _, _) ->
166 
:-(
 {keep_state_and_data,
167 [nei({telemetry,
168 EventName,
169 #{count => 1},
170 #{message => Decoded}}),
171 nei({send, resp_codec:encode(Decoded)})]};
172
173 handle_event(internal,
174 {send = EventName, Encoded},
175 _,
176 #{arg := #{socket := Socket}}) ->
177 
:-(
 case socket:send(Socket, Encoded) of
178 ok ->
179 
:-(
 {keep_state_and_data,
180 nei({telemetry,
181 EventName,
182 #{bytes => iolist_size(Encoded)}})};
183
184 {error, econnreset} ->
185 
:-(
 stop;
186
187 {error, closed} ->
188 
:-(
 stop;
189
190 {error, Reason} ->
191 
:-(
 {stop, Reason}
192 end;
193
194 handle_event(internal,
195 recv = EventName,
196 _,
197 #{arg := #{socket := Socket},
198 partial := Partial} = Data) ->
199 
:-(
 case socket:recv(Socket, 0, nowait) of
200 {ok, Received} ->
201 
:-(
 {keep_state,
202 Data#{partial := <<>>},
203 [nei({telemetry,
204 EventName,
205 #{bytes => iolist_size(Received)}}),
206 nei({recv, iolist_to_binary([Partial, Received])}),
207 nei(recv)]};
208
209 {select, {select_info, _, _}} ->
210 
:-(
 keep_state_and_data;
211
212 {error, econnreset} ->
213 
:-(
 stop;
214
215 {error, closed} ->
216 
:-(
 stop;
217
218 {error, Reason} ->
219 
:-(
 {stop, Reason}
220 end;
221
222 handle_event(internal,
223 {telemetry, EventName, Measurements},
224 _,
225 _) ->
226 
:-(
 {keep_state_and_data,
227 nei({telemetry, EventName, Measurements, #{}})};
228
229 handle_event(internal,
230 {telemetry, EventName, Measurements, Metadata},
231 _,
232 #{arg := #{callback := #{module := M}}}) ->
233 
:-(
 ok = telemetry:execute(
234 [resp, connection, EventName],
235 Measurements,
236 maps:merge(
237 #{callback => M},
238 Metadata)),
239 
:-(
 keep_state_and_data;
240
241 handle_event(info, Msg, _, #{requests := Existing} = Data) ->
242 
:-(
 case gen_statem:check_response(Msg, Existing, true) of
243 {{reply, ok}, expire, Updated} ->
244 
:-(
 {keep_state, Data#{requests := Updated}};
245
246 {{reply, ok}, flush_all, Updated} ->
247 
:-(
 {keep_state, Data#{requests := Updated}};
248
249 no_request ->
250 
:-(
 {keep_state_and_data, nei({callback, info, [Msg]})};
251
252 no_reply ->
253 
:-(
 ?LOG_ERROR(#{msg => Msg, data => Data}),
254 
:-(
 keep_state_and_data
255 end.
256
257
258 terminate(_Reason, _State, #{arg := #{socket := Socket}}) ->
259 
:-(
 _ = socket:close(Socket);
260
261 terminate(_Reason, _State, _Data) ->
262 
:-(
 ok.
263
264
265 command(Command) ->
266 
:-(
 #{name => binary_to_atom(Command),
267 info => resp_command:info(string:casefold(Command))}.
Line Hits Source