emqx / esockd / 354

Committed 14 Dec 2023 12:54PM UTC coverage: 71.491%. First build

Build # 354

Build Type

Pull #183

github

Committed by

web-flow

Commit Message

Merge 3297859f4 into 5cb22a8b1

Pull Request Pull Request #183: feat(listener): support changing options on the fly

Run Details

170 of 192 new or added lines in 10 files covered. (88.54%)

820 of 1147 relevant lines covered (71.49%)

60.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.46

/src/esockd_sup.erl

%%--------------------------------------------------------------------
%% Copyright (c) 2020 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------

-module(esockd_sup).

-behaviour(supervisor).

-export([start_link/0]).

-export([child_id/2]).

-export([ start_listener/4
        , stop_listener/2
        , restart_listener/2
        ]).

-export([ listeners/0
        , listener/1
        , listener_and_module/1
        ]).

-export([ child_spec/4
        , udp_child_spec/4
        , dtls_child_spec/4
        , start_child/1
        ]).

%% supervisor callback
-export([init/1]).

%%--------------------------------------------------------------------
%% API
%%--------------------------------------------------------------------

-spec(start_link() -> {ok, pid()} | ignore | {error, term()}).
start_link() ->
    supervisor:start_link({local, ?MODULE}, ?MODULE, []).

-spec(start_listener(atom(), esockd:listen_on(), [esockd:option()], esockd:mfargs())
      -> {ok, pid()} | {error, term()}).
start_listener(Proto, ListenOn, Opts, MFA) ->
    start_child(child_spec(Proto, ListenOn, Opts, MFA)).

-spec(child_spec(atom(), esockd:listen_on(), [esockd:option()], esockd:mfargs())
      -> supervisor:child_spec()).
child_spec(Proto, ListenOn, Opts, MFA) when is_atom(Proto) ->
    ListenerRef = {Proto, ListenOn},
    _ = esockd_server:set_listener_prop(ListenerRef, type, tcp),
    _ = esockd_server:set_listener_prop(ListenerRef, options, Opts),    
    #{id => child_id(Proto, ListenOn),
      start => {esockd_listener_sup, start_link, [Proto, ListenOn, MFA]},
      restart => transient,
      shutdown => infinity,
      type => supervisor,
      modules => [esockd_listener_sup]}.

-spec(udp_child_spec(atom(), esockd:listen_on(), [esockd:option()], esockd:mfargs())
      -> supervisor:child_spec()).
udp_child_spec(Proto, Port, Opts, MFA) when is_atom(Proto) ->
    #{id => child_id(Proto, Port),
      start => {esockd_udp, server, [Proto, Port, Opts, MFA]},
      restart => transient,
      shutdown => 5000,
      type => worker,
      modules => [esockd_udp]}.

-spec(dtls_child_spec(atom(), esockd:listen_on(), [esockd:option()], esockd:mfargs())
      -> supervisor:child_spec()).
dtls_child_spec(Proto, Port, Opts, MFA) when is_atom(Proto) ->
    ListenerRef = {Proto, Port},
    _ = esockd_server:set_listener_prop(ListenerRef, type, dtls),
    _ = esockd_server:set_listener_prop(ListenerRef, options, Opts),    
    #{id => child_id(Proto, Port),
      start => {esockd_listener_sup, start_link, [Proto, Port, MFA]},
      restart => transient,
      shutdown => infinity,
      type => supervisor,
      modules => [esockd_listener_sup]}.

-spec(start_child(supervisor:child_spec()) -> {ok, pid()} | {error, term()}).
start_child(ChildSpec) ->
        supervisor:start_child(?MODULE, ChildSpec).

-spec(stop_listener(atom(), esockd:listen_on()) -> ok | {error, term()}).
stop_listener(Proto, ListenOn) ->
    ListenerRef = {Proto, ListenOn},
    case match_listeners(Proto, ListenOn) of
        [] -> {error, not_found};
        Listeners ->
            Results = [terminate_and_delete(ChildId) || ChildId <- Listeners],
            case ok_or_error(Results) of
                ok ->
                    _ = esockd_server:erase_listener_props(ListenerRef),
                    ok;
                Error ->
                    Error
            end
    end.

