emqx / esockd / 356

Committed 15 Dec 2023 02:41PM UTC coverage: 72.414%. First build

Build # 356

Build Type

Pull #183

github

Committed by

web-flow

Commit Message

Merge 8bfe0c31c into 5cb22a8b1

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

Run Details

191 of 213 new or added lines in 10 files covered. (89.67%)

840 of 1160 relevant lines covered (72.41%)

62.94 hits per line

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

67.39

/src/esockd_listener.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_listener).

-behaviour(gen_server).

-include("esockd.hrl").

-export([ start_link/3
        , start_supervised/1
        ]).

-export([ get_port/1
        , get_lsock/1
        , get_state/1
        , set_options/2
        ]).

%% gen_server callbacks
-export([ init/1
        , handle_call/3
        , handle_cast/2
        , handle_info/2
        , terminate/2
        , code_change/3
        ]).

-record(state, {
          proto     :: atom(),
          listen_on :: esockd:listen_on(),
          options   :: [esockd:option()],
          lsock     :: inet:socket(),
          laddr     :: inet:ip_address(),
          lport     :: inet:port_number()
         }).

-type option() :: {tcp_options, [gen_tcp:option()]}.

-define(DEFAULT_TCP_OPTIONS,
        [{nodelay, true},
         {reuseaddr, true},
         {send_timeout, 30000},
         {send_timeout_close, true}
        ]).

-spec start_link(atom(), esockd:listen_on(), [esockd:option()])
      -> {ok, pid()} | ignore | {error, term()}.
start_link(Proto, ListenOn, Opts) ->
    gen_server:start_link(?MODULE, {Proto, ListenOn, Opts}, []).

-spec start_supervised(esockd:listener_ref())
      -> {ok, pid()} | ignore | {error, term()}.
start_supervised(ListenerRef = {Proto, ListenOn}) ->
    Opts = esockd_server:get_listener_prop(ListenerRef, options),
    case start_link(Proto, ListenOn, Opts) of
        {ok, Pid} ->
            _ = esockd_server:set_listener_prop(ListenerRef, listener, {?MODULE, Pid}),
            {ok, Pid};
        Error ->
            Error
    end.

-spec get_port(pid()) -> inet:port_number().
get_port(Listener) ->
    gen_server:call(Listener, get_port).

-spec get_lsock(pid())  -> inet:socket().
get_lsock(Listener) ->
    gen_server:call(Listener, get_lsock).

-spec get_state(pid())  -> proplists:proplist().
get_state(Listener) ->
    gen_server:call(Listener, get_state).

-spec set_options(pid(), [option()])  -> ok.
set_options(Listener, Opts) ->
    gen_server:call(Listener, {set_options, Opts}).

%%--------------------------------------------------------------------
%% gen_server callbacks
%%--------------------------------------------------------------------

init({Proto, ListenOn, Opts}) ->
    Port = port(ListenOn),
    process_flag(trap_exit, true),
    esockd_server:ensure_stats({Proto, ListenOn}),
    SockOpts = merge_addr(ListenOn, sockopts(Opts)),
    case esockd_transport:listen(Port, esockd:merge_opts(?DEFAULT_TCP_OPTIONS, SockOpts)) of
        {ok, LSock} ->
            {ok, {LAddr, LPort}} = inet:sockname(LSock),
            %%error_logger:info_msg("~s listen on ~s:~p with ~p acceptors.~n",
            %%                      [Proto, inet:ntoa(LAddr), LPort, AcceptorNum]),
            {ok, #state{proto = Proto, listen_on = ListenOn, options = Opts,
                        lsock = LSock, laddr = LAddr, lport = LPort}};
        {error, Reason} ->
            error_logger:error_msg("~s failed to listen on ~p - ~p (~s)",
                                   [Proto, Port, Reason, inet:format_error(Reason)]),
            {stop, Reason}
    end.

sockopts(Opts) ->
    %% Don't active the socket...
    SockOpts = proplists:get_value(tcp_options, Opts, []),
    [{active, false} | proplists:delete(active, SockOpts)].

port(Port) when is_integer(Port) -> Port;
port({_Addr, Port}) -> Port.

merge_addr(Port, SockOpts) when is_integer(Port) ->
    SockOpts;
merge_addr({Addr, _Port}, SockOpts) ->
    lists:keystore(ip, 1, SockOpts, {ip, Addr}).

