SisMaker
/
eWSrv
1
0
Forkuj 0
Kod Zgłoszenia 0 Oczekujące zmiany 0 Projects 0 Wydania 0 Wiki Aktywność
Przeglądaj źródła

ft: 代码修改

master
SisMaker 3 lat temu
rodzic
c2b400d7df
commit
1187d5cf9a
24 zmienionych plików z 9 dodań i 1639 usunięć
Widok podzielony
Opcje porównania
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania
  1. +0
    -47
      include/eNet.hrl
  2. +1
    -3
      include/eWSrv.hrl
  3. +0
    -5
      include/ntCom.hrl
  4. +0
    -37
      include/ntProxyPt.hrl
  5. +1
    -0
      rebar.config
  6. +4
    -14
      src/eWSrv.erl
  7. +0
    -168
      src/wsNet/proxyPt/wsPptAcceptor.erl
  8. +0
    -29
      src/wsNet/proxyPt/wsPptAcceptorSup.erl
  9. +0
    -143
      src/wsNet/proxyPt/wsPptListener.erl
  10. +0
    -53
      src/wsNet/proxyPt/wsPptMgrSup.erl
  11. +0
    -197
      src/wsNet/proxyPt/ws_proxy_protocol.erl
  12. +0
    -145
      src/wsNet/ssl/wsSslAcceptor.erl
  13. +0
    -29
      src/wsNet/ssl/wsSslAcceptorSup.erl
  14. +0
    -143
      src/wsNet/ssl/wsSslListener.erl
  15. +0
    -51
      src/wsNet/ssl/wsSslMgrSup.erl
  16. +0
    -133
      src/wsNet/tcp/wsTcpAcceptor.erl
  17. +0
    -29
      src/wsNet/tcp/wsTcpAcceptorSup.erl
  18. +0
    -143
      src/wsNet/tcp/wsTcpListener.erl
  19. +0
    -48
      src/wsNet/tcp/wsTcpMgrSup.erl
  20. +0
    -118
      src/wsNet/wsCom.erl
  21. +0
    -69
      src/wsNet/wsNet.erl
  22. +0
    -28
      src/wsSrv/wsConSup.erl
  23. +3
    -3
      src/wsSrv/wsHttp.erl
  24. +0
    -4
      src/wsSrv/wsUtil.erl

+ 0
- 47
include/eNet.hrl Wyświetl plik

@ -1,47 +0,0 @@
-ifndef(eNet_H).
-define(eNet_H, true).
%% gen_tcp ready maybe to set sock options
%% ssl ready and then need do ntSslAcceptor:handshake/3 and maybe to set other options
%% ppt ready and then need do ntPptAcceptor:pptAndHS/5 and maybe to set other options
-define(mSockReady, mSockReady).
-define(DefTpOpts, [
binary
, {packet, 4}
, {active, false}
, {reuseaddr, true}
, {nodelay, false}
, {delay_send, true}
, {send_timeout, 15000}
, {keepalive, true}
, {exit_on_close, true}
, {back_log, 1024}
]).
-define(AptCnt, 16).
-define(DefSslHSTet, 15000).
-define(DefProxyPtTet, 5000).
-export_type([listenOpt/0]).
-type listenOpt() ::
{aptCnt, non_neg_integer()} |
{conMod, atom()} |
{conArgs, atom()} |
{tcpOpts, [gen_tcp:listen_option()]} |
{sslOpts, [ssl:ssl_option()]} |
{sslHSTet, timeout()} |
{udpOpts, [gen_udp:option()]} |
{proxyPt, boolean()} |
{proxyPtTet, timeout()}.
%% 令牌桶相关定义
-record(tBucket, {
rate :: pos_integer() %% 速率
, tokens :: non_neg_integer() %% 剩余tokens数量
, lastTime :: pos_integer() %% 最后一次更新访问时间单位毫秒
, bucketSize :: pos_integer() %% 桶大小 可以容纳的令牌数量
}).
-endif.

+ 1
- 3
include/eWSrv.hrl Wyświetl plik

@ -1,6 +1,4 @@
-include("eNet.hrl").
-include_lib("eNet/include/eNet.hrl").
-export_type([wsOpt/0]).
-type wsOpt() ::

+ 0
- 5
include/ntCom.hrl Wyświetl plik

@ -1,5 +0,0 @@
-define(ntErr(Format, Args), error_logger:error_msg(Format, Args)).
-define(ntWarn(Format, Args), error_logger:warning_msg(Format, Args)).
-define(ntInfo(Format, Args), error_logger:info_msg(Format, Args)).
-define(getLValue(Key, List, Default), wsCom:getListValue(Key, List, Default)).

+ 0
- 37
include/ntProxyPt.hrl Wyświetl plik

@ -1,37 +0,0 @@
-ifndef(ntProxyPt_H).
-define(ntProxyPt_H, true).
%%--------------------------------------------------------------------
%% Proxy-Protocol Socket Wrapper
%%--------------------------------------------------------------------
-type(pp2_additional_ssl_field() :: {pp2_ssl_client, boolean()}
| {pp2_ssl_client_cert_conn, boolean()}
| {pp2_ssl_client_cert_sess, boolean()}
| {pp2_ssl_verify, success | failed}
| {pp2_ssl_version, binary()} % US-ASCII string
| {pp2_ssl_cn, binary()} % UTF8-encoded string
| {pp2_ssl_cipher, binary()} % US-ASCII string
| {pp2_ssl_sig_alg, binary()} % US-ASCII string
| {pp2_ssl_key_alg, binary()} % US-ASCII string
).
-type(pp2_additional_field() ::
{pp2_alpn, binary()} % byte sequence
| {pp2_authority, binary()} % UTF8-encoded string
| {pp2_crc32c, integer()} % 32-bit number
| {pp2_netns, binary()} % US-ASCII string
| {pp2_ssl, list(pp2_additional_ssl_field())}
).
-record(proxy_socket, {
inet = inet4 :: inet4 | inet6 | 'unix' | 'unspec',
src_addr = {0, 0, 0, 0} :: inet:ip_address() | undefined,
dst_addr = {0, 0, 0, 0} :: inet:ip_address() | undefined,
src_port = 0 :: inet:port_number() | undefined,
dst_port = 0 :: inet:port_number() | undefined,
%% Proxy protocol v2 addtional fields
pp2_additional_info = [] :: list(pp2_additional_field())
}).
-endif.

+ 1
- 0
rebar.config Wyświetl plik

@ -9,6 +9,7 @@
]}.
{deps, [
{eNet, ".*", {git, "http://sismaker.tpddns.cn:53000/SisMaker/eNet.git", {branch, "master"}}},
{eSync, ".*", {git, "http://sismaker.tpddns.cn:53000/SisMaker/eSync.git", {branch, "master"}}}
]}.

+ 4
- 14
src/eWSrv.erl Wyświetl plik

@ -8,26 +8,16 @@
]).
startWSrv(WSrvName, Port, WsOpts) ->
ListenName = wsUtil:lsName(WSrvName),
TWsOpts = lists:keystore(conMod, 1, WsOpts, {conMod, wsHttp}),
LWsOpts = lists:keystore(conArgs, 1, TWsOpts, {conArgs, WSrvName}),
WsMod = ?wsGLV(wsMod, WsOpts, wsEgHer),
LWsOpts = lists:keystore(conArgs, 1, TWsOpts, {conArgs, WsMod}),
case ?wsGLV(sslOpts, WsOpts, false) of
false ->
{ok, _} = wsNet:openTcp(ListenName, Port, LWsOpts);
{ok, _} = wsNet:openTcp(WSrvName, Port, LWsOpts);
_ ->
{ok, _} = wsNet:openSsl(ListenName, Port, LWsOpts)
end,
{ok, _} = wsNet:openSsl(WSrvName, Port, LWsOpts)
end.
ConSupSpec = #{
id => WSrvName,
start => {wsConSup, start_link, [WSrvName, WsMod]},
restart => permanent,
shutdown => infinity,
type => supervisor,
modules => [wsConSup]
},
supervisor:start_child(eWSrv_sup, ConSupSpec).
stopWSrv(WSrvName) ->
ListenName = wsUtil:lsName(WSrvName),

+ 0
- 168
src/wsNet/proxyPt/wsPptAcceptor.erl Wyświetl plik

