|
| 1 | +%%-------------------------------------------------------------------- |
| 2 | +%% Copyright (c) 2024 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_acceptor_SUITE). |
| 18 | + |
| 19 | +-compile(export_all). |
| 20 | +-compile(nowarn_export_all). |
| 21 | + |
| 22 | +-include("esockd.hrl"). |
| 23 | +-include_lib("eunit/include/eunit.hrl"). |
| 24 | +-include_lib("common_test/include/ct.hrl"). |
| 25 | + |
| 26 | +-define(PORT, 30000 + ?LINE). |
| 27 | +-define(COUNTER_ACCPETED, 1). |
| 28 | +-define(COUNTER_NOSTART, 2). |
| 29 | +-define(COUNTER_OVERLOADED, 3). |
| 30 | +-define(COUNTER_RATE_LIMITTED, 4). |
| 31 | +-define(COUNTER_OTHER_REASONSE, 5). |
| 32 | +-define(COUNTER_LAST, 10). |
| 33 | + |
| 34 | +counter_tag_to_index(accepted) -> ?COUNTER_ACCPETED; |
| 35 | +counter_tag_to_index(closed_nostart) -> ?COUNTER_NOSTART; |
| 36 | +counter_tag_to_index(closed_overloaded) -> ?COUNTER_OVERLOADED; |
| 37 | +counter_tag_to_index(closed_rate_limitted) -> ?COUNTER_RATE_LIMITTED; |
| 38 | +counter_tag_to_index(closed_other_reasons) -> ?COUNTER_OTHER_REASONSE. |
| 39 | + |
| 40 | +all() -> esockd_ct:all(?MODULE). |
| 41 | + |
| 42 | +init_per_suite(Config) -> |
| 43 | + Config. |
| 44 | + |
| 45 | +end_per_suite(_Config) -> |
| 46 | + ok. |
| 47 | + |
| 48 | +init_per_testcase(_Case, Config) -> |
| 49 | + Counters = counters:new(?COUNTER_LAST, []), |
| 50 | + meck:new(esockd_server, [passthrough, no_history]), |
| 51 | + meck:expect(esockd_server, inc_stats, |
| 52 | + fun(_, Tag, Count) -> |
| 53 | + Index = counter_tag_to_index(Tag), |
| 54 | + counters:add(Counters, Index, Count) |
| 55 | + end), |
| 56 | + [{counters, Counters} | Config]. |
| 57 | + |
| 58 | +end_per_testcase(_Case, _Config) -> |
| 59 | + meck:unload(esockd_server). |
| 60 | + |
| 61 | +start(PortNumber, Limiter) -> |
| 62 | + start(PortNumber, Limiter, #{}). |
| 63 | + |
| 64 | +start(PortNumber, Limiter, Opts) -> |
| 65 | + SockOpts = [binary, |
| 66 | + {active, false}, |
| 67 | + {reuseaddr, true}, |
| 68 | + {nodelay, true}, |
| 69 | + {backlog, maps:get(backlog, Opts, 1024)}], |
| 70 | + {ok, ListenSocket} = gen_tcp:listen(PortNumber, SockOpts), |
| 71 | + TuneFun = esockd_acceptor_sup:tune_socket_fun([]), |
| 72 | + StartConn = {fun ?MODULE:start_connection/3, [Opts]}, |
| 73 | + {ok, AccPid} = esockd_acceptor:start_link(tcp, PortNumber, StartConn, TuneFun, _UpFuns = [], Limiter, ListenSocket), |
| 74 | + #{lsock => ListenSocket, acceptor => AccPid}. |
| 75 | + |
| 76 | +stop(#{lsock := ListenSocket, acceptor := AccPid}) -> |
| 77 | + ok = gen_statem:stop(AccPid), |
| 78 | + gen_tcp:close(ListenSocket), |
| 79 | + ok. |
| 80 | + |
| 81 | +connect(Port) -> |
| 82 | + connect(Port, 1000, #{}). |
| 83 | + |
| 84 | +connect(Port, Timeout, Opts0) -> |
| 85 | + Opts = [binary, |
| 86 | + {active, maps:get(active, Opts0, false)}, |
| 87 | + {nodelay, true}], |
| 88 | + gen_tcp:connect("localhost", Port, Opts, Timeout). |
| 89 | + |
| 90 | +%% This is the very basic test, if this fails, nothing elese matters. |
| 91 | +t_normal(Config) -> |
| 92 | + Port = ?PORT, |
| 93 | + Server = start(Port, no_limit()), |
| 94 | + {ok, ClientSock} = connect(Port), |
| 95 | + try |
| 96 | + ok = wait_for_counter(Config, ?COUNTER_ACCPETED, 1, 2000) |
| 97 | + after |
| 98 | + disconnect(ClientSock), |
| 99 | + stop(Server) |
| 100 | + end. |
| 101 | + |
| 102 | +t_rate_limitted(Config) -> |
| 103 | + Port = ?PORT, |
| 104 | + Pause = 200, |
| 105 | + Server = start(Port, pause_then_allow(Pause)), |
| 106 | + try |
| 107 | + Count = 10, |
| 108 | + Socks = lists:map(fun(_) -> |
| 109 | + {ok, Sock} = connect(Port, 1000, #{active => true}), |
| 110 | + Sock |
| 111 | + end, lists:seq(1, Count)), |
| 112 | + lists:foreach(fun(Sock) -> |
| 113 | + receive |
| 114 | + {tcp_closed, Sock} -> |
| 115 | + ok; |
| 116 | + Other -> |
| 117 | + ct:fail({unexpected, Other}) |
| 118 | + end |
| 119 | + end, Socks), |
| 120 | + ok = wait_for_counter(Config, ?COUNTER_RATE_LIMITTED, Count, 2000), |
| 121 | + timer:sleep(Pause), |
| 122 | + {ok, Sock2} = connect(Port), |
| 123 | + ok = wait_for_counter(Config, ?COUNTER_ACCPETED, 1, 2000), |
| 124 | + disconnect(Sock2) |
| 125 | + after |
| 126 | + stop(Server) |
| 127 | + end. |
| 128 | + |
| 129 | +t_overloaded(Config) -> |
| 130 | + Port = ?PORT, |
| 131 | + Server = start(Port, no_limit(), #{start_connection_result => {error, overloaded}}), |
| 132 | + {ok, Sock1} = connect(Port), |
| 133 | + try |
| 134 | + ok = wait_for_counter(Config, ?COUNTER_OVERLOADED, 1, 2000), |
| 135 | + disconnect(Sock1) |
| 136 | + after |
| 137 | + stop(Server) |
| 138 | + end. |
| 139 | + |
| 140 | +disconnect(Socket) -> |
| 141 | + port_close(Socket), |
| 142 | + ok. |
| 143 | + |
| 144 | +%% no connection can get through |
| 145 | +pause_then_allow(Pause) -> |
| 146 | + #{module => ?MODULE, |
| 147 | + name => pause_then_allow, |
| 148 | + current => pause, |
| 149 | + next => allow, |
| 150 | + pause => Pause |
| 151 | + }. |
| 152 | + |
| 153 | +%% make a no-limit limiter |
| 154 | +no_limit() -> |
| 155 | + #{module => ?MODULE, name => no_limit}. |
| 156 | + |
| 157 | +%% limiter callback |
| 158 | +consume(_Token, #{name := pause_then_allow} = Limiter) -> |
| 159 | + case Limiter of |
| 160 | + #{current := pause} -> |
| 161 | + {pause, maps:get(pause, Limiter), Limiter#{current => allow}}; |
| 162 | + #{current := allow} -> |
| 163 | + {ok, Limiter} |
| 164 | + end; |
| 165 | +consume(_Token, #{name := no_limit} = Limiter) -> |
| 166 | + {ok, Limiter}. |
| 167 | + |
| 168 | +%% inspect during tests |
| 169 | +get_pd_counter(Tag) -> |
| 170 | + get({counter, Tag}). |
| 171 | + |
| 172 | +now_ts() -> erlang:system_time(millisecond). |
| 173 | + |
| 174 | +wait_for_counter(Config, Index, Count, Timeout) -> |
| 175 | + Counters = proplists:get_value(counters, Config), |
| 176 | + Now = now_ts(), |
| 177 | + Deadline = Now + Timeout, |
| 178 | + do_wait_for_counter(Counters, Index, Count, Deadline). |
| 179 | + |
| 180 | +do_wait_for_counter(Counters, Index, Count, Deadline) -> |
| 181 | + case counters:get(Counters, Index) of |
| 182 | + Count -> |
| 183 | + ok; |
| 184 | + Other when Other > Count -> |
| 185 | + error(#{cause => counter_exceeded_expect, |
| 186 | + expected => Count, |
| 187 | + counter_index => Index, |
| 188 | + got => Other}); |
| 189 | + Other -> |
| 190 | + case now_ts() > Deadline of |
| 191 | + true -> |
| 192 | + error(#{cause => timeout, |
| 193 | + expected => Count, |
| 194 | + counter_index=> Index, |
| 195 | + got => Other}); |
| 196 | + false -> |
| 197 | + timer:sleep(100), |
| 198 | + do_wait_for_counter(Counters, Index, Count, Deadline) |
| 199 | + end |
| 200 | + end. |
| 201 | + |
| 202 | +%% dummy callback to start connection |
| 203 | +start_connection(Opts, _Sock, _UpgradeFuns) -> |
| 204 | + case maps:get(start_connection_result, Opts, undefined) of |
| 205 | + undefined -> |
| 206 | + {ok, pid}; |
| 207 | + Other -> |
| 208 | + Other |
| 209 | + end. |
0 commit comments