emqx / esockd / 352

Committed 12 Dec 2023 09:17PM UTC coverage: 70.175%. First build

Build # 352

Build Type

Pull #183

github

Committed by

web-flow

Commit Message

Merge c60472aab into 5cb22a8b1

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

Run Details

143 of 177 new or added lines in 10 files covered. (80.79%)

800 of 1140 relevant lines covered (70.18%)

59.41 hits per line

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

63.41

/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/4
        , start_supervised/1
        ]).

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

%% 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()
         }).

-define(DEFAULT_TCP_OPTIONS,
        [{nodelay, true},
         {reuseaddr, true},
         {send_timeout, 30000},
         {send_timeout_close, true}
        ]).
-spec start_link(atom(), esockd:listen_on(), [esockd:option()], pid() | ignore)
      -> {ok, pid()} | ignore | {error, term()}.
start_link(Proto, ListenOn, Opts, AcceptorSup) ->
    gen_server:start_link(?MODULE, {Proto, ListenOn, Opts, AcceptorSup}, []).

-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, ignore) 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).

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

init({Proto, ListenOn, Opts, _Ignore}) ->
    Port = port(ListenOn),
    process_flag(trap_exit, true),
    esockd_server:ensure_stats({Proto, ListenOn}),
    SockOpts = merge_addr(ListenOn, sockopts(Opts)),
    %% Don't active the socket...
    case esockd_transport:listen(Port, [{active, false} | proplists:delete(active, 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) ->
    esockd:merge_opts(?DEFAULT_TCP_OPTIONS,
                      proplists:get_value(tcp_options, Opts, [])).

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

emqx / esockd / 352

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