@ -1,168 +0,0 @@
-module(wsPptAcceptor).
-include("eNet.hrl").
-include("ntCom.hrl").
-include("ntProxyPt.hrl").
-compile(inline).
-compile({inline_size, 128}).
-export([
start_link/8
, pptAndHS/5
, init/1
, handleMsg/2
, init_it/2
, system_code_change/4
, system_continue/3
, system_get_state/1
, system_terminate/4
]).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% genActor start %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-spec start_link(list(), timeout(), boolean(), timeout(), inet:socket(), module(), term(), [proc_lib:spawn_option()]) -> {ok, pid()}.
start_link(SslOpts, SslHSTet, ProxyPt, ProxyPtTet, LSock, ConMod, ConArgs, SpawnOpts) ->
proc_lib:start_link(?MODULE, init_it, [self(), {SslOpts, SslHSTet, ProxyPt, ProxyPtTet, LSock, ConMod, ConArgs}], infinity, SpawnOpts).
init_it(Parent, Args) ->
process_flag(trap_exit, true),
modInit(Parent, Args).
-spec system_code_change(term(), module(), undefined | term(), term()) -> {ok, term()}.
system_code_change(State, _Module, _OldVsn, _Extra) ->
{ok, State}.
-spec system_continue(pid(), [], {module(), atom(), pid(), term()}) -> ok.
system_continue(_Parent, _Debug, {Parent, State}) ->
loop(Parent, State).
-spec system_get_state(term()) -> {ok, term()}.
system_get_state(State) ->
{ok, State}.
-spec system_terminate(term(), pid(), [], term()) -> none().
system_terminate(Reason, _Parent, _Debug, _State) ->
exit(Reason).
modInit(Parent, Args) ->
case init(Args) of
{ok, State} ->
proc_lib:init_ack(Parent, {ok, self()}),
loop(Parent, State);
{stop, Reason} ->
proc_lib:init_ack(Parent, {error, Reason}),
exit(Reason)
end.
loop(Parent, State) ->
receive
{system, From, Request} ->
sys:handle_system_msg(Request, From, Parent, ?MODULE, [], {Parent, State});
{'EXIT', Parent, Reason} ->
exit(Reason);
Msg ->
case handleMsg(Msg, State) of
{ok, NewState} ->
loop(Parent, NewState);
{stop, Reason} ->
exit(Reason)
end
end.
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% genActor end %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-record(state, {
lSock
, sslOpts
, sslHSTet
, proxyPt
, proxyPtTet
, ref
, conMod
, conArgs
, sockMod
}).
-spec init(Args :: term()) -> ok.
init({SslOpts, SslHSTet, ProxyPt, ProxyPtTet, LSock, ConMod, ConArgs}) ->
case prim_inet:async_accept(LSock, -1) of
{ok, Ref} ->
{ok, SockMod} = inet_db:lookup_socket(LSock),
{ok, #state{lSock = LSock, sslOpts = SslOpts, sslHSTet = SslHSTet, proxyPt = ProxyPt, proxyPtTet = ProxyPtTet, ref = Ref, conMod = ConMod, conArgs = ConArgs, sockMod = SockMod}};
{error, Reason} ->
?ntErr("init prim_inet:async_accept error ~p~n", [Reason]),
{stop, Reason}
end.
handleMsg({inet_async, LSock, Ref, Msg}, #state{lSock = LSock, sslOpts = SslOpts, sslHSTet = SslHSTet, proxyPt = ProxyPt, proxyPtTet = ProxyPtTet, ref = Ref, conMod = ConMod, conArgs = ConArgs, sockMod = SockMod} = State) ->
case Msg of
{ok, Sock} ->
%% make it look like gen_tcp:accept
inet_db:register_socket(Sock, SockMod),
try ConMod:newConn(Sock, ConArgs) of
{ok, Pid} ->
gen_tcp:controlling_process(Sock, Pid),
Pid ! {?mSockReady, Sock, SslOpts, SslHSTet, ProxyPt, ProxyPtTet},
newAsyncAccept(LSock, State);
{close, Reason} ->
?ntErr("handleMsg ConMod:newAcceptor return close ~p~n", [Reason]),
catch port_close(Sock),
newAsyncAccept(LSock, State);
_Ret ->
?ntErr("ConMod:newAcceptor return error ~p~n", [_Ret]),
{stop, error_ret}
catch
E:R:S ->
?ntErr("CliMod:newConnect crash: ~p:~p~n~p~n ~n ", [E, R, S]),
newAsyncAccept(LSock, State)
end;
{error, closed} ->
% ?ntErr("error, closed listen sock error ~p~n", [closed]),
{stop, normal};
{error, Reason} ->
?ntErr("listen sock error ~p~n", [Reason]),
{stop, {lsock, Reason}}
end;
handleMsg(_Msg, State) ->
?ntErr("~p receive unexpected ~p msg: ~p", [?MODULE, self(), _Msg]),
{ok, State}.
newAsyncAccept(LSock, State) ->
case prim_inet:async_accept(LSock, -1) of
{ok, Ref} ->
{ok, State#state{ref = Ref}};
{error, Reason} ->
?ntErr("~p prim_inet:async_accept error ~p~n", [?MODULE, Reason]),
{stop, Reason}
end.
pptAndHS(OSock, SslOpts, SslHSTet, ProxyPt, ProxyPtTet) ->
PptRet =
case ProxyPt of
true ->
ws_proxy_protocol:recv(OSock, ProxyPtTet);
_ ->
{ok, OSock, #proxy_socket{}}
end,
case PptRet of
{ok, TemSock, ProxySock} ->
case SslOpts /= undefined of
true ->
case ssl:handshake(TemSock, SslOpts, SslHSTet) of
{ok, SslSock} ->
{ok, SslSock, ProxySock};
{ok, SslSock, _Ext} -> %% OTP 21.0
{ok, SslSock, ProxySock};
{error, _} = Err -> Err
end;
_ ->
PptRet
end;
_ ->
PptRet
end.

+ 0
- 29
src/wsNet/proxyPt/wsPptAcceptorSup.erl Wyświetl plik

@ -1,29 +0,0 @@
-module(wsPptAcceptorSup).
-behaviour(supervisor).
-export([
start_link/5
]).
-export([
init/1
]).
-spec(start_link(SupName :: atom(), SslOpts :: list(), SslHSTet :: timeout(), ProxyPt :: boolean(), ProxyPtTet :: timeout()) -> {ok, pid()}).
start_link(SupName, SslOpts, SslHSTet, ProxyPt, ProxyPtTet) ->
supervisor:start_link({local, SupName}, ?MODULE, {SslOpts, SslHSTet, ProxyPt, ProxyPtTet}).
init({SslOpts, SslHSTet, ProxyPt, ProxyPtTet}) ->
SupFlags = #{strategy => simple_one_for_one, intensity => 100, period => 3600},
Acceptor = #{
id => wsPptAcceptor,
start => {wsPptAcceptor, start_link, [SslOpts, SslHSTet, ProxyPt, ProxyPtTet]},
restart => transient,
shutdown => 3000,
type => worker,
modules => [wsPptAcceptor]
},
{ok, {SupFlags, [Acceptor]}}.

+ 0
- 143
src/wsNet/proxyPt/wsPptListener.erl Wyświetl plik

@ -1,143 +0,0 @@
-module(wsPptListener).
-include("eNet.hrl").
-include("ntCom.hrl").
-compile(inline).
-compile({inline_size, 128}).
-export([
start_link/4
, getOpts/1
, getListenPort/1
, init_it/3
, system_code_change/4
, system_continue/3
, system_get_state/1
, system_terminate/4
]).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% genActor start %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-spec(start_link(atom(), atom(), inet:port_number(), [listenOpt()]) -> {ok, pid()} | ignore | {error, term()}).
start_link(ListenName, AptSupName, Port, ListenOpts) ->
proc_lib:start_link(?MODULE, init_it, [ListenName, self(), {AptSupName, Port, ListenOpts}], infinity, []).
init_it(Name, Parent, Args) ->
case safeRegister(Name) of
true ->
process_flag(trap_exit, true),
modInit(Parent, Args);
{false, Pid} ->
proc_lib:init_ack(Parent, {error, {already_started, Pid}})
end.
-spec system_code_change(term(), module(), undefined | term(), term()) -> {ok, term()}.
system_code_change(State, _Module, _OldVsn, _Extra) ->
{ok, State}.
-spec system_continue(pid(), [], {module(), atom(), pid(), term()}) -> ok.
system_continue(_Parent, _Debug, {Parent, State}) ->
loop(Parent, State).
-spec system_get_state(term()) -> {ok, term()}.
system_get_state(State) ->
{ok, State}.
-spec system_terminate(term(), pid(), [], term()) -> none().
system_terminate(Reason, _Parent, _Debug, _State) ->
exit(Reason).
safeRegister(Name) ->
try register(Name, self()) of
true -> true
catch
_:_ -> {false, whereis(Name)}
end.
modInit(Parent, Args) ->
case init(Args) of
{ok, State} ->
proc_lib:init_ack(Parent, {ok, self()}),
loop(Parent, State);
{stop, Reason} ->
proc_lib:init_ack(Parent, {error, Reason}),
exit(Reason)
end.
loop(Parent, State) ->
receive
{system, From, Request} ->
sys:handle_system_msg(Request, From, Parent, ?MODULE, [], {Parent, State});
{'EXIT', Parent, Reason} ->
terminate(Reason, State);
Msg ->
case handleMsg(Msg, State) of
{ok, NewState} ->
loop(Parent, NewState);
{stop, Reason} ->
terminate(Reason, State),
exit(Reason)
end
end.
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% genActor end %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-record(state, {
listenAddr :: inet:ip_address()
, listenPort :: inet:port_number()
, lSock :: inet:socket()
, opts :: [listenOpt()]
}).
-define(DefTcpOpts, [{nodelay, true}, {reuseaddr, true}, {send_timeout, 30000}, {send_timeout_close, true}]).
init({AptSupName, Port, ListenOpts}) ->
TcpOpts = ?getLValue(tcpOpts, ListenOpts, []),
LastTcpOpts = wsCom:mergeOpts(?DefTcpOpts, TcpOpts),
%% Don't active the socket...
case gen_tcp:listen(Port, lists:keystore(active, 1, LastTcpOpts, {active, false})) of
{ok, LSock} ->
AptCnt = ?getLValue(aptCnt, ListenOpts, ?AptCnt),
ConMod = ?getLValue(conMod, ListenOpts, undefined),
ConArgs = ?getLValue(conArgs, ListenOpts, undefined),
startAcceptor(AptCnt, LSock, AptSupName, ConMod, ConArgs),
{ok, {LAddr, LPort}} = inet:sockname(LSock),
% ?ntInfo("success to listen on ~p ~n", [Port]),
{ok, #state{listenAddr = LAddr, listenPort = LPort, lSock = LSock, opts = [{acceptors, AptCnt}, {tcpOpts, LastTcpOpts}]}};
{error, Reason} ->
?ntErr("failed to listen on ~p - ~p (~s) ~n", [Port, Reason, inet:format_error(Reason)]),
{stop, Reason}
end.
handleMsg({'$gen_call', From, miOpts}, #state{opts = Opts} = State) ->
gen_server:reply(From, Opts),
{ok, State};
handleMsg({'$gen_call', From, miListenPort}, #state{listenPort = LPort} = State) ->
gen_server:reply(From, LPort),
{ok, State};
handleMsg(_Msg, State) ->
?ntErr("~p unexpected info: ~p ~n", [?MODULE, _Msg]),
{noreply, State}.
terminate(_Reason, #state{lSock = LSock, listenAddr = Addr, listenPort = Port}) ->
?ntInfo("stopped on ~s:~p ~n", [inet:ntoa(Addr), Port]),
%% 关闭这个监听LSock 监听进程收到tcp_close 然后终止acctptor进程
catch port_close(LSock),
ok.
startAcceptor(0, _LSock, _AptSupName, _ConMod, _ConArgs) ->
ok;
startAcceptor(N, LSock, AptSupName, ConMod, ConArgs) ->
supervisor:start_child(AptSupName, [LSock, ConMod, ConArgs, []]),
startAcceptor(N - 1, LSock, AptSupName, ConMod, ConArgs).
-spec getOpts(pid()) -> [listenOpt()].
getOpts(Listener) ->
gen_server:call(Listener, miOpts).
-spec getListenPort(pid()) -> inet:port_number().
getListenPort(Listener) ->
gen_server:call(Listener, miListenPort).

+ 0
- 53
src/wsNet/proxyPt/wsPptMgrSup.erl Wyświetl plik

@ -1,53 +0,0 @@
-module(wsPptMgrSup).
-behaviour(supervisor).
-include("eNet.hrl").
-include("ntCom.hrl").
-export([
start_link/3
]).
-export([
init/1
]).
-spec(start_link(SupName :: atom(), Port :: inet:port_number(), ListenOpts :: [listenOpt()]) -> {ok, pid()} | {error, term()}).
start_link(SupName, Port, ListenOpts) ->
supervisor:start_link({local, SupName}, ?MODULE, {SupName, Port, ListenOpts}).
init({SupName, Port, ListenOpts}) ->
SupFlag = #{strategy => one_for_one, intensity => 100, period => 3600},
AptSupName = wsCom:asName(ssl, SupName),
ListenName = wsCom:lsName(ssl, SupName),
SslOpts = ?getLValue(sslOpts, ListenOpts, undefined),
SslHSTet = ?getLValue(sslHSTet, ListenOpts, ?DefSslHSTet),
ProxyPt = ?getLValue(proxyPt, ListenOpts, false),
ProxyPtTet = ?getLValue(proxyPtTet, ListenOpts, ?DefProxyPtTet),
ChildSpecs = [
#{
id => AptSupName,
start => {wsPptAcceptorSup, start_link, [AptSupName, SslOpts, SslHSTet, ProxyPt, ProxyPtTet]},
restart => permanent,
shutdown => infinity,
type => supervisor,
modules => [wsPptAcceptorSup]
},
#{
id => ListenName,
start => {wsPptListener, start_link, [ListenName, AptSupName, Port, ListenOpts]},
restart => permanent,
shutdown => 3000,
type => worker,
modules => [wsPptListener]
}],
{ok, {SupFlag, ChildSpecs}}.

+ 0
- 197
src/wsNet/proxyPt/ws_proxy_protocol.erl Wyświetl plik

@ -1,197 +0,0 @@
%%--------------------------------------------------------------------
%% 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.
%%--------------------------------------------------------------------
%% @doc [Proxy Protocol](https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt)
-module(ws_proxy_protocol).
-include("ntProxyPt.hrl").
-export([recv/2]).
-ifdef(TEST).
-export([parse_v1/3, parse_v2/5, parse_pp2_tlv/2, parse_pp2_ssl/1]).
-endif.
%% Protocol Command
-define(LOCAL, 16#0).
-define(PROXY, 16#1).
%% Address families
-define(UNSPEC, 16#0).
-define(INET, 16#1).
-define(INET6, 16#2).
-define(UNIX, 16#3).
-define(STREAM, 16#1).
-define(DGRAM, 16#2).
-define(SPACE, 16#20).
-define(TIMEOUT, 5000).
%% Proxy Protocol Additional Fields
-define(PP2_TYPE_ALPN, 16#01).
-define(PP2_TYPE_AUTHORITY, 16#02).
-define(PP2_TYPE_CRC32C, 16#03).
-define(PP2_TYPE_NOOP, 16#04).
-define(PP2_TYPE_SSL, 16#20).
-define(PP2_SUBTYPE_SSL_VERSION, 16#21).
-define(PP2_SUBTYPE_SSL_CN, 16#22).
-define(PP2_SUBTYPE_SSL_CIPHER, 16#23).
-define(PP2_SUBTYPE_SSL_SIG_ALG, 16#24).
-define(PP2_SUBTYPE_SSL_KEY_ALG, 16#25).
-define(PP2_TYPE_NETNS, 16#30).
%% Protocol signature:
%% 16#0D,16#0A,16#00,16#0D,16#0A,16#51,16#55,16#49,16#54,16#0A
-define(SIG, "\r\n\0\r\nQUIT\n").
-spec(recv(inet:socket(), timeout()) -> {ok, inet:socket(), #proxy_socket{}} | {error, term()}).
recv(Sock, Timeout) ->
{ok, OriginOpts} = inet:getopts(Sock, [mode, active, packet]),
ok = inet:setopts(Sock, [binary, {active, once}, {packet, line}]),
receive
%% V1 TCP
{_, Sock, <<"PROXY TCP", Proto, ?SPACE, ProxyInfo/binary>>} ->
inet:setopts(Sock, OriginOpts),
parse_v1(Sock, ProxyInfo, #proxy_socket{inet = inet_family(Proto)});
%% V1 Unknown
{_, _Sock, <<"PROXY UNKNOWN", _ProxyInfo/binary>>} ->
inet:setopts(Sock, OriginOpts),
{ok, Sock, #proxy_socket{}};
%% V2 TCP
{_, _Sock, <<"\r\n">>} ->
inet:setopts(Sock, [{active, false}, {packet, raw}]),
{ok, Header} = gen_tcp:recv(Sock, 14, 1000),
<<?SIG, 2:4, Cmd:4, AF:4, Trans:4, Len:16>> = Header,
case gen_tcp:recv(Sock, Len, 1000) of
{ok, ProxyInfo} ->
inet:setopts(Sock, OriginOpts),
parse_v2(Cmd, Trans, ProxyInfo, #proxy_socket{inet = inet_family(AF)}, Sock);
{error, Reason} ->
{error, {recv_proxy_info_error, Reason}}
end;
{tcp_error, _Sock, Reason} ->
{error, {recv_proxy_info_error, Reason}};
{tcp_closed, _Sock} ->
{error, {recv_proxy_info_error, tcp_closed}};
{_, _Sock, ProxyInfo} ->
{error, {invalid_proxy_info, ProxyInfo}}
after
Timeout ->
{error, proxy_proto_timeout}
end.
parse_v1(Sock, ProxyInfo, ProxySock) ->
[SrcAddrBin, DstAddrBin, SrcPortBin, DstPortBin] = binary:split(ProxyInfo, [<<" ">>, <<"\r\n">>], [global, trim]),
{ok, SrcAddr} = inet:parse_address(binary_to_list(SrcAddrBin)),
{ok, DstAddr} = inet:parse_address(binary_to_list(DstAddrBin)),
SrcPort = list_to_integer(binary_to_list(SrcPortBin)),
DstPort = list_to_integer(binary_to_list(DstPortBin)),
{ok, Sock, ProxySock#proxy_socket{src_addr = SrcAddr, dst_addr = DstAddr, src_port = SrcPort, dst_port = DstPort}}.
parse_v2(?LOCAL, _Trans, _ProxyInfo, ProxySocket, Sock) ->
{ok, Sock, ProxySocket};
parse_v2(?PROXY, ?STREAM, ProxyInfo, ProxySock = #proxy_socket{inet = inet4}, Sock) ->
<<A:8, B:8, C:8, D:8, W:8, X:8, Y:8, Z:8, SrcPort:16, DstPort:16, AdditionalBytes/binary>> = ProxyInfo,
parse_pp2_additional(AdditionalBytes, ProxySock#proxy_socket{src_addr = {A, B, C, D}, src_port = SrcPort, dst_addr = {W, X, Y, Z}, dst_port = DstPort}, Sock);
parse_v2(?PROXY, ?STREAM, ProxyInfo, ProxySock = #proxy_socket{inet = inet6}, Sock) ->
<<A:16, B:16, C:16, D:16, E:16, F:16, G:16, H:16, R:16, S:16, T:16, U:16, V:16, W:16, X:16, Y:16, SrcPort:16, DstPort:16, AdditionalBytes/binary>> = ProxyInfo,
parse_pp2_additional(AdditionalBytes, ProxySock#proxy_socket{src_addr = {A, B, C, D, E, F, G, H}, src_port = SrcPort, dst_addr = {R, S, T, U, V, W, X, Y}, dst_port = DstPort}, Sock);
parse_v2(_, _, _, _, _) ->
{error, unsupported_proto_v2}.
parse_pp2_additional(<<>>, ProxySock, Sock) ->
{ok, Sock, ProxySock};
parse_pp2_additional(Bytes, ProxySock, Sock) when is_binary(Bytes) ->
IgnoreGuard = fun(?PP2_TYPE_NOOP) -> false; (_Type) -> true end,
AdditionalInfo = parse_pp2_tlv(fun pp2_additional_field/1, Bytes, IgnoreGuard),
{ok, Sock, ProxySock#proxy_socket{pp2_additional_info = AdditionalInfo}}.
parse_pp2_tlv(Fun, Bytes) ->
parse_pp2_tlv(Fun, Bytes, fun(_Any) -> true end).
parse_pp2_tlv(Fun, Bytes, Guard) ->
[Fun({Type, Val}) || <<Type:8, Len:16, Val:Len/binary>> <= Bytes, Guard(Type)].
pp2_additional_field({?PP2_TYPE_ALPN, PP2_ALPN}) ->
{pp2_alpn, PP2_ALPN};
pp2_additional_field({?PP2_TYPE_AUTHORITY, PP2_AUTHORITY}) ->
{pp2_authority, PP2_AUTHORITY};
pp2_additional_field({?PP2_TYPE_CRC32C, PP2_CRC32C}) ->
{pp2_crc32c, PP2_CRC32C};
pp2_additional_field({?PP2_TYPE_NETNS, PP2_NETNS}) ->
{pp2_netns, PP2_NETNS};
pp2_additional_field({?PP2_TYPE_SSL, PP2_SSL}) ->
{pp2_ssl, parse_pp2_ssl(PP2_SSL)};
pp2_additional_field({Field, Value}) ->
{{pp2_raw, Field}, Value}.
parse_pp2_ssl(<<_Unused:5, PP2_CLIENT_CERT_SESS:1, PP2_CLIENT_CERT_CONN:1, PP2_CLIENT_SSL:1,
PP2_SSL_VERIFY:32, SubFields/bitstring>>) ->
[
%% The PP2_CLIENT_SSL flag indicates that the client connected over SSL/TLS. When
%% this field is present, the US-ASCII string representation of the TLS version is
%% appended at the end of the field in the TLV format using the type PP2_SUBTYPE_SSL_VERSION.
{pp2_ssl_client, bool(PP2_CLIENT_SSL)},
%% PP2_CLIENT_CERT_CONN indicates that the client provided a certificate over the
%% current connection.
{pp2_ssl_client_cert_conn, bool(PP2_CLIENT_CERT_CONN)},
%% PP2_CLIENT_CERT_SESS indicates that the client provided a
%% certificate at least once over the TLS session this connection belongs to.
{pp2_ssl_client_cert_sess, bool(PP2_CLIENT_CERT_SESS)},
%% The <verify> field will be zero if the client presented a certificate
%% and it was successfully verified, and non-zero otherwise.
{pp2_ssl_verify, ssl_certificate_verified(PP2_SSL_VERIFY)}
| parse_pp2_tlv(fun pp2_additional_ssl_field/1, SubFields)
].
pp2_additional_ssl_field({?PP2_SUBTYPE_SSL_VERSION, PP2_SSL_VERSION}) ->
{pp2_ssl_version, PP2_SSL_VERSION};
%% In all cases, the string representation (in UTF8) of the Common Name field
%% (OID: 2.5.4.3) of the client certificate's Distinguished Name, is appended
%% using the TLV format and the type PP2_SUBTYPE_SSL_CN. E.g. "example.com".
pp2_additional_ssl_field({?PP2_SUBTYPE_SSL_CN, PP2_SSL_CN}) ->
{pp2_ssl_cn, PP2_SSL_CN};
pp2_additional_ssl_field({?PP2_SUBTYPE_SSL_CIPHER, PP2_SSL_CIPHER}) ->
{pp2_ssl_cipher, PP2_SSL_CIPHER};
pp2_additional_ssl_field({?PP2_SUBTYPE_SSL_SIG_ALG, PP2_SSL_SIG_ALG}) ->
{pp2_ssl_sig_alg, PP2_SSL_SIG_ALG};
pp2_additional_ssl_field({?PP2_SUBTYPE_SSL_KEY_ALG, PP2_SSL_KEY_ALG}) ->
{pp2_ssl_key_alg, PP2_SSL_KEY_ALG};
pp2_additional_ssl_field({Field, Val}) ->
{{pp2_ssl_raw, Field}, Val}.
ssl_certificate_verified(0) -> success;
ssl_certificate_verified(_) -> failed.
%% V1
inet_family($4) -> inet4;
inet_family($6) -> inet6;
%% V2
inet_family(?UNSPEC) -> unspec;
inet_family(?INET) -> inet4;
inet_family(?INET6) -> inet6;
inet_family(?UNIX) -> unix.
bool(1) -> true;
bool(_) -> false.

+ 0
- 145
src/wsNet/ssl/wsSslAcceptor.erl Wyświetl plik

@ -1,145 +0,0 @@
-module(wsSslAcceptor).
-include("eNet.hrl").
-include("ntCom.hrl").
-compile(inline).
-compile({inline_size, 128}).
-export([
start_link/6
, handshake/3
, init/1
, handleMsg/2
, init_it/2
, system_code_change/4
, system_continue/3
, system_get_state/1
, system_terminate/4
]).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% genActor start %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-spec start_link(list(), timeout(), inet:socket(), module(), term(), [proc_lib:spawn_option()]) -> {ok, pid()}.
start_link(SslOpts, HandshakeTimeout, LSock, ConMod, ConArgs, SpawnOpts) ->
proc_lib:start_link(?MODULE, init_it, [self(), {SslOpts, HandshakeTimeout, LSock, ConMod, ConArgs}], infinity, SpawnOpts).
init_it(Parent, Args) ->
process_flag(trap_exit, true),
modInit(Parent, Args).
-spec system_code_change(term(), module(), undefined | term(), term()) -> {ok, term()}.
system_code_change(State, _Module, _OldVsn, _Extra) ->
{ok, State}.
-spec system_continue(pid(), [], {module(), atom(), pid(), term()}) -> ok.
system_continue(_Parent, _Debug, {Parent, State}) ->
loop(Parent, State).
-spec system_get_state(term()) -> {ok, term()}.
system_get_state(State) ->
{ok, State}.
-spec system_terminate(term(), pid(), [], term()) -> none().
system_terminate(Reason, _Parent, _Debug, _State) ->
exit(Reason).
modInit(Parent, Args) ->
case init(Args) of
{ok, State} ->
proc_lib:init_ack(Parent, {ok, self()}),
loop(Parent, State);
{stop, Reason} ->
proc_lib:init_ack(Parent, {error, Reason}),
exit(Reason)
end.
loop(Parent, State) ->
receive
{system, From, Request} ->
sys:handle_system_msg(Request, From, Parent, ?MODULE, [], {Parent, State});
{'EXIT', Parent, Reason} ->
exit(Reason);
Msg ->
case handleMsg(Msg, State) of
{ok, NewState} ->
loop(Parent, NewState);
{stop, Reason} ->
exit(Reason)
end
end.
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% genActor end %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-record(state, {
lSock
, sslOpts
, sslHSTet
, ref
, conMod
, conArgs
, sockMod
}).
-spec init(Args :: term()) -> ok.
init({SslOpts, SslHSTet, LSock, ConMod, ConArgs}) ->
case prim_inet:async_accept(LSock, -1) of
{ok, Ref} ->
{ok, SockMod} = inet_db:lookup_socket(LSock),
{ok, #state{lSock = LSock, sslOpts = SslOpts, sslHSTet = SslHSTet, ref = Ref, conMod = ConMod, conArgs = ConArgs, sockMod = SockMod}};
{error, Reason} ->
?ntErr("init prim_inet:async_accept error ~p~n", [Reason]),
{stop, Reason}
end.
handleMsg({inet_async, LSock, Ref, Msg}, #state{lSock = LSock, sslOpts = SslOpts, sslHSTet = SslHSTet, ref = Ref, conMod = ConMod, conArgs = ConArgs, sockMod = SockMod} = State) ->
case Msg of
{ok, Sock} ->
%% make it look like gen_tcp:accept
inet_db:register_socket(Sock, SockMod),
try ConMod:newConn(Sock, ConArgs) of
{ok, Pid} ->
gen_tcp:controlling_process(Sock, Pid),
Pid ! {?mSockReady, Sock, SslOpts, SslHSTet},
newAsyncAccept(LSock, State);
{close, Reason} ->
?ntErr("handleMsg ConMod:newAcceptor return close ~p~n", [Reason]),
catch port_close(Sock),
newAsyncAccept(LSock, State);
_Ret ->
?ntErr("ConMod:newAcceptor return error ~p~n", [_Ret]),
{stop, error_ret}
catch
E:R:S ->
?ntErr("CliMod:newConnect crash: ~p:~p~n~p~n ~n ", [E, R, S]),
newAsyncAccept(LSock, State)
end;
{error, closed} ->
% ?ntErr("error, closed listen sock error ~p~n", [closed]),
{stop, normal};
{error, Reason} ->
?ntErr("listen sock error ~p~n", [Reason]),
{stop, {lsock, Reason}}
end;
handleMsg(_Msg, State) ->
?ntErr("~p receive unexpected ~p msg: ~p", [?MODULE, self(), _Msg]),
{ok, State}.
newAsyncAccept(LSock, State) ->
case prim_inet:async_accept(LSock, -1) of
{ok, Ref} ->
{ok, State#state{ref = Ref}};
{error, Reason} ->
?ntErr("~p prim_inet:async_accept error ~p~n", [?MODULE, Reason]),
{stop, Reason}
end.
handshake(Sock, SslOpts, Timeout) ->
case ssl:handshake(Sock, SslOpts, Timeout) of
{ok, _SslSock} = Ret ->
Ret;
{ok, SslSock, _Ext} -> %% OTP 21.0
{ok, SslSock};
{error, _} = Err -> Err
end.

+ 0
- 29
src/wsNet/ssl/wsSslAcceptorSup.erl Wyświetl plik

@ -1,29 +0,0 @@
-module(wsSslAcceptorSup).
-behaviour(supervisor).
-export([
start_link/3
]).
-export([
init/1
]).
-spec(start_link(SupName :: atom(), SslOpts :: list(), HandshakeTimeout :: timeout()) -> {ok, pid()}).
start_link(SupName, SslOpts, HandshakeTimeout) ->
supervisor:start_link({local, SupName}, ?MODULE, {SslOpts, HandshakeTimeout}).
init({SslOpts, HandshakeTimeout}) ->
SupFlags = #{strategy => simple_one_for_one, intensity => 100, period => 3600},
Acceptor = #{
id => wsSslAcceptor,
start => {wsSslAcceptor, start_link, [SslOpts, HandshakeTimeout]},
restart => transient,
shutdown => 3000,
type => worker,
modules => [wsSslAcceptor]
},
{ok, {SupFlags, [Acceptor]}}.

+ 0
- 143
src/wsNet/ssl/wsSslListener.erl Wyświetl plik

@ -1,143 +0,0 @@
-module(wsSslListener).
-include("eNet.hrl").
-include("ntCom.hrl").
-compile(inline).
-compile({inline_size, 128}).
-export([
start_link/4
, getOpts/1
, getListenPort/1
, init_it/3
, system_code_change/4
, system_continue/3
, system_get_state/1
, system_terminate/4
]).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% genActor start %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-spec(start_link(atom(), atom(), inet:port_number(), [listenOpt()]) -> {ok, pid()} | ignore | {error, term()}).
start_link(ListenName, AptSupName, Port, ListenOpts) ->
proc_lib:start_link(?MODULE, init_it, [ListenName, self(), {AptSupName, Port, ListenOpts}], infinity, []).
init_it(Name, Parent, Args) ->
case safeRegister(Name) of
true ->
process_flag(trap_exit, true),
modInit(Parent, Args);
{false, Pid} ->
proc_lib:init_ack(Parent, {error, {already_started, Pid}})
end.
-spec system_code_change(term(), module(), undefined | term(), term()) -> {ok, term()}.
system_code_change(State, _Module, _OldVsn, _Extra) ->
{ok, State}.
-spec system_continue(pid(), [], {module(), atom(), pid(), term()}) -> ok.
system_continue(_Parent, _Debug, {Parent, State}) ->
loop(Parent, State).
-spec system_get_state(term()) -> {ok, term()}.
system_get_state(State) ->
{ok, State}.
-spec system_terminate(term(), pid(), [], term()) -> none().
system_terminate(Reason, _Parent, _Debug, _State) ->
exit(Reason).
safeRegister(Name) ->
try register(Name, self()) of
true -> true
catch
_:_ -> {false, whereis(Name)}
end.
modInit(Parent, Args) ->
case init(Args) of
{ok, State} ->
proc_lib:init_ack(Parent, {ok, self()}),
loop(Parent, State);
{stop, Reason} ->
proc_lib:init_ack(Parent, {error, Reason}),
exit(Reason)
end.
loop(Parent, State) ->
receive
{system, From, Request} ->
sys:handle_system_msg(Request, From, Parent, ?MODULE, [], {Parent, State});
{'EXIT', Parent, Reason} ->
terminate(Reason, State);
Msg ->
case handleMsg(Msg, State) of
{ok, NewState} ->
loop(Parent, NewState);
{stop, Reason} ->
terminate(Reason, State),
exit(Reason)
end
end.
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% genActor end %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-record(state, {
listenAddr :: inet:ip_address()
, listenPort :: inet:port_number()
, lSock :: inet:socket()
, opts :: [listenOpt()]
}).
-define(DefTcpOpts, [{nodelay, true}, {reuseaddr, true}, {send_timeout, 30000}, {send_timeout_close, true}]).
init({AptSupName, Port, ListenOpts}) ->
TcpOpts = ?getLValue(tcpOpts, ListenOpts, []),
LastTcpOpts = wsCom:mergeOpts(?DefTcpOpts, TcpOpts),
%% Don't active the socket...
case gen_tcp:listen(Port, lists:keystore(active, 1, LastTcpOpts, {active, false})) of
{ok, LSock} ->
AptCnt = ?getLValue(aptCnt, ListenOpts, ?AptCnt),
ConMod = ?getLValue(conMod, ListenOpts, undefined),
ConArgs = ?getLValue(conArgs, ListenOpts, undefined),
startAcceptor(AptCnt, LSock, AptSupName, ConMod, ConArgs),
{ok, {LAddr, LPort}} = inet:sockname(LSock),
% ?ntInfo("success to listen on ~p ~n", [Port]),
{ok, #state{listenAddr = LAddr, listenPort = LPort, lSock = LSock, opts = [{acceptors, AptCnt}, {tcpOpts, LastTcpOpts}]}};
{error, Reason} ->
?ntErr("failed to listen on ~p - ~p (~s) ~n", [Port, Reason, inet:format_error(Reason)]),
{stop, Reason}
end.
handleMsg({'$gen_call', From, miOpts}, #state{opts = Opts} = State) ->
gen_server:reply(From, Opts),
{ok, State};
handleMsg({'$gen_call', From, miListenPort}, #state{listenPort = LPort} = State) ->
gen_server:reply(From, LPort),
{ok, State};
handleMsg(_Msg, State) ->
?ntErr("~p unexpected info: ~p ~n", [?MODULE, _Msg]),
{noreply, State}.
terminate(_Reason, #state{lSock = LSock, listenAddr = Addr, listenPort = Port}) ->
?ntInfo("stopped on ~s:~p ~n", [inet:ntoa(Addr), Port]),
%% 关闭这个监听LSock 监听进程收到tcp_close 然后终止acctptor进程
catch port_close(LSock),
ok.
startAcceptor(0, _LSock, _AptSupName, _ConMod, _ConArgs) ->
ok;
startAcceptor(N, LSock, AptSupName, ConMod, ConArgs) ->
supervisor:start_child(AptSupName, [LSock, ConMod, ConArgs, []]),
startAcceptor(N - 1, LSock, AptSupName, ConMod, ConArgs).
-spec getOpts(pid()) -> [listenOpt()].
getOpts(Listener) ->
gen_server:call(Listener, miOpts).
-spec getListenPort(pid()) -> inet:port_number().
getListenPort(Listener) ->
gen_server:call(Listener, miListenPort).

+ 0
- 51
src/wsNet/ssl/wsSslMgrSup.erl Wyświetl plik

@ -1,51 +0,0 @@
-module(wsSslMgrSup).
-behaviour(supervisor).
-include("eNet.hrl").
-include("ntCom.hrl").
-export([
start_link/3
]).
-export([
init/1
]).
-spec(start_link(SupName :: atom(), Port :: inet:port_number(), ListenOpts :: [listenOpt()]) -> {ok, pid()} | {error, term()}).
start_link(SupName, Port, ListenOpts) ->
supervisor:start_link({local, SupName}, ?MODULE, {SupName, Port, ListenOpts}).
init({SupName, Port, ListenOpts}) ->
SupFlag = #{strategy => one_for_one, intensity => 100, period => 3600},
AptSupName = wsCom:asName(ssl, SupName),
ListenName = wsCom:lsName(ssl, SupName),
SslOpts = ?getLValue(sslOpts, ListenOpts, []),
SslHSTet = ?getLValue(sslHSTet, ListenOpts, ?DefSslHSTet),
ChildSpecs = [
#{
id => AptSupName,
start => {wsSslAcceptorSup, start_link, [AptSupName, SslOpts, SslHSTet]},
restart => permanent,
shutdown => infinity,
type => supervisor,
modules => [wsSslAcceptorSup]
},
#{
id => ListenName,
start => {wsSslListener, start_link, [ListenName, AptSupName, Port, ListenOpts]},
restart => permanent,
shutdown => 3000,
type => worker,
modules => [wsSslListener]
}],
{ok, {SupFlag, ChildSpecs}}.

+ 0
- 133
src/wsNet/tcp/wsTcpAcceptor.erl Wyświetl plik

@ -1,133 +0,0 @@
-module(wsTcpAcceptor).
-include("eNet.hrl").
-include("ntCom.hrl").
-compile(inline).
-compile({inline_size, 128}).
-export([
start_link/4
, init/1
, handleMsg/2
, init_it/2
, system_code_change/4
, system_continue/3
, system_get_state/1
, system_terminate/4
]).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% genActor start %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-spec start_link(inet:socket(), module(), term(), [proc_lib:spawn_option()]) -> {ok, pid()}.
start_link(LSock, ConMod, ConArgs, SpawnOpts) ->
proc_lib:start_link(?MODULE, init_it, [self(), {LSock, ConMod, ConArgs}], infinity, SpawnOpts).
init_it(Parent, Args) ->
process_flag(trap_exit, true),
modInit(Parent, Args).
-spec system_code_change(term(), module(), undefined | term(), term()) -> {ok, term()}.
system_code_change(State, _Module, _OldVsn, _Extra) ->
{ok, State}.
-spec system_continue(pid(), [], {module(), atom(), pid(), term()}) -> ok.
system_continue(_Parent, _Debug, {Parent, State}) ->
loop(Parent, State).
-spec system_get_state(term()) -> {ok, term()}.
system_get_state(State) ->
{ok, State}.
-spec system_terminate(term(), pid(), [], term()) -> none().
system_terminate(Reason, _Parent, _Debug, _State) ->
exit(Reason).
modInit(Parent, Args) ->
case init(Args) of
{ok, State} ->
proc_lib:init_ack(Parent, {ok, self()}),
loop(Parent, State);
{stop, Reason} ->
proc_lib:init_ack(Parent, {error, Reason}),
exit(Reason)
end.
loop(Parent, State) ->
receive
{system, From, Request} ->
sys:handle_system_msg(Request, From, Parent, ?MODULE, [], {Parent, State});
{'EXIT', Parent, Reason} ->
exit(Reason);
Msg ->
case handleMsg(Msg, State) of
{ok, NewState} ->
loop(Parent, NewState);
{stop, Reason} ->
exit(Reason)
end
end.
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% genActor end %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-record(state, {
lSock
, ref
, conMod
, conArgs
, sockMod
}).
-spec init(Args :: term()) -> ok.
init({LSock, ConMod, ConArgs}) ->
case prim_inet:async_accept(LSock, -1) of
{ok, Ref} ->
{ok, SockMod} = inet_db:lookup_socket(LSock),
{ok, #state{lSock = LSock, ref = Ref, conMod = ConMod, conArgs = ConArgs, sockMod = SockMod}};
{error, Reason} ->
?ntErr("init prim_inet:async_accept error ~p~n", [Reason]),
{stop, Reason}
end.
handleMsg({inet_async, LSock, Ref, Msg}, #state{lSock = LSock, ref = Ref, conMod = ConMod, conArgs = ConArgs, sockMod = SockMod} = State) ->
case Msg of
{ok, Sock} ->
%% make it look like gen_tcp:accept
inet_db:register_socket(Sock, SockMod),
try ConMod:newConn(Sock, ConArgs) of
{ok, Pid} ->
gen_tcp:controlling_process(Sock, Pid),
Pid ! {?mSockReady, Sock},
newAsyncAccept(LSock, State);
{close, Reason} ->
?ntErr("handleMsg ConMod:newAcceptor return close ~p~n", [Reason]),
catch port_close(Sock),
newAsyncAccept(LSock, State);
_Ret ->
?ntErr("ConMod:newAcceptor return error ~p~n", [_Ret]),
{stop, error_ret}
catch
E:R:S ->
?ntErr("CliMod:newConnect crash: ~p:~p~n~p~n ~n ", [E, R, S]),
newAsyncAccept(LSock, State)
end;
{error, closed} ->
% ?ntErr("error, closed listen sock error ~p~n", [closed]),
{stop, normal};
{error, Reason} ->
?ntErr("listen sock error ~p~n", [Reason]),
{stop, {lsock, Reason}}
end;
handleMsg(_Msg, State) ->
?ntErr("~p receive unexpected ~p msg: ~p", [?MODULE, self(), _Msg]),
{ok, State}.
newAsyncAccept(LSock, State) ->
case prim_inet:async_accept(LSock, -1) of
{ok, Ref} ->
{ok, State#state{ref = Ref}};
{error, Reason} ->
?ntErr("~p prim_inet:async_accept error ~p~n", [?MODULE, Reason]),
{stop, Reason}
end.

+ 0
- 29
src/wsNet/tcp/wsTcpAcceptorSup.erl Wyświetl plik

@ -1,29 +0,0 @@
-module(wsTcpAcceptorSup).
-behaviour(supervisor).
-export([
start_link/1
]).
-export([
init/1
]).
-spec(start_link(SupName :: atom()) -> {ok, pid()}).
start_link(SupName) ->
supervisor:start_link({local, SupName}, ?MODULE, undefined).
init(_Args) ->
SupFlags = #{strategy => simple_one_for_one, intensity => 100, period => 3600},
Acceptor = #{
id => wsTcpAcceptor,
start => {wsTcpAcceptor, start_link, []},
restart => transient,
shutdown => 3000,
type => worker,
modules => [wsTcpAcceptor]
},
{ok, {SupFlags, [Acceptor]}}.

+ 0
- 143
src/wsNet/tcp/wsTcpListener.erl Wyświetl plik

@ -1,143 +0,0 @@
-module(wsTcpListener).
-include("eNet.hrl").
-include("ntCom.hrl").
-compile(inline).
-compile({inline_size, 128}).
-export([
start_link/4
, getOpts/1
, getListenPort/1
, init_it/3
, system_code_change/4
, system_continue/3
, system_get_state/1
, system_terminate/4
]).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% genActor start %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-spec(start_link(atom(), atom(), inet:port_number(), [listenOpt()]) -> {ok, pid()} | ignore | {error, term()}).
start_link(ListenName, AptSupName, Port, ListenOpts) ->
proc_lib:start_link(?MODULE, init_it, [ListenName, self(), {AptSupName, Port, ListenOpts}], infinity, []).
init_it(Name, Parent, Args) ->
case safeRegister(Name) of
true ->
process_flag(trap_exit, true),
modInit(Parent, Args);
{false, Pid} ->
proc_lib:init_ack(Parent, {error, {already_started, Pid}})
end.
-spec system_code_change(term(), module(), undefined | term(), term()) -> {ok, term()}.
system_code_change(State, _Module, _OldVsn, _Extra) ->
{ok, State}.
-spec system_continue(pid(), [], {module(), atom(), pid(), term()}) -> ok.
system_continue(_Parent, _Debug, {Parent, State}) ->
loop(Parent, State).
-spec system_get_state(term()) -> {ok, term()}.
system_get_state(State) ->
{ok, State}.
-spec system_terminate(term(), pid(), [], term()) -> none().
system_terminate(Reason, _Parent, _Debug, _State) ->
exit(Reason).
safeRegister(Name) ->
try register(Name, self()) of
true -> true
catch
_:_ -> {false, whereis(Name)}
end.
modInit(Parent, Args) ->
case init(Args) of
{ok, State} ->
proc_lib:init_ack(Parent, {ok, self()}),
loop(Parent, State);
{stop, Reason} ->
proc_lib:init_ack(Parent, {error, Reason}),
exit(Reason)
end.
loop(Parent, State) ->
receive
{system, From, Request} ->
sys:handle_system_msg(Request, From, Parent, ?MODULE, [], {Parent, State});
{'EXIT', Parent, Reason} ->
terminate(Reason, State);
Msg ->
case handleMsg(Msg, State) of
{ok, NewState} ->
loop(Parent, NewState);
{stop, Reason} ->
terminate(Reason, State),
exit(Reason)
end
end.
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% genActor end %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-record(state, {
listenAddr :: inet:ip_address()
, listenPort :: inet:port_number()
, lSock :: inet:socket()
, opts :: [listenOpt()]
}).
-define(DefTcpOpts, [{nodelay, true}, {reuseaddr, true}, {send_timeout, 30000}, {send_timeout_close, true}]).
init({AptSupName, Port, ListenOpts}) ->
TcpOpts = ?getLValue(tcpOpts, ListenOpts, []),
LastTcpOpts = wsCom:mergeOpts(?DefTcpOpts, TcpOpts),
%% Don't active the socket...
case gen_tcp:listen(Port, lists:keystore(active, 1, LastTcpOpts, {active, false})) of
{ok, LSock} ->
AptCnt = ?getLValue(aptCnt, ListenOpts, ?AptCnt),
ConMod = ?getLValue(conMod, ListenOpts, undefined),
ConArgs = ?getLValue(conArgs, ListenOpts, undefined),
startAcceptor(AptCnt, LSock, AptSupName, ConMod, ConArgs),
{ok, {LAddr, LPort}} = inet:sockname(LSock),
% ?ntInfo("success to listen on ~p ~n", [Port]),
{ok, #state{listenAddr = LAddr, listenPort = LPort, lSock = LSock, opts = [{acceptors, AptCnt}, {tcpOpts, LastTcpOpts}]}};
{error, Reason} ->
?ntErr("failed to listen on ~p - ~p (~s) ~n", [Port, Reason, inet:format_error(Reason)]),
{stop, Reason}
end.
handleMsg({'$gen_call', From, miOpts}, #state{opts = Opts} = State) ->
gen_server:reply(From, Opts),
{ok, State};
handleMsg({'$gen_call', From, miListenPort}, #state{listenPort = LPort} = State) ->
gen_server:reply(From, LPort),
{ok, State};
handleMsg(_Msg, State) ->
?ntErr("~p unexpected info: ~p ~n", [?MODULE, _Msg]),
{noreply, State}.
terminate(_Reason, #state{lSock = LSock, listenAddr = Addr, listenPort = Port}) ->
?ntInfo("stopped on ~s:~p ~n", [inet:ntoa(Addr), Port]),
%% 关闭这个监听LSock 监听进程收到tcp_close 然后终止acctptor进程
catch port_close(LSock),
ok.
startAcceptor(0, _LSock, _AptSupName, _ConMod, _ConArgs) ->
ok;
startAcceptor(N, LSock, AptSupName, ConMod, ConArgs) ->
supervisor:start_child(AptSupName, [LSock, ConMod, ConArgs, []]),
startAcceptor(N - 1, LSock, AptSupName, ConMod, ConArgs).
-spec getOpts(pid()) -> [listenOpt()].
getOpts(Listener) ->
gen_server:call(Listener, miOpts).
-spec getListenPort(pid()) -> inet:port_number().
getListenPort(Listener) ->
gen_server:call(Listener, miListenPort).

+ 0
- 48
src/wsNet/tcp/wsTcpMgrSup.erl Wyświetl plik

@ -1,48 +0,0 @@
-module(wsTcpMgrSup).
-behaviour(supervisor).
-include("eNet.hrl").
-include("ntCom.hrl").
-export([
start_link/3
]).
-export([
init/1
]).
-spec(start_link(SupName :: atom(), Port :: inet:port_number(), ListenOpts :: [listenOpt()]) -> {ok, pid()} | {error, term()}).
start_link(SupName, Port, ListenOpts) ->
supervisor:start_link({local, SupName}, ?MODULE, {SupName, Port, ListenOpts}).
init({SupName, Port, ListenOpts}) ->
SupFlag = #{strategy => one_for_one, intensity => 100, period => 3600},
AptSupName = wsCom:asName(tcp, SupName),
ListenName = wsCom:lsName(tcp, SupName),
ChildSpecs = [
#{
id => AptSupName,
start => {wsTcpAcceptorSup, start_link, [AptSupName]},
restart => permanent,
shutdown => infinity,
type => supervisor,
modules => [wsTcpAcceptorSup]
},
#{
id => ListenName,
start => {wsTcpListener, start_link, [ListenName, AptSupName, Port, ListenOpts]},
restart => permanent,
shutdown => 3000,
type => worker,
modules => [wsTcpListener]
}],
{ok, {SupFlag, ChildSpecs}}.

+ 0
- 118
src/wsNet/wsCom.erl Wyświetl plik

@ -1,118 +0,0 @@
-module(wsCom).
-compile([export_all, nowarn_export_all]).
-spec mergeOpts(Defaults :: list(), Options :: list()) -> list().
mergeOpts(Defaults, Options) ->
lists:foldl(
fun({Opt, Val}, Acc) ->
lists:keystore(Opt, 1, Acc, {Opt, Val});
(Opt, Acc) ->
lists:usort([Opt | Acc])
end,
Defaults, Options).
mergeAddr({Addr, _Port}, SockOpts) ->
lists:keystore(ip, 1, SockOpts, {ip, Addr});
mergeAddr(_Port, SockOpts) ->
SockOpts.
getPort({_Addr, Port}) -> Port;
getPort(Port) -> Port.
fixIpPort(IpOrStr, Port) ->
if
is_list(IpOrStr), is_integer(Port) ->
{ok, IP} = inet:parse_address(v),
{IP, Port};
is_tuple(IpOrStr), is_integer(Port) ->
case isIpv4OrIpv6(IpOrStr) of
true ->
{IpOrStr, Port};
false ->
error({invalid_ip, IpOrStr})
end;
true ->
error({invalid_ip_port, IpOrStr, Port})
end.
parseAddr({Addr, Port}) when is_list(Addr), is_integer(Port) ->
{ok, IPAddr} = inet:parse_address(Addr),
{IPAddr, Port};
parseAddr({Addr, Port}) when is_tuple(Addr), is_integer(Port) ->
case isIpv4OrIpv6(Addr) of
true ->
{Addr, Port};
false ->
error(invalid_ipaddr)
end;
parseAddr(Port) ->
Port.
isIpv4OrIpv6({A, B, C, D}) ->
A >= 0 andalso A =< 255 andalso
B >= 0 andalso B =< 255 andalso
C >= 0 andalso C =< 255 andalso
D >= 0 andalso D =< 255;
isIpv4OrIpv6({A, B, C, D, E, F, G, H}) ->
A >= 0 andalso A =< 65535 andalso
B >= 0 andalso B =< 65535 andalso
C >= 0 andalso C =< 65535 andalso
D >= 0 andalso D =< 65535 andalso
E >= 0 andalso E =< 65535 andalso
F >= 0 andalso F =< 65535 andalso
G >= 0 andalso G =< 65535 andalso
H >= 0 andalso H =< 65535;
isIpv4OrIpv6(_) ->
false.
%% @doc Return true if the value is an ipv4 address
isIpv4({A, B, C, D}) ->
A >= 0 andalso A =< 255 andalso
B >= 0 andalso B =< 255 andalso
C >= 0 andalso C =< 255 andalso
D >= 0 andalso D =< 255;
isIpv4(_) ->
false.
%% @doc Return true if the value is an ipv6 address
isIpv6({A, B, C, D, E, F, G, H}) ->
A >= 0 andalso A =< 65535 andalso
B >= 0 andalso B =< 65535 andalso
C >= 0 andalso C =< 65535 andalso
D >= 0 andalso D =< 65535 andalso
E >= 0 andalso E =< 65535 andalso
F >= 0 andalso F =< 65535 andalso
G >= 0 andalso G =< 65535 andalso
H >= 0 andalso H =< 65535;
isIpv6(_) ->
false.
getListValue(Key, List, Default) ->
case lists:keyfind(Key, 1, List) of
false ->
Default;
{Key, Value} ->
Value
end.
serverName(PoolName, Index) ->
list_to_atom(atom_to_list(PoolName) ++ "_" ++ integer_to_list(Index)).
asName(tcp, PrName) ->
binary_to_atom(<<(atom_to_binary(PrName))/binary, "TAs">>);
asName(ssl, PrName) ->
binary_to_atom(<<(atom_to_binary(PrName))/binary, "SAs">>);
asName(udp, PrName) ->
binary_to_atom(<<(atom_to_binary(PrName))/binary, "UOs">>).
lsName(tcp, PrName) ->
binary_to_atom(<<(atom_to_binary(PrName))/binary, "TLs">>);
lsName(ssl, PrName) ->
binary_to_atom(<<(atom_to_binary(PrName))/binary, "SLs">>);
lsName(udp, PrName) ->
binary_to_atom(<<(atom_to_binary(PrName))/binary, "URs">>).

+ 0
- 69
src/wsNet/wsNet.erl Wyświetl plik

@ -1,69 +0,0 @@
-module(wsNet).
-include("eNet.hrl").
-export([
openTcp/3
, openSsl/3
, openUdp/3
, openPpt/3
, close/1
]).
%% add a TCP listener
-spec openTcp(ListenName :: atom(), Port :: inet:port_number(), ListenOpts :: [listenOpt()]) -> {ok, pid()} | {error, term()}.
openTcp(ListenName, Port, ListenOpts) ->
TcpMgrSupSpec = #{
id => ListenName,
start => {wsTcpMgrSup, start_link, [ListenName, Port, ListenOpts]},
restart => permanent,
shutdown => infinity,
type => supervisor,
modules => [wsTcpMgrSup]
},
supervisor:start_child(eWSrv_sup, TcpMgrSupSpec).
%% add a Ssl listener
-spec openSsl(ListenName :: atom(), Port :: inet:port_number(), ListenOpts :: [listenOpt()]) -> {ok, pid()} | {error, term()}.
openSsl(ListenName, Port, ListenOpts) ->
SslMgrSupSpec = #{
id => ListenName,
start => {wsntSslMgrSup, start_link, [ListenName, Port, ListenOpts]},
restart => permanent,
shutdown => infinity,
type => supervisor,
modules => [wsSslMgrSup]
},
supervisor:start_child(eWSrv_sup, SslMgrSupSpec).
%% add a Udp listener
-spec openUdp(UdpName :: atom(), Port :: inet:port_number(), ListenOpts :: [listenOpt()]) -> {ok, pid()} | {error, term()}.
openUdp(UdpName, Port, ListenOpts) ->
UdpSrvSpec = #{
id => UdpName,
start => {wsUdpSrv, start_link, [UdpName, Port, ListenOpts]},
restart => permanent,
shutdown => 5000,
type => worker,
modules => [wsUdpSrv]
},
supervisor:start_child(eWSrv_sup, UdpSrvSpec).
%% add a Proxy protocol listener
-spec openPpt(ListenName :: atom(), Port :: inet:port_number(), ListenOpts :: [listenOpt()]) -> {ok, pid()} | {error, term()}.
openPpt(ListenName, Port, ListenOpts) ->
SslMgrSupSpec = #{
id => ListenName,
start => {wsPptMgrSup, start_link, [ListenName, Port, ListenOpts]},
restart => permanent,
shutdown => infinity,
type => supervisor,
modules => [wsPptMgrSup]
},
supervisor:start_child(eWSrv_sup, SslMgrSupSpec).
%% stop a listener
-spec close(atom()) -> ignore | {ok, pid()} | {error, term()}.
close(ListenName) ->
supervisor:terminate_child(eWSrv_sup, ListenName),
supervisor:delete_child(eWSrv_sup, ListenName).

+ 0
- 28
src/wsSrv/wsConSup.erl Wyświetl plik

@ -1,28 +0,0 @@
-module(wsConSup).
-behaviour(supervisor).
-export([
start_link/2
]).
-export([
init/1
]).
-spec(start_link(SupName :: atom(), WsMod :: module()) -> {ok, pid()}).
start_link(SupName, WsMod) ->
supervisor:start_link({local, SupName}, ?MODULE, WsMod).
init(WsMod) ->
SupFlags = #{strategy => simple_one_for_one, intensity => 100, period => 3600},
WsHttpSpec = #{
id => wsHttp,
start => {wsHttp, start_link, [WsMod]},
restart => transient,
shutdown => 3000,
type => worker,
modules => [wsHttp]
},
{ok, {SupFlags, [WsHttpSpec]}}.

+ 3
- 3
src/wsSrv/wsHttp.erl Wyświetl plik

@ -1,6 +1,6 @@
-module(wsHttp).
-include("eNet.hrl").
-include_lib("eNet/include/eNet.hrl").
-include("wsCom.hrl").
-export([
@ -24,8 +24,8 @@
, system_terminate/4
]).
newCon(_Sock, SupPid) ->
supervisor:start_link(SupPid, []).
newCon(_Sock, WsMod) ->
wsHttp:start_link(WsMod).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% genActor start %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-spec(start_link(atom()) -> {ok, pid()} | ignore | {error, term()}).

+ 0
- 4
src/wsSrv/wsUtil.erl Wyświetl plik

@ -9,7 +9,6 @@
, encode_range/2
, file_size/1
, gLV/3
, lsName/1
]).
-export_type([range/0]).
@ -80,6 +79,3 @@ gLV(Key, List, Default) ->
{Key, Value} ->
Value
end.
lsName(BaseName) ->
binary_to_atom(<<(atom_to_binary(BaseName))/binary, "Ls">>).

Napisz Podgląd
Ładowanie…
Anuluj
Zapisz