emqx / esockd / 527

Committed 16 Sep 2025 07:26AM UTC coverage: 67.052% (+1.0%) from 66.039%

Build # 527

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #211 from JimMoen/fix-rate-limit-pause

fix: the next check start time should be `Now + Pasue`

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

228 existing lines in 13 files now uncovered.

696 of 1038 relevant lines covered (67.05%)

106.85 hits per line

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

65.63

/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]).

-export([ options/1
        , get_port/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(ACCEPTOR_POOL, 16).
-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())
      -> {ok, pid()} | ignore | {error, term()}).
start_link(Proto, ListenOn, Opts, AcceptorSup) ->
    gen_server:start_link(?MODULE, {Proto, ListenOn, Opts, AcceptorSup}, []).

-spec(options(pid()) -> [esockd:option()]).
options(Listener) ->
    gen_server:call(Listener, options).

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

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

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

emqx / esockd / 527

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