terminate_and_delete(ChildId) ->
        case supervisor:terminate_child(?MODULE, ChildId) of
        ok    -> supervisor:delete_child(?MODULE, ChildId);
        Error -> Error
        end.

-spec(listeners() -> [{term(), pid()}]).
listeners() ->
    [{Id, Pid} || {{listener_sup, Id}, Pid, _Type, _} <- supervisor:which_children(?MODULE)].

-spec(listener({atom(), esockd:listen_on()}) -> pid()).
listener({Proto, ListenOn}) ->
    ChildId = child_id(Proto, ListenOn),
    case [Pid || {Id, Pid, _Type, _} <- supervisor:which_children(?MODULE), Id =:= ChildId] of
        [] -> error(not_found);
        L  -> hd(L)
    end.

-spec(listener_and_module({atom(), esockd:listen_on()})
     -> undefined
      | {ListenerSup :: pid(), Mod :: esockd_listener_sup | esockd_udp}).
listener_and_module({Proto, ListenOn}) ->
    ChildId = child_id(Proto, ListenOn),
    case [{Pid, Mod} || {Id, Pid, _Type, [Mod|_]} <- supervisor:which_children(?MODULE), Id =:= ChildId] of
        [] -> undefined;
        L  -> hd(L)
    end.

-spec(restart_listener(atom(), esockd:listen_on()) -> ok | {error, term()}).
restart_listener(Proto, ListenOn) ->
    case match_listeners(Proto, ListenOn) of
        [] -> {error, not_found};
        Listeners ->
            ok_or_error([terminate_and_restart(ChildId) || ChildId <- Listeners])
    end.

terminate_and_restart(ChildId) ->
    case supervisor:terminate_child(?MODULE, ChildId) of
        ok    -> supervisor:restart_child(?MODULE, ChildId);
        Error -> Error
    end.

match_listeners(Proto, ListenOn) ->
    [ChildId || {ChildId, _Pid, _Type, _} <- supervisor:which_children(?MODULE),
                match_listener(Proto, ListenOn, ChildId)].

match_listener(Proto, ListenOn, {listener_sup, {Proto, ListenOn}}) ->
    true;
match_listener(Proto, Port, {listener_sup, {Proto, {_IP, Port}}}) ->
    true;
match_listener(_Proto, _ListenOn, _ChildId) ->
    false.

child_id(Proto, ListenOn) ->
    {listener_sup, {Proto, ListenOn}}.

ok_or_error([]) -> ok;
ok_or_error([ok|Results]) ->
    ok_or_error(Results);
ok_or_error([{ok, _Pid}|Results]) ->
    ok_or_error(Results);
ok_or_error([{error, Reason}|_]) ->
    {error, Reason}.

%%--------------------------------------------------------------------
%% Supervisor callbacks
%%--------------------------------------------------------------------

init([]) ->
    SupFlags = #{strategy => one_for_one,
                 intensity => 10,
                 period => 100
                },
    Limiter = #{id => esockd_limiter,
                start => {esockd_limiter, start_link, []},
                restart => permanent,
                shutdown => 5000,
                type => worker,
                modules => [esockd_limiter]
               },
    Server = #{id => esockd_server,
               start => {esockd_server, start_link, []},
               restart => permanent,
               shutdown => 5000,
               type => worker,
               modules => [esockd_server]
              },
    {ok, {SupFlags, [Limiter, Server]}}.