handle_call(options, _From, State = #state{options = Opts}) ->
    {reply, Opts, State};

handle_call(get_port, _From, State = #state{lport = LPort}) ->
    {reply, LPort, State};

handle_call(get_lsock, _From, State = #state{lsock = LSock}) ->
    {reply, LSock, State};

handle_call(get_state, _From, State = #state{lsock = LSock, lport = LPort}) ->
    Reply = [ {listen_sock, LSock}
            , {listen_port, LPort}
            ],
    {reply, Reply, State};

handle_call({set_options, Opts}, _From, State = #state{lsock = LSock}) ->
    case inet:setopts(LSock, sockopts(Opts)) of
        ok ->
            {reply, ok, State#state{options = Opts}};
        Error = {error, _} ->
            {reply, Error, State}
    end;

handle_call(Req, _From, State) ->
    error_logger:error_msg("[~s] Unexpected call: ~p", [?MODULE, Req]),
    {noreply, State}.

handle_cast(Msg, State) ->
    error_logger:error_msg("[~s] Unexpected cast: ~p", [?MODULE, Msg]),
    {noreply, State}.

handle_info({'EXIT', LSock, _}, #state{lsock = LSock} = State) ->
    error_logger:error_msg("~s Lsocket ~p closed", [?MODULE, LSock]),
    {stop, lsock_closed, State};
handle_info(Info, State) ->
    error_logger:error_msg("[~s] Unexpected info: ~p", [?MODULE, Info]),
    {noreply, State}.

terminate(_Reason, #state{proto = Proto, listen_on = ListenOn,
                          lsock = LSock, laddr = Addr, lport = Port}) ->
    error_logger:info_msg("~s stopped on ~s~n", [Proto, esockd:format({Addr, Port})]),
    esockd_limiter:delete({listener, Proto, ListenOn}),
    esockd_server:del_stats({Proto, ListenOn}),
    esockd_transport:fast_close(LSock).

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.


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_listener).
18
19	-behaviour(gen_server).
20
21	-include("esockd.hrl").
22
23	-export([ start_link/3
24	, start_supervised/1
25	]).
26
27	-export([ get_port/1
28	, get_lsock/1
29	, get_state/1
30	, set_options/2
31	]).
32
33	%% gen_server callbacks
34	-export([ init/1
35	, handle_call/3
36	, handle_cast/2
37	, handle_info/2
38	, terminate/2
39	, code_change/3
40	]).
41
42	-record(state, {
43	proto :: atom(),
44	listen_on :: esockd:listen_on(),
45	options :: [esockd:option()],
46	lsock :: inet:socket(),
47	laddr :: inet:ip_address(),
48	lport :: inet:port_number()
49	}).
50
51	-type option() :: {tcp_options, [gen_tcp:option()]}.
52
53	-define(DEFAULT_TCP_OPTIONS,
54	[{nodelay, true},
55	{reuseaddr, true},
56	{send_timeout, 30000},
57	{send_timeout_close, true}
58	]).
59
60	-spec start_link(atom(), esockd:listen_on(), [esockd:option()])
61	-> {ok, pid()} \| ignore \| {error, term()}.
62	start_link(Proto, ListenOn, Opts) ->
63	gen_server:start_link(?MODULE, {Proto, ListenOn, Opts}, []).	48✔
64
65	-spec start_supervised(esockd:listener_ref())
66	-> {ok, pid()} \| ignore \| {error, term()}.
67	start_supervised(ListenerRef = {Proto, ListenOn}) ->
68	Opts = esockd_server:get_listener_prop(ListenerRef, options),	48✔
69	case start_link(Proto, ListenOn, Opts) of	48✔
70	{ok, Pid} ->
71	_ = esockd_server:set_listener_prop(ListenerRef, listener, {?MODULE, Pid}),	48✔
72	{ok, Pid};	48✔
73	Error ->
NEW 74	Error	×
75	end.
76
77	-spec get_port(pid()) -> inet:port_number().
78	get_port(Listener) ->
79	gen_server:call(Listener, get_port).	×
80
81	-spec get_lsock(pid()) -> inet:socket().
82	get_lsock(Listener) ->
83	gen_server:call(Listener, get_lsock).	×
84
85	-spec get_state(pid()) -> proplists:proplist().
86	get_state(Listener) ->
87	gen_server:call(Listener, get_state).	59✔
88
89	-spec set_options(pid(), [option()]) -> ok.
90	set_options(Listener, Opts) ->
91	gen_server:call(Listener, {set_options, Opts}).	8✔
92
93	%%--------------------------------------------------------------------
94	%% gen_server callbacks
95	%%--------------------------------------------------------------------
96
97	init({Proto, ListenOn, Opts}) ->
98	Port = port(ListenOn),	48✔
99	process_flag(trap_exit, true),	48✔
100	esockd_server:ensure_stats({Proto, ListenOn}),	48✔
101	SockOpts = merge_addr(ListenOn, sockopts(Opts)),	48✔
102	case esockd_transport:listen(Port, esockd:merge_opts(?DEFAULT_TCP_OPTIONS, SockOpts)) of	48✔
103	{ok, LSock} ->
104	{ok, {LAddr, LPort}} = inet:sockname(LSock),	48✔
105	%%error_logger:info_msg("~s listen on ~s:~p with ~p acceptors.~n",
106	%% [Proto, inet:ntoa(LAddr), LPort, AcceptorNum]),
107	{ok, #state{proto = Proto, listen_on = ListenOn, options = Opts,	48✔
108	lsock = LSock, laddr = LAddr, lport = LPort}};
109	{error, Reason} ->
110	error_logger:error_msg("~s failed to listen on ~p - ~p (~s)",	×
111	[Proto, Port, Reason, inet:format_error(Reason)]),
112	{stop, Reason}	×
113	end.
114
115	sockopts(Opts) ->
116	%% Don't active the socket...
117	SockOpts = proplists:get_value(tcp_options, Opts, []),	56✔
118	[{active, false} \| proplists:delete(active, SockOpts)].	56✔
119
120	port(Port) when is_integer(Port) -> Port;	45✔
121	port({_Addr, Port}) -> Port.	3✔
122
123	merge_addr(Port, SockOpts) when is_integer(Port) ->
124	SockOpts;	45✔
125	merge_addr({Addr, _Port}, SockOpts) ->
126	lists:keystore(ip, 1, SockOpts, {ip, Addr}).	3✔
127
128	handle_call(options, _From, State = #state{options = Opts}) ->
129	{reply, Opts, State};	×
130
131	handle_call(get_port, _From, State = #state{lport = LPort}) ->
132	{reply, LPort, State};	×
133
134	handle_call(get_lsock, _From, State = #state{lsock = LSock}) ->
135	{reply, LSock, State};	×
136
137	handle_call(get_state, _From, State = #state{lsock = LSock, lport = LPort}) ->
138	Reply = [ {listen_sock, LSock}	59✔
139	, {listen_port, LPort}
140	],
141	{reply, Reply, State};	59✔
142
143	handle_call({set_options, Opts}, _From, State = #state{lsock = LSock}) ->
144	case inet:setopts(LSock, sockopts(Opts)) of	8✔
145	ok ->
146	{reply, ok, State#state{options = Opts}};	7✔
147	Error = {error, _} ->
148	{reply, Error, State}	1✔
149	end;
150
151	handle_call(Req, _From, State) ->
152	error_logger:error_msg("[~s] Unexpected call: ~p", [?MODULE, Req]),	×
153	{noreply, State}.	×
154
155	handle_cast(Msg, State) ->
156	error_logger:error_msg("[~s] Unexpected cast: ~p", [?MODULE, Msg]),	×
157	{noreply, State}.	×
158
159	handle_info({'EXIT', LSock, _}, #state{lsock = LSock} = State) ->
160	error_logger:error_msg("~s Lsocket ~p closed", [?MODULE, LSock]),	2✔
161	{stop, lsock_closed, State};	2✔
162	handle_info(Info, State) ->
163	error_logger:error_msg("[~s] Unexpected info: ~p", [?MODULE, Info]),	×
164	{noreply, State}.	×
165
166	terminate(_Reason, #state{proto = Proto, listen_on = ListenOn,
167	lsock = LSock, laddr = Addr, lport = Port}) ->
168	error_logger:info_msg("~s stopped on ~s~n", [Proto, esockd:format({Addr, Port})]),	48✔
169	esockd_limiter:delete({listener, Proto, ListenOn}),	48✔
170	esockd_server:del_stats({Proto, ListenOn}),	48✔
171	esockd_transport:fast_close(LSock).	48✔
172
173	code_change(_OldVsn, State, _Extra) ->
174	{ok, State}.	×
175

emqx / esockd / 356

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