_site/cover/resp_client.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_client).
17
18
19 -export([callback_mode/0]).
20 -export([handle_event/4]).
21 -export([init/1]).
22 -export([send/1]).
23 -export([start/0]).
24 -export([start/1]).
25 -export([start_link/0]).
26 -export([start_link/1]).
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], []).
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 send(#{data := Data} = Arg) ->
49 
:-(
 send_request(
50 maps:without(
51 [data],
52 Arg#{request => {?FUNCTION_NAME, Data}})).
53
54
55 send_request(#{label := _} = Arg) ->
56 
:-(
 resp_statem:send_request(Arg);
57
58 send_request(Arg) ->
59 
:-(
 resp_statem:send_request(Arg#{label => ?MODULE}).
60
61
62 init([Arg]) ->
63 
:-(
 process_flag(trap_exit, true),
64 
:-(
 {ok, disconnected, Arg#{requests => gen_statem:reqids_new()}}.
65
66
67 callback_mode() ->
68 
:-(
 handle_event_function.
69
70
71 handle_event({call, _}, {send, _}, disconnected, Data) ->
72 
:-(
 {next_state, connecting, Data, [postpone, nei(open)]};
73
74 handle_event({call, _}, {send, _}, connecting, _) ->
75 
:-(
 {keep_state_and_data, postpone};
76
77 handle_event({call, _}, {send, _}, {busy, _}, _) ->
78 
:-(
 {keep_state_and_data, postpone};
79
80 handle_event({call, From}, {send, _} = Send, connected, Data) ->
81 
:-(
 {next_state,
82 {busy, From},
83 Data#{replies => []},
84 nei(Send)};
85
86 handle_event(internal, open = EventName, connecting, Data) ->
87 
:-(
 case socket:open(inet, stream, default) of
88 {ok, Socket} ->
89 
:-(
 {keep_state,
90 Data#{socket => Socket},
91 [nei({telemetry, EventName, #{count => 1}}),
92 nei(connect)]};
93
94 {error, Reason} ->
95 
:-(
 {stop, Reason}
96 end;
97
98 handle_event(internal, {send, #{} = Decoded}, {busy, _}, _) ->
99 
:-(
 {keep_state_and_data, nei({send, [Decoded]})};
100
101 handle_event(internal,
102 {send = EventName, Decoded},
103 {busy, _},
104 #{socket := Socket}) ->
105 
:-(
 Encoded = resp_codec:encode(Decoded),
106 
:-(
 case socket:send(Socket, Encoded) of
107 ok ->
108 
:-(
 {keep_state_and_data,
109 [nei({telemetry,
110 EventName,
111 #{count => 1,
112 bytes => iolist_size(Encoded)},
113 #{messages => Decoded}}),
114 nei({reply_expected, resp_protocol:reply_expected(Decoded)})]};
115
116 {error, Reason} ->
117 
:-(
 {stop, Reason}
118 end;
119
120 handle_event(internal, {reply_expected, []}, {busy, From}, Data) ->
121 
:-(
 {next_state,
122 connected,
123 maps:without([replies], Data),
124 {reply, From, ok}};
125
126 handle_event(internal, {reply_expected, ReplyExpected}, {busy, _}, Data) ->
127 
:-(
 {keep_state, Data#{reply_expected => ReplyExpected}};
128
129 handle_event(internal,
130 recv,
131 _,
132 #{socket := Socket, partial := Partial} = Data) ->
133 
:-(
 case socket:recv(Socket, 0, nowait) of
134 {ok, Received} ->
135 
:-(
 {keep_state,
136 Data#{partial := <<>>},
137 [nei({telemetry, recv, #{bytes => iolist_size(Received)}}),
138 nei({recv, iolist_to_binary([Partial, Received])}),
139 nei(recv)]};
140
141 {select, {select_info, _, _}} ->
142 
:-(
 keep_state_and_data;
143
144 {error, Reason} ->
145 
:-(
 {stop, Reason}
146 end;
147
148 handle_event(info,
149 {'$socket', Socket, select, Handle},
150 _,
151 #{socket := Socket, partial := Partial} = Data) ->
152 
:-(
 case socket:recv(Socket, 0, Handle) of
153 {ok, Received} ->
154 
:-(
 {keep_state,
155 Data#{partial := <<>>},
156 [nei({telemetry, recv, #{bytes => iolist_size(Received)}}),
157 nei({recv, iolist_to_binary([Partial, Received])}),
158 nei(recv)]};
159
160 {select, {select_info, _, _}} ->
161 
:-(
 keep_state_and_data;
162
163 {error, Reason} ->
164 
:-(
 {stop, Reason}
165 end;
166
167 handle_event(info, Msg, _, #{requests := Existing} = Data) ->
168 
:-(
 case gen_statem:check_response(Msg, Existing, true) of
169 {{reply, Reply}, Label, Updated} ->
170 
:-(
 {keep_state,
171 Data#{requests := Updated},
172 nei({response, #{label => Label, reply => Reply}})};
173
174 {{error, {Reason, ServerRef}}, Label, UpdatedRequests} ->
175 
:-(
 {stop,
176 #{reason => Reason,
177 server_ref => ServerRef,
178 label => Label},
179 Data#{requests := UpdatedRequests}}
180 end;
181
182 handle_event(internal, {recv, Command}, _, _) ->
183 
:-(
 {keep_state_and_data, nei({decode, Command})};
184
185 handle_event(internal, {decode, <<>>}, _, _) ->
186 
:-(
 keep_state_and_data;
187
188 handle_event(internal, {decode, Encoded}, _, #{partial := <<>>} = Data) ->
189 
:-(
 case resp_codec:decode(Encoded) of
190 {Decoded, Remainder} ->
191 
:-(
 {keep_state_and_data,
192 [nei({message, Decoded}), nei({decode, Remainder})]};
193
194 partial ->
195 
:-(
 {keep_state, Data#{partial := Encoded}}
196 end;
197
198 handle_event(internal,
199 {message, Decoded},
200 {busy, From},
201 #{replies := Replies} = Data) ->
202 
:-(
 {next_state,
203 connected,
204 maps:without([replies, reply_expected], Data),
205 [{reply, From, lists:reverse([Decoded | Replies])},
206 nei({telemetry, message, #{count => 1}, #{message => Decoded}})]};
207
208 handle_event(internal,
209 {telemetry, EventName, Measurements},
210 _,
211 _) ->
212 
:-(
 {keep_state_and_data,
213 nei({telemetry, EventName, Measurements, #{}})};
214
215 handle_event(internal,
216 {telemetry, EventName, Measurements, Metadata},
217 _,
218 Data) ->
219 
:-(
 ok = telemetry:execute(
220 [resp, client, EventName],
221 Measurements,
222 maps:merge(
223 maps:with([socket], Data),
224 Metadata)),
225 
:-(
 keep_state_and_data;
226
227 handle_event(internal, connect, connecting, _) ->
228 
:-(
 {keep_state_and_data,
229 nei({connect,
230 #{family => inet,
231 port => resp_config:client(port),
232 addr => addr()}})};
233
234 handle_event(internal,
235 {connect = EventName, Arg},
236 connecting,
237 #{socket := Socket} = Data) ->
238 
:-(
 case socket:connect(Socket, Arg) of
239 ok ->
240 
:-(
 {next_state,
241 connected,
242 Data#{partial => <<>>},
243 [nei({telemetry, EventName, #{count => 1}, Arg}),
244 nei(recv)]};
245
246 {error, Reason} ->
247 
:-(
 {stop, Reason}
248 end.
249
250
251 addr() ->
252 
:-(
 ?FUNCTION_NAME(resp_config:client(hostname)).
253
254
255 addr(Hostname) ->
256 
:-(
 {ok, #hostent{h_addr_list = Addresses}} = inet:gethostbyname(Hostname),
257 
:-(
 pick_one(Addresses).
258
259
260 pick_one(L) ->
261 
:-(
 lists:nth(rand:uniform(length(L)), L).
Line Hits Source