1	%%--------------------------------------------------------------------
2	%% Copyright (c) 2020 EMQ Technologies Co., Ltd. All Rights Reserved.
3	%%
4	%% Licensed under the Apache License, Version 2.0 (the "License");
5	%% you may not use this file except in compliance with the License.
6	%% You may obtain a copy of the License at
7	%%
8	%% http://www.apache.org/licenses/LICENSE-2.0
9	%%
10	%% Unless required by applicable law or agreed to in writing, software
11	%% distributed under the License is distributed on an "AS IS" BASIS,
12	%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	%% See the License for the specific language governing permissions and
14	%% limitations under the License.
15	%%--------------------------------------------------------------------
16
17	-module(esockd_sup).
18
19	-behaviour(supervisor).
20
21	-export([start_link/0]).
22
23	-export([child_id/2]).
24
25	-export([ start_listener/4
26	, stop_listener/2
27	, restart_listener/2
28	]).
29
30	-export([ listeners/0
31	, listener/1
32	, listener_and_module/1
33	]).
34
35	-export([ child_spec/4
36	, udp_child_spec/4
37	, dtls_child_spec/4
38	, start_child/1
39	]).
40
41	%% supervisor callback
42	-export([init/1]).
43
44	%%--------------------------------------------------------------------
45	%% API
46	%%--------------------------------------------------------------------
47
48	-spec(start_link() -> {ok, pid()} \| ignore \| {error, term()}).
49	start_link() ->
50	supervisor:start_link({local, ?MODULE}, ?MODULE, []).	4✔
51
52	-spec(start_listener(atom(), esockd:listen_on(), [esockd:option()], esockd:mfargs())
53	-> {ok, pid()} \| {error, term()}).
54	start_listener(Proto, ListenOn, Opts, MFA) ->
55	start_child(child_spec(Proto, ListenOn, Opts, MFA)).	44✔
56
57	-spec(child_spec(atom(), esockd:listen_on(), [esockd:option()], esockd:mfargs())
58	-> supervisor:child_spec()).
59	child_spec(Proto, ListenOn, Opts, MFA) when is_atom(Proto) ->
60	ListenerRef = {Proto, ListenOn},	45✔
61	_ = esockd_server:set_listener_prop(ListenerRef, type, tcp),	45✔
62	_ = esockd_server:set_listener_prop(ListenerRef, options, Opts),	45✔
63	#{id => child_id(Proto, ListenOn),	45✔
64	start => {esockd_listener_sup, start_link, [Proto, ListenOn, MFA]},
65	restart => transient,
66	shutdown => infinity,
67	type => supervisor,
68	modules => [esockd_listener_sup]}.
69
70	-spec(udp_child_spec(atom(), esockd:listen_on(), [esockd:option()], esockd:mfargs())
71	-> supervisor:child_spec()).
72	udp_child_spec(Proto, Port, Opts, MFA) when is_atom(Proto) ->
73	#{id => child_id(Proto, Port),	8✔
74	start => {esockd_udp, server, [Proto, Port, Opts, MFA]},
75	restart => transient,
76	shutdown => 5000,
77	type => worker,
78	modules => [esockd_udp]}.
79
80	-spec(dtls_child_spec(atom(), esockd:listen_on(), [esockd:option()], esockd:mfargs())
81	-> supervisor:child_spec()).
82	dtls_child_spec(Proto, Port, Opts, MFA) when is_atom(Proto) ->
83	ListenerRef = {Proto, Port},	11✔
84	_ = esockd_server:set_listener_prop(ListenerRef, type, dtls),	11✔
85	_ = esockd_server:set_listener_prop(ListenerRef, options, Opts),	11✔
86	#{id => child_id(Proto, Port),	11✔
87	start => {esockd_listener_sup, start_link, [Proto, Port, MFA]},
88	restart => transient,
89	shutdown => infinity,
90	type => supervisor,
91	modules => [esockd_listener_sup]}.
92
93	-spec(start_child(supervisor:child_spec()) -> {ok, pid()} \| {error, term()}).
94	start_child(ChildSpec) ->
95	supervisor:start_child(?MODULE, ChildSpec).	61✔
96
97	-spec(stop_listener(atom(), esockd:listen_on()) -> ok \| {error, term()}).
98	stop_listener(Proto, ListenOn) ->
99	ListenerRef = {Proto, ListenOn},	61✔
100	case match_listeners(Proto, ListenOn) of	61✔
101	[] -> {error, not_found};	×
102	Listeners ->
103	Results = [terminate_and_delete(ChildId) \|\| ChildId <- Listeners],	61✔
104	case ok_or_error(Results) of	61✔
105	ok ->
106	_ = esockd_server:erase_listener_props(ListenerRef),	61✔
107	ok;	61✔
108	Error ->
NEW 109	Error	×
110	end
111	end.
112
113	terminate_and_delete(ChildId) ->
114	case supervisor:terminate_child(?MODULE, ChildId) of	61✔
115	ok -> supervisor:delete_child(?MODULE, ChildId);	61✔
116	Error -> Error	×
117	end.
118
119	-spec(listeners() -> [{term(), pid()}]).
120	listeners() ->
121	[{Id, Pid} \|\| {{listener_sup, Id}, Pid, _Type, _} <- supervisor:which_children(?MODULE)].	2✔
122
123	-spec(listener({atom(), esockd:listen_on()}) -> pid()).
124	listener({Proto, ListenOn}) ->
125	ChildId = child_id(Proto, ListenOn),	2✔
126	case [Pid \|\| {Id, Pid, _Type, _} <- supervisor:which_children(?MODULE), Id =:= ChildId] of	2✔
127	[] -> error(not_found);	1✔
128	L -> hd(L)	1✔
129	end.
130
131	-spec(listener_and_module({atom(), esockd:listen_on()})
132	-> undefined
133	\| {ListenerSup :: pid(), Mod :: esockd_listener_sup \| esockd_udp}).
134	listener_and_module({Proto, ListenOn}) ->
135	ChildId = child_id(Proto, ListenOn),	64✔
136	case [{Pid, Mod} \|\| {Id, Pid, _Type, [Mod\|_]} <- supervisor:which_children(?MODULE), Id =:= ChildId] of	64✔
137	[] -> undefined;	6✔
138	L -> hd(L)	58✔
139	end.
140
141	-spec(restart_listener(atom(), esockd:listen_on()) -> ok \| {error, term()}).
142	restart_listener(Proto, ListenOn) ->
143	case match_listeners(Proto, ListenOn) of	3✔
144	[] -> {error, not_found};	1✔
145	Listeners ->
146	ok_or_error([terminate_and_restart(ChildId) \|\| ChildId <- Listeners])	2✔
147	end.
148
149	terminate_and_restart(ChildId) ->
150	case supervisor:terminate_child(?MODULE, ChildId) of	2✔
151	ok -> supervisor:restart_child(?MODULE, ChildId);	2✔
152	Error -> Error	×
153	end.
154
155	match_listeners(Proto, ListenOn) ->
156	[ChildId \|\| {ChildId, _Pid, _Type, _} <- supervisor:which_children(?MODULE),	64✔
157	match_listener(Proto, ListenOn, ChildId)].	192✔
158
159	match_listener(Proto, ListenOn, {listener_sup, {Proto, ListenOn}}) ->
160	true;	63✔
161	match_listener(Proto, Port, {listener_sup, {Proto, {_IP, Port}}}) ->
162	true;	×
163	match_listener(_Proto, _ListenOn, _ChildId) ->
164	false.	129✔
165
166	child_id(Proto, ListenOn) ->
167	{listener_sup, {Proto, ListenOn}}.	130✔
168
169	ok_or_error([]) -> ok;	63✔
170	ok_or_error([ok\|Results]) ->
171	ok_or_error(Results);	61✔
172	ok_or_error([{ok, _Pid}\|Results]) ->
173	ok_or_error(Results);	2✔
174	ok_or_error([{error, Reason}\|_]) ->
175	{error, Reason}.	×
176
177	%%--------------------------------------------------------------------
178	%% Supervisor callbacks
179	%%--------------------------------------------------------------------
180
181	init([]) ->
182	SupFlags = #{strategy => one_for_one,	4✔
183	intensity => 10,
184	period => 100
185	},
186	Limiter = #{id => esockd_limiter,	4✔
187	start => {esockd_limiter, start_link, []},
188	restart => permanent,
189	shutdown => 5000,
190	type => worker,
191	modules => [esockd_limiter]
192	},
193	Server = #{id => esockd_server,	4✔
194	start => {esockd_server, start_link, []},
195	restart => permanent,
196	shutdown => 5000,
197	type => worker,
198	modules => [esockd_server]
199	},
200	{ok, {SupFlags, [Limiter, Server]}}.	4✔
201

emqx / esockd / 354

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous