proxy

#!/usr/local/bin/escript
% -module(proxy).
% -compile(export_all).

%    +------------------------[ Company Network ]-------------------------+
%    |                                                                    |
%    |   <proxy -connect-left 127.0.0.1:22 -connect-right x.x.x.x:2222>   | <-------------------+
%    |                                |                                   |                     |
%    |                                |                                   |                     |
%    |           +------------+       |                                   |                     |
%    |           | ssh server | <-----+                                   |                     |
%    |           +------------+                                           |                     |
%    |                                                                    |                     |
%    +------------------------[ Company Network ]-------------------------+                     |
%                                                                                               |
%                                                                                               |
%                                                                                               |
%   +----------[ Home ]----------+      +--------------[ Public Network ]---------------+       |
%   |                            |      |                                               |       |
%   |                            |      |                                               |       |
%   | <ssh root@x.x.x.x -p 1222> | ---> | <proxy -listen-left 1222 -listen-right 2222>  | <-----+
%   |                            |      |                                               |
%   |                            |      |                                               |
%   +----------[ Home ]----------+      +--------------[ Public Network ]---------------+

-record(ip_port, {
	ip,
	port
}).

-record(state_for_inject_statistics, {
	connect_count = 0,
	sent_count = 0,
    lock_left_socket = void,
    force_close_connect_index = 0,
    force_close_sent_index = 0,
    force_lock_ip = void,
    force_stop_now = false
}).

main(Args) when Args == ["help"] orelse Args == ["-help"] orelse Args == ["--help"] orelse Args == ["-h"] orelse Args == ["--h"] orelse Args == ["/?"] ->
	io:format("usage:
        1: ./proxy -listen-left 13306 -listen-right 23306 -dump no|size|str
        2: ./proxy -connect-left 127.0.0.1:3306 -connect-right 127.0.0.1:23306 -dump no|size|str
        3: ./proxy -listen-left 33306 -connect-right 127.0.0.1:3306 -dump no|size|str
        4: ./proxy -listen-left 33306 -connect-right 127.0.0.1:3306 -inject-port 33305
    ");

% 这种模式的处理方案: 收到left socket连接后, 去连接right socket; 而后互相转发它俩; 传统正向代理
% ./proxy -listen-left 13306 -connect-right 127.0.0.1:3306
main(["-listen-left", LeftPort, "-connect-right", RightIpPort, "-dump", DoDump]) ->
    put(do_dump, DoDump),
    try
        %spawn(fun() -> observer:start() end),
        do_listen_left_then_connect_right(parse_int(LeftPort), make_ip_port_record(RightIpPort))
    catch
        throw:Err ->
            io:format("~s Error: ~p~n", [get_yymmddhhmmssmm_str(), Err]);
        exit:Err ->
            io:format("~s Error: ~p~n", [get_yymmddhhmmssmm_str(), Err]);
        error:Err ->
            case Err of
                {badmatch, {error, eaddrinuse}} ->
                    io:format("~s Error! Address in use: tcp://127.0.0.1:~s~n", [get_yymmddhhmmssmm_str(), LeftPort]);
                _ ->
                    io:format("~s Error: ~p~n", [get_yymmddhhmmssmm_str(), Err])
            end
    end;

% 这种模式的处理方案: 先连接right socket, 连上后, 等待right socket发来信号, 才去连left socket; 而后互相转发它俩; 实现反向代理的内网发起端
main(["-connect-left", LeftIpPort, "-connect-right", RightIpPort, "-dump", DoDump]) ->
    put(do_dump, DoDump),
    try
        do_connect_right_then_wait_then_connect_left(split_ip_and_port(LeftIpPort), make_ip_port_map(RightIpPort))
    catch
        throw:Err ->
            io:format("~s Error: ~p~n", [get_yymmddhhmmssmm_str(), Err]);
        exit:Err ->
            io:format("~s Error: ~p~n", [get_yymmddhhmmssmm_str(), Err]);
        error:Err ->
            io:format("~s Error: ~p~n", [get_yymmddhhmmssmm_str(), Err])
    end;

% 这种模式的处理方案: right socket 要先被连上, 然后 left socket 再被客户端连接, 然后互相转发它俩;
% ./proxy -listen-left 2222 -listen-right 2223
main(["-listen-left", LeftPort, "-listen-right", RightPort, "-dump", DoDump]) ->
    put(do_dump, DoDump),
    try
        {ok, LeftListenSock} = tcp_listen(parse_int(LeftPort)),
        {ok, RightListenSock} = tcp_listen(parse_int(RightPort)),
        io:format("~s Start proxy listen tcp://127.0.0.1:~s and listen tcp://127.0.0.1:~s~n", [get_yymmddhhmmssmm_str(), LeftPort, RightPort]),
        wait_right_connected_then_wait_left_connected(LeftListenSock, RightListenSock)
    catch
        throw:Err ->
            io:format("~s Error: ~p~n", [get_yymmddhhmmssmm_str(), Err]);
        exit:Err ->
            io:format("~s Error: ~p~n", [get_yymmddhhmmssmm_str(), Err]);
        error:Err ->
            case Err of
                {badmatch, {error, eaddrinuse}} ->
                    io:format("~s Error! Address in use: tcp://127.0.0.1:~s~n", [get_yymmddhhmmssmm_str(), LeftPort]),
                    io:format("~s Error! Address in use: tcp://127.0.0.1:~s~n", [get_yymmddhhmmssmm_str(), RightPort]);
                _ ->
                    io:format("~s Error: ~p~n", [get_yymmddhhmmssmm_str(), Err])
            end
    end;

% 在第一种模式的基础上, 也监听inject-port, 用于控制当前运行的这套proxy要不要特意断路。 一般用于做Fault Injection。一种软件测试的技术，用于模拟和注入系统中的故障和错误。通过有意地引入故障和错误情况，测试人员可以评估系统的稳定性、可靠性和安全性。
main(["-listen-left", LeftPort, "-connect-right", RightIpPort, "-inject-port", InjectPort]) ->
    InjectPid = spawn_link(fun() -> inject_handle(InjectPort) end),
    main(["-listen-left", LeftPort, "-connect-right", RightIpPort, "-dump", InjectPid]);

main(_) ->
	io:format("param error, use proxy --help ~n").

do_listen_left_then_connect_right(LeftPort, RightIpPortRcd) ->
    {ok, LeftListenSock} = tcp_listen(LeftPort),
    io:format("~s Start proxy tcp://127.0.0.1:~p to tcp://~s:~p~n", [get_yymmddhhmmssmm_str(), LeftPort, RightIpPortRcd#ip_port.ip, RightIpPortRcd#ip_port.port]),
    wait_a_new_left_socket(LeftListenSock, RightIpPortRcd#ip_port.ip, RightIpPortRcd#ip_port.port).

% do_connect_right_then_wait_then_connect_left([LeftIp, LeftPort], #{ip := RightIp, port := RightPort} = RightIpPortMap) ->
do_connect_right_then_wait_then_connect_left([LeftIp, LeftPort], RightIpPortMap) ->
    RightIp = maps:get(ip, RightIpPortMap),
    RightPort = maps:get(port, RightIpPortMap),
    case tcp_connect(RightIp, RightPort) of
        {ok, RightSocket} ->
            io:format("~s Proxy connected on a new right socket: ~p~n", [get_yymmddhhmmssmm_str(), RightSocket]),
            wait_right_socket_initial_signal(RightSocket, {LeftIp, LeftPort}, RightIpPortMap);
        {error, can_not_connect} ->
            io:format("~s Error! Can not connect right socket: tcp://~s:~s~n", [get_yymmddhhmmssmm_str(), RightIp, RightPort]),
            timer:sleep(1000),
            do_connect_right_then_wait_then_connect_left([LeftIp, LeftPort], RightIpPortMap)
    end.

wait_right_socket_initial_signal(RightSocket, {LeftIp, LeftPort}, RightIpPortMap) ->
    receive
        {tcp, RightSocket, <<1:8>>} -> % 收到 1byte 1, 则链接左边
            io:format("~s got <<1:8>> ~p~n", [get_yymmddhhmmssmm_str(), RightSocket]),
            case tcp_connect(LeftIp, LeftPort) of
                {ok, LeftSocket} ->
                    io:format("~s Proxy connected on a new left socket: ~p~n", [get_yymmddhhmmssmm_str(), LeftSocket]),
                    gen_tcp:send(RightSocket, <<1:8>>), % 发送 1byte 1, 通知那头可以开始转发数据
                    bind_left_and_right(LeftSocket, RightSocket),
                    do_connect_right_then_wait_then_connect_left([LeftIp, LeftPort], RightIpPortMap);
                {error, can_not_connect} ->
                    io:format("~s Error! Can not connect left socket: tcp://~s:~p~n", [get_yymmddhhmmssmm_str(), LeftIp, LeftPort]),
                    gen_tcp:send(RightSocket, <<0:8>>), % 发送 1byte 0, 通知那头无法连接
                    wait_right_socket_initial_signal(RightSocket, {LeftIp, LeftPort}, RightIpPortMap)
            end;
        {tcp, RightSocket, <<0:8>>} -> % 收到 1byte 0, 心跳而已, 要不要回?
            %io:format("got <<0:8>> ~p~n", [RightSocket]),
            gen_tcp:send(RightSocket, <<0:8>>), % 发送 1byte 0, 回一下心跳
            wait_right_socket_initial_signal(RightSocket, {LeftIp, LeftPort}, RightIpPortMap);
        {tcp, RightSocket, Bin} ->
            io:format("~s Error! Right socket is not our proxy, ~p got message: ~p~n", [get_yymmddhhmmssmm_str(), RightSocket, Bin]),
            gen_tcp:close(RightSocket),
            timer:sleep(3000),
            do_connect_right_then_wait_then_connect_left([LeftIp, LeftPort], RightIpPortMap);
        {tcp_closed, RightSocket} ->
            io:format("~s Right socket disconnected: ~p~n", [get_yymmddhhmmssmm_str(), RightSocket]),
            gen_tcp:close(RightSocket),
            %timer:sleep(3000),
            do_connect_right_then_wait_then_connect_left([LeftIp, LeftPort], RightIpPortMap)
    after 60000 -> % 每60秒无响应则断开重连
        io:format("~s Close right socket after 60 seconds: ~p~n", [get_yymmddhhmmssmm_str(), RightSocket]),
        gen_tcp:close(RightSocket),
        do_connect_right_then_wait_then_connect_left([LeftIp, LeftPort], RightIpPortMap)
    end.
    
wait_a_new_left_socket(LeftListenSock, RightIp, RightPort) ->
    case wait_until_socket_connected(LeftListenSock) of
        {ok, LeftSocket} ->
            case inet:peername(LeftSocket) of
                {ok, {LeftClientIp, LeftClientPort}} ->
                    my_echo(io_lib:format("~s A new left socket connected: ~p ~p:~p~n", [get_yymmddhhmmssmm_str(), LeftSocket, inet:ntoa(LeftClientIp), LeftClientPort])),
                    Blocked = check_blocked(LeftClientIp),

                    case Blocked of 
                        true ->
                            my_echo(io_lib:format("~s Close left socket: ~p cause blocked temporarily!~n", [get_yymmddhhmmssmm_str(), LeftSocket])),
                            gen_tcp:close(LeftSocket),
                            c:flush();
                        false ->
                            case tcp_connect(RightIp, RightPort) of
                                {ok, RightSocket} ->
                                    my_echo(io_lib:format("~s A new right socket connected: ~p~n", [get_yymmddhhmmssmm_str(), RightSocket])),
                                    bind_left_and_right(LeftSocket, RightSocket);
                                {error, can_not_connect} ->
                                    my_echo(io_lib:format("~s Error! Can not connect right socket: tcp://~s:~p~n", [get_yymmddhhmmssmm_str(), RightIp, RightPort])),
                                    my_echo(io_lib:format("~s Close left socket: ~p~n", [get_yymmddhhmmssmm_str(), LeftSocket])),
                                    gen_tcp:close(LeftSocket),
                                    c:flush()
                            end
                    end;
                W ->
                    io:format("~s Opps! left socket peername return error! ~p~n", [get_yymmddhhmmssmm_str(), W]),
                    gen_tcp:close(LeftSocket),
                    c:flush()
            end;
        O ->
            io:format("~s Opps! left socket error! ~p~n", [get_yymmddhhmmssmm_str(), O])
    end,
    wait_a_new_left_socket(LeftListenSock, RightIp, RightPort).

check_blocked(LeftClientIp) ->
    DoDump = get(do_dump),
    case is_pid(DoDump) of
        true ->
            InjectPid = DoDump,
            InjectPid ! {check_is_lock_ip, self(), inet:ntoa(LeftClientIp)},
            receive 
                blocked ->
                    true;
                allow ->
                    false
            end;
        _ ->
            false
    end.

wait_right_connected_then_wait_left_connected(LeftListenSock, RightListenSock) ->
    case wait_until_socket_connected(RightListenSock) of
        {ok, RightSocket} ->
            case inet:peername(RightSocket) of
                {ok, {RightClientIp, RightClientPort}} ->
                    io:format("~s A new right socket connected: ~p ~p:~p~n", [get_yymmddhhmmssmm_str(), RightSocket, inet:ntoa(RightClientIp), RightClientPort]),
                    wait_left_connected(LeftListenSock, RightListenSock, RightSocket);
                {error, Reason} ->
                    io:format("~s Opps! A right socket leaved! {error, ~p}~n", [get_yymmddhhmmssmm_str(), Reason]),
                    gen_tcp:close(RightSocket),
                    c:flush(),
                    wait_right_connected_then_wait_left_connected(LeftListenSock, RightListenSock)
            end;
        O ->
            io:format("~s Opps! A right socket leaved! ~p~n", [get_yymmddhhmmssmm_str(), O]),
            wait_right_connected_then_wait_left_connected(LeftListenSock, RightListenSock)
    end.

wait_left_connected(LeftListenSock, RightListenSock, RightSocket) ->
    case wait_until_socket_connected(LeftListenSock, 5000) of
        {error, timeout} ->
            gen_tcp:send(RightSocket, <<0:8>>), % 发送 1byte 0, 心跳那头
            receive
                {tcp, RightSocket, <<0:8>>} -> % 收到 1byte 0, 心跳的回复
                    wait_left_connected(LeftListenSock, RightListenSock, RightSocket);
                {tcp_closed, RightSocket} ->
                    io:format("~s Error! Right socket closed! ~p ~n", [get_yymmddhhmmssmm_str(), RightSocket]),
                    gen_tcp:close(RightSocket),
                    c:flush(),
                    wait_right_connected_then_wait_left_connected(LeftListenSock, RightListenSock)
            after 3000 ->
                io:format("~s Heartbeat Error! Right socket not response! ~p ~n", [get_yymmddhhmmssmm_str(), RightSocket]),
                gen_tcp:close(RightSocket),
                c:flush(),
                wait_right_connected_then_wait_left_connected(LeftListenSock, RightListenSock)
            end;
        {ok, LeftSocket} ->
            case inet:peername(LeftSocket) of
                {ok, {LeftClientIp, LeftClientPort}} ->
                    io:format("~s A new left socket connected: ~p ~p:~p~n", [get_yymmddhhmmssmm_str(), LeftSocket, inet:ntoa(LeftClientIp), LeftClientPort]),
                    gen_tcp:send(RightSocket, <<1:8>>), % 发送 1byte 1, 通知那头可以开始连left
                    receive
                        {tcp, RightSocket, <<0:8>>} -> % 收到 1byte 0, 那头无法连接左边, 我也断开左边
                            % io:format("got <<0:8>> ~p~n", [RightSocket]),
                            gen_tcp:close(LeftSocket),
                            wait_left_connected(LeftListenSock, RightListenSock, RightSocket);
                        {tcp, RightSocket, <<1:8>>} -> % 收到 1byte 1, 可以开始转发数据
                            bind_left_and_right(LeftSocket, RightSocket),
                            c:flush(),
                            wait_right_connected_then_wait_left_connected(LeftListenSock, RightListenSock)
                    after 5000 ->
                        io:format("~s Error! Right socket not response! Maybe Change left and right! ~p ~n", [get_yymmddhhmmssmm_str(), RightSocket]),
                        gen_tcp:close(LeftSocket),
                        gen_tcp:close(RightSocket),
                        c:flush(),
                        wait_right_connected_then_wait_left_connected(LeftListenSock, RightListenSock)
                    end;
                {error, Reason} ->
                    io:format("~s Opps! A left socket leaved! {error, ~p}~n", [get_yymmddhhmmssmm_str(), Reason]),
                    gen_tcp:close(RightSocket),
                    c:flush(),
                    wait_right_connected_then_wait_left_connected(LeftListenSock, RightListenSock)
            end
    end.

bind_left_and_right(LeftSocket, RightSocket) ->
    DoDump = get(do_dump),
    my_echo(io_lib:format("~s Success! Now proxy two sockets! ~p <-> ~p ~n", [get_yymmddhhmmssmm_str(), LeftSocket, RightSocket])),
    connect_count_for_inject_plus(DoDump, LeftSocket),
    Pid = spawn_link(fun() -> loop(LeftSocket, RightSocket, DoDump) end),
    gen_tcp:controlling_process(LeftSocket, Pid),
    gen_tcp:controlling_process(RightSocket, Pid).

my_echo(Msg) ->
    DoDump = get(do_dump),
    my_echo(Msg, DoDump).

my_echo(Msg, DoDump) ->
    io:format("~s", [Msg]),
    case is_pid(DoDump) of
        true ->
            InjectPid = DoDump,
            InjectPid ! {connecting_message, Msg};
        _ ->
            void
    end.

connect_count_for_inject_plus(DoDump, LeftSocket) ->
    case is_pid(DoDump) of
        true ->
            InjectPid = DoDump,
            InjectPid ! {connect_count_for_inject_plus, LeftSocket};
        _ ->
            void
    end.


loop(LeftSocket, RightSocket, DoDump) ->
    receive
        {tcp, LeftSocket, Bin} ->
            case DoDump of
                "no" ->
                    void;
                "size" ->
                    io:format("~s ~p -----> ~p ->~n", [get_yymmddhhmmssmm_str(), LeftSocket, size(Bin)]);
                "str" ->
                    PprintBin = iterate_binary(Bin),
                    io:format("~s ~p --[~p]--> ```~s``` ->~n", [get_yymmddhhmmssmm_str(), LeftSocket, size(Bin), PprintBin]);
                InjectPid when is_pid(InjectPid) ->
                    InjectPid ! {sent_count_for_inject_plus, LeftSocket, self()},
                    PprintBin = iterate_binary(Bin),
                    T = get_yymmddhhmmssmm_str(),
                    receive 
                        force_stop ->
                            InjectPid ! {inject_message, 
                                io_lib:format("~s ~p -----> ~p ->X~n", [T, LeftSocket, size(Bin)]),
                                io_lib:format("~s ~p --[~p]--> ```~s``` ->X~n", [T, LeftSocket, size(Bin), PprintBin])
                            },
                            ForceStopMsg = io_lib:format("~s      -----!!!!!{force_stop! So close left socket ~p}!!!!!----- ~n", [T, LeftSocket]),
                            io:format("~s", [ForceStopMsg]),
                            InjectPid ! {connecting_message, ForceStopMsg},
                            InjectPid ! {inject_message, ForceStopMsg, ForceStopMsg},
                            gen_tcp:close(LeftSocket),
                            gen_tcp:close(RightSocket),
                            exit(normal);
                        allow ->
                            InjectPid ! {inject_message, 
                                io_lib:format("~s ~p -----> ~p ->~n", [T, LeftSocket, size(Bin)]),
                                io_lib:format("~s ~p --[~p]--> ```~s``` ->~n", [T, LeftSocket, size(Bin), PprintBin])
                            }
                    end;
                _ ->
                    void
            end,
            
            %gen_tcp:send(LeftSocket, <<"yoyo~", Bin/big-binary>>),
            case gen_tcp:send(RightSocket, Bin) of
                ok ->
                    loop(LeftSocket, RightSocket, DoDump);
                {error, Reason} ->
                    io:format("~s                   {error, ~p} <----- ~p~n", [get_yymmddhhmmssmm_str(), Reason, RightSocket]),
                    gen_tcp:close(LeftSocket)
            end;
        {tcp, RightSocket, Bin} ->
            case DoDump of
                "no" ->
                    void;
                "size" ->
                    io:format("~s                <- ~p <----- ~p~n", [get_yymmddhhmmssmm_str(), size(Bin), RightSocket]);
                "str" ->
                    PprintBin = iterate_binary(Bin),
                    io:format("~s                <- ```~s``` <--[~p]-- ~p~n", [get_yymmddhhmmssmm_str(), PprintBin, size(Bin), RightSocket]);
                InjectPid when is_pid(InjectPid) ->
                    PprintBin = iterate_binary(Bin),
                    T = get_yymmddhhmmssmm_str(),
                    InjectPid ! {inject_message, 
                        io_lib:format("~s                <- ~p <----- ~p~n", [T, size(Bin), RightSocket]), 
                        io_lib:format("~s                <- ```~s``` <--[~p]-- ~p~n", [T, PprintBin, size(Bin), RightSocket])
                    };
                _ ->
                    void
            end,
            
            case gen_tcp:send(LeftSocket, Bin) of
                ok ->
                    loop(LeftSocket, RightSocket, DoDump);
                {error, Reason} ->
                    io:format("~s ~p -----> {error, ~p}~n", [get_yymmddhhmmssmm_str(), LeftSocket, Reason]),
                    gen_tcp:close(RightSocket)
            end;
        {tcp_closed, LeftSocket} ->
            my_echo(io_lib:format("~s ~p -----> disconnect -----> ~p~n", [get_yymmddhhmmssmm_str(), LeftSocket, RightSocket]), DoDump),
            gen_tcp:close(RightSocket);
        {tcp_closed, RightSocket} ->
            my_echo(io_lib:format("~s ~p <----- disconnect <----- ~p~n", [get_yymmddhhmmssmm_str(), LeftSocket, RightSocket]), DoDump),
            gen_tcp:close(LeftSocket);
        {tcp_error, LeftSocket, Reason} ->
            my_echo(io_lib:format("~s ~p ~p -----> disconnect -----> ~p~n", [get_yymmddhhmmssmm_str(), Reason, LeftSocket, RightSocket]), DoDump),
            gen_tcp:close(LeftSocket),
            gen_tcp:close(RightSocket);
        {tcp_error, RightSocket, Reason} ->
            my_echo(io_lib:format("~s ~p ~p <----- disconnect <----- ~p~n", [get_yymmddhhmmssmm_str(), Reason, LeftSocket, RightSocket]), DoDump),
            gen_tcp:close(RightSocket),
            gen_tcp:close(LeftSocket)
    after 125000 -> % 125秒双方无通信则断开双方
        my_echo(io_lib:format("~s Close both socket after 125 seconds no communicate: ~p and ~p~n", [get_yymmddhhmmssmm_str(), LeftSocket, RightSocket]), DoDump),
        gen_tcp:close(LeftSocket),
        gen_tcp:close(RightSocket)
    end.

inject_handle(InjectPort) ->
    InjectPid = self(),
    spawn_link(fun() -> inject_listen(InjectPort, InjectPid) end),
    loop_inject_handle_message(InjectPid, false, [], [], [], void, 0, #state_for_inject_statistics{}).

loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics) ->
    receive
        {connect_count_for_inject_plus, LeftSocket} ->
            case DoRecord of 
                true ->
                    case StateForInjectStatistics#state_for_inject_statistics.lock_left_socket =:= void of
                        true ->
                            case StateForInjectStatistics#state_for_inject_statistics.connect_count + 1 =:= StateForInjectStatistics#state_for_inject_statistics.force_close_connect_index of
                                true ->
                                    loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics#state_for_inject_statistics{connect_count = StateForInjectStatistics#state_for_inject_statistics.connect_count + 1, lock_left_socket = LeftSocket});
                                false ->
                                    loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics#state_for_inject_statistics{connect_count = StateForInjectStatistics#state_for_inject_statistics.connect_count + 1})
                            end;
                        false ->
                            loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics)
                    end;
                _ ->
                    loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics)
            end;
        {sent_count_for_inject_plus, LeftSocket, BindPid} ->
            case DoRecord of 
                true ->
                    case StateForInjectStatistics#state_for_inject_statistics.lock_left_socket =:= LeftSocket of
                        true ->
                            case StateForInjectStatistics#state_for_inject_statistics.sent_count + 1 =:= StateForInjectStatistics#state_for_inject_statistics.force_close_sent_index of
                                true ->
                                    BindPid ! force_stop,
                                    loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics#state_for_inject_statistics{force_stop_now = true, sent_count = StateForInjectStatistics#state_for_inject_statistics.sent_count + 1});
                                false ->
                                    BindPid ! allow,
                                    loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics#state_for_inject_statistics{sent_count = StateForInjectStatistics#state_for_inject_statistics.sent_count + 1})
                            end;
                        false ->
                            BindPid ! allow,
                            loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics)
                    end;
                _ ->
                    BindPid ! allow,
                    loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics)
            end;
        {connecting_message, ConnectingLog} ->
            case DoRecord of 
                true ->
                    loop_inject_handle_message(InjectPid, DoRecord, [ConnectingLog | ConnectingLogArr], SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics);
                _ ->
                    loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics)
            end;
        {inject_message, SizeLog, DetailLog} ->
            case DoRecord of 
                true ->
                    loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, [SizeLog | SizeLogArr], [DetailLog | DetailLogArr], InjectSocketControllPid, Timer, StateForInjectStatistics);
                _ ->
                    loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics)
            end;
        {force_close, Ms, ConnectionIndex, SentFinishIndex, LockIp, NewInjectSocketControllPid} ->
            case InjectSocketControllPid of
                void ->
                    NewTimer = erlang:send_after(Ms, self(), {record_time_end});
                _ ->
                    erlang:cancel_timer(Timer),
                    NewTimer = erlang:send_after(Ms, self(), {record_time_end}),
                    void
            end,
            loop_inject_handle_message(InjectPid, true, [], [], [], NewInjectSocketControllPid, NewTimer, #state_for_inject_statistics{
                force_close_connect_index = ConnectionIndex,
                force_close_sent_index = SentFinishIndex,
                force_lock_ip = LockIp
            });
        {check_is_lock_ip, ListenPid, NewIP} ->
            case StateForInjectStatistics#state_for_inject_statistics.force_stop_now =:= true andalso StateForInjectStatistics#state_for_inject_statistics.force_lock_ip =:= NewIP of
                true ->
                    ListenPid ! blocked;
                false ->
                    ListenPid ! allow
            end,
            loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics);
        {start_record, Ms, NewInjectSocketControllPid} ->
            case InjectSocketControllPid of
                void ->
                    NewTimer = erlang:send_after(Ms, self(), {record_time_end});
                _ ->
                    erlang:cancel_timer(Timer),
                    NewTimer = erlang:send_after(Ms, self(), {record_time_end}),
                    void
            end,
            loop_inject_handle_message(InjectPid, true, [], [], [], NewInjectSocketControllPid, NewTimer, StateForInjectStatistics);
        {inject_socket_controll_pid_leave, InjectSocketControllPid} ->
            erlang:cancel_timer(Timer),
            io:format("cancel timer~n"),
            loop_inject_handle_message(InjectPid, false, [], [], [], void, 0, #state_for_inject_statistics{});
        {record_time_end} ->
            ConnectingLogBin = iolist_to_binary(lists:reverse(ConnectingLogArr)),
            SizeLogBin = iolist_to_binary(lists:reverse(SizeLogArr)),
            DetailLogBin = iolist_to_binary(lists:reverse(DetailLogArr)),
            Echo = <<"---Size Info Below---\r\n", ConnectingLogBin/binary, SizeLogBin/binary, "---Detail Below---\r\n", ConnectingLogBin/binary, DetailLogBin/binary>>,
            ContentLength = integer_to_binary(size(Echo)),
            InjectSocketControllPid ! {result, <<"HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Type: text/html\r\nContent-Length: ", ContentLength/binary, "\r\nServer: sss0.1\r\n\r\n", Echo/binary, "\r\n">>},
            io:format("~s Inject Pid ~p finish record and send result to ~p ~n", [get_yymmddhhmmssmm_str(), InjectPid, InjectSocketControllPid]),
            io:format("~p~n", [StateForInjectStatistics]),
            loop_inject_handle_message(InjectPid, false, [], [], [], void, 0, #state_for_inject_statistics{});
        {'DOWN', _Ref, process, PidWorker, Reason} ->
            io:format("~s Inject Pid ~p received DOWN Message: ~p ~p ~n", [get_yymmddhhmmssmm_str(), InjectPid, PidWorker, Reason]),
            loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics);
        Other ->
            io:format("~s Inject Pid ~p received: ~p ~n", [get_yymmddhhmmssmm_str(), InjectPid, Other]),
            loop_inject_handle_message(InjectPid, DoRecord, ConnectingLogArr, SizeLogArr, DetailLogArr, InjectSocketControllPid, Timer, StateForInjectStatistics)
    end.

inject_listen(InjectPort, InjectPid) ->
    try
        {ok, InjectListenSock} = tcp_listen(parse_int(InjectPort)),
        io:format("~s Inject Pid ~p listen on tcp://127.0.0.1:~p~n", [get_yymmddhhmmssmm_str(), InjectPid, parse_int(InjectPort)]),
        InjectListenProcessPid = self(),
        wait_a_new_inject_socket(InjectListenSock, InjectPid, InjectListenProcessPid, 0)
    catch
        throw:Err ->
            io:format("~s Inject Error: ~p~n", [get_yymmddhhmmssmm_str(), Err]);
        exit:Err ->
            io:format("~s Inject Error: ~p~n", [get_yymmddhhmmssmm_str(), Err]);
        error:Err ->
            case Err of
                {badmatch, {error, eaddrinuse}} ->
                    io:format("~s Inject Error! Address in use: tcp://127.0.0.1:~s~n", [get_yymmddhhmmssmm_str(), InjectPort]);
                _ ->
                    io:format("~s Inject Error: ~p~n", [get_yymmddhhmmssmm_str(), Err])
            end
    end.

wait_a_new_inject_socket(InjectListenSock, InjectPid, InjectListenProcessPid, CurrentInjectCount) ->
    case wait_until_socket_connected(InjectListenSock, 3000) of
        {error, timeout} ->
            receive 
                {last_http_finish} ->
                    CurrentInjectCount2 = 0;
                {'DOWN', _Ref, process, PidWorker, Reason} ->
                    CurrentInjectCount2 = 0,
                    io:format("Worker2 process exited with reason: ~p ~p~n", [PidWorker, Reason]);
                Other ->
                    CurrentInjectCount2 = 0,
                    io:format("~s Inject listen process ~p received: ~p ~n", [get_yymmddhhmmssmm_str(), InjectListenProcessPid, Other])
            after 0 ->
                CurrentInjectCount2 = CurrentInjectCount,
                void
            end;
        {ok, InjectSocket} ->
            case CurrentInjectCount of 
                0 ->
                    case inet:peername(InjectSocket) of
                        {ok, {LeftClientIp, LeftClientPort}} ->
                            Pid = spawn_link(fun() -> loop_inject_socket(InjectSocket, InjectPid, InjectListenProcessPid) end),
                            io:format("~s A new Inject socket ~p connected: ~p:~p, give it to ~p~n", [get_yymmddhhmmssmm_str(), InjectSocket, inet:ntoa(LeftClientIp), LeftClientPort, Pid]),
                            gen_tcp:controlling_process(InjectSocket, Pid),
                            CurrentInjectCount2 = 1;
                        W ->
                            CurrentInjectCount2 = 0,
                            io:format("~s Opps! Inject socket ~p peername return error! ~p~n", [get_yymmddhhmmssmm_str(), InjectSocket, W]),
                            gen_tcp:close(InjectSocket)
                    end;
                1 ->
                    CurrentInjectCount2 = 1,
                    Pid = spawn(fun() -> 
                        receive 
                            _Any -> 
                                % io:format("~p", [Any]),
                                void
                            after 3000 ->
                                void
                        end,
                        case gen_tcp:send(InjectSocket, <<"HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: 20\r\nServer: sss0.1\r\n\r\n{\"status\":\"working\"}\r\n">>) of
                            ok ->
                                % io:format("~s ~p -----> {send close done}~n", [get_yymmddhhmmssmm_str(), InjectSocket]),
                                void;
                            {error, _Reason} ->
                                % io:format("~s ~p -----> {error, ~p}~n", [get_yymmddhhmmssmm_str(), InjectSocket, Reason])
                                void
                        end,
                        gen_tcp:close(InjectSocket)    
                    end),
                    gen_tcp:controlling_process(InjectSocket, Pid)
            end;
        O ->
            CurrentInjectCount2 = 0,
            io:format("~s Opps! Inject Pid ~p gen_tcp:accept got error! ~p~n", [get_yymmddhhmmssmm_str(), InjectPid, O])
    end,
    wait_a_new_inject_socket(InjectListenSock, InjectPid, InjectListenProcessPid, CurrentInjectCount2).

loop_inject_socket(InjectSocket, InjectPid, InjectListenProcessPid) ->
    receive
        {result, BinResult} ->
            gen_tcp:send(InjectSocket, BinResult),
            InjectListenProcessPid ! {last_http_finish},
            gen_tcp:close(InjectSocket);
        {tcp, InjectSocket, Bin} ->
            Arr = binary:split(Bin, <<"|">>, [global]),
            case erlang:length(Arr) of
                4 -> % <<"http header|start_record|300|\r\n", Res/binary>> ->
                    [_Bin_HttpHeader, Bin_Action, TimerMs, _Bin_HttpRes] = Arr,
                    case Bin_Action of
                        <<"start_record">> ->
                            Ms = binary_to_integer2(TimerMs),
                            case Ms > 1 of
                                true ->
                                    InjectPid ! {start_record, Ms, self()},
                                    loop_inject_socket(InjectSocket, InjectPid, InjectListenProcessPid);
                                false ->
                                    gen_tcp:send(InjectSocket, <<"HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: 24\r\nServer: sss0.1\r\n\r\n{\"status\":\"not_support\"}\r\n">>),
                                    InjectListenProcessPid ! {last_http_finish},
                                    gen_tcp:close(InjectSocket)
                            end;
                        _ ->
                            gen_tcp:send(InjectSocket, <<"HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: 24\r\nServer: sss0.1\r\n\r\n{\"status\":\"not_support\"}\r\n">>),
                            InjectListenProcessPid ! {last_http_finish},
                            gen_tcp:close(InjectSocket)
                    end;
                7 -> % <<"http header|force_close|300|1|2|192.168.1.5|\r\n", Res/binary>> ->
                    [_Bin_HttpHeader, Bin_Action, TimerMs, ConnectionIndex, SentFinishIndex, LockIp, _Bin_HttpRes] = Arr,
                    case Bin_Action of
                        <<"force_close">> ->
                            Ms = binary_to_integer2(TimerMs),
                            ConnectionIndex2 = binary_to_integer2(ConnectionIndex),
                            SentFinishIndex2 = binary_to_integer2(SentFinishIndex),
                            LockIp2 = binary_to_list2(LockIp),
                            case Ms > 1 andalso ConnectionIndex2 > 0 andalso SentFinishIndex2 > 0 of
                                true ->
                                    InjectPid ! {force_close, Ms, ConnectionIndex2, SentFinishIndex2, LockIp2, self()},
                                    loop_inject_socket(InjectSocket, InjectPid, InjectListenProcessPid);
                                false ->
                                    gen_tcp:send(InjectSocket, <<"HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: 24\r\nServer: sss0.1\r\n\r\n{\"status\":\"not_support\"}\r\n">>),
                                    InjectListenProcessPid ! {last_http_finish},
                                    gen_tcp:close(InjectSocket)
                            end;
                        _ ->
                            gen_tcp:send(InjectSocket, <<"HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: 24\r\nServer: sss0.1\r\n\r\n{\"status\":\"not_support\"}\r\n">>),
                            InjectListenProcessPid ! {last_http_finish},
                            gen_tcp:close(InjectSocket)
                    end;
                _ ->
                    io:format("~s Inject socket ~p Got-----> ~p ~n", [get_yymmddhhmmssmm_str(), InjectSocket, Bin]),
                    gen_tcp:send(InjectSocket, <<"HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: 24\r\nServer: sss0.1\r\n\r\n{\"status\":\"not_support\"}\r\n">>),
                    % loop_inject_socket(InjectSocket, InjectPid, InjectListenProcessPid)
                    InjectListenProcessPid ! {last_http_finish},
                    gen_tcp:close(InjectSocket)
            end;
        {tcp_closed, InjectSocket} ->
            io:format("~s Inject socket ~p-----> disconnect~n", [get_yymmddhhmmssmm_str(), InjectSocket]),
            InjectListenProcessPid ! {last_http_finish},
            InjectPid ! {inject_socket_controll_pid_leave, self()},
            gen_tcp:close(InjectSocket);
        {tcp_error, InjectSocket, Reason} ->
            io:format("~s Inject socket ~p ~p -----> disconnect~n", [get_yymmddhhmmssmm_str(), Reason, InjectSocket]),
            InjectListenProcessPid ! {last_http_finish},
            InjectPid ! {inject_socket_controll_pid_leave, self()},
            gen_tcp:close(InjectSocket)
    after 120000 -> % 写死了120秒超时
        io:format("~s Close inject socket ~p after 120 seconds no communicate ~n", [get_yymmddhhmmssmm_str(), InjectSocket]),
        InjectListenProcessPid ! {last_http_finish},
        InjectPid ! {inject_socket_controll_pid_leave, self()},
        gen_tcp:close(InjectSocket)
    end.

split_ip_and_port(IpPort) -> % 127.0.0.1:3306
    [
        string:substr(IpPort, 1, string:rstr(IpPort, ":") - 1), % 127.0.0.1
        parse_int(string:substr(IpPort, string:rstr(IpPort, ":") + 1)) % 3306
    ].

make_ip_port_record(IpPort) -> % 127.0.0.1:3306
    #ip_port{
        ip = string:substr(IpPort, 1, string:rstr(IpPort, ":") - 1), % 127.0.0.1
        port = parse_int(string:substr(IpPort, string:rstr(IpPort, ":") + 1)) % 3306
    }.

make_ip_port_map(IpPort) -> % 127.0.0.1:3306
    #{
        ip => string:substr(IpPort, 1, string:rstr(IpPort, ":") - 1), % 127.0.0.1
        port => parse_int(string:substr(IpPort, string:rstr(IpPort, ":") + 1)) % 3306
    }.
    
parse_int(String) ->
    erlang:list_to_integer(String).

binary_to_integer2(Bin) ->
    try
        erlang:binary_to_integer(Bin)
    catch
        throw:Err ->
            io:format("~s throw Error: binary_to_integer ~p~n", [get_yymmddhhmmssmm_str(), Err]),
            -1;
        exit:Err ->
            io:format("~s exit Error: binary_to_integer ~p~n", [get_yymmddhhmmssmm_str(), Err]),
            -1;
        error:Err ->
            io:format("~s Error: binary_to_integer ~p~n", [get_yymmddhhmmssmm_str(), Err]),
            -1
    end.

binary_to_list2(Bin) ->
    case Bin of
        <<"">> -> void;
        _ -> binary_to_list(Bin)
    end.

wait_until_socket_connected(Listen) ->
    gen_tcp:accept(Listen).
    
wait_until_socket_connected(Listen, Timeout) ->
    gen_tcp:accept(Listen, Timeout).

tcp_listen(Port) ->
    gen_tcp:listen(Port, [binary, {packet, 0}, {reuseaddr, true}, {active, true}]).
    
tcp_connect(Ip, Port) ->
    tcp_connect(Ip, Port, 1).

tcp_connect(_, _, Count) when Count > 2 ->
    {error, can_not_connect};
tcp_connect(Ip, Port, Count) ->
    case gen_tcp:connect(Ip, Port, [binary, {packet, 0}]) of
        {ok, Socket} ->
            {ok, Socket};
        _ ->
            timer:sleep(1000),
            tcp_connect(Ip, Port, Count + 1)
    end.

get_yymmddhhmmssmm_str() ->
    NowMillisecond = erlang:system_time(millisecond), % 1610803593978
    A = NowMillisecond div 1000000000, % 1610
    B = (NowMillisecond div 1000) rem (A * 1000000), % 803593
    C = (NowMillisecond rem ((NowMillisecond div 1000) * 1000)), % 978
	{{Year, Month, Day}, {Hour, Minute, Second}} = calendar:now_to_local_time({A, B, C}),
	lists:flatten(io_lib:format("~4..0w-~2..0w-~2..0w ~2..0w:~2..0w:~2..0w.~3..0w", [Year, Month, Day, Hour, Minute, Second, C])).  % "2021-01-16 21:35:15.321"

iterate_binary(Bin) ->
    iterate_binary(Bin, []).

iterate_binary(<<>>, Acc) -> io_lib:format("~s", [lists:reverse(Acc)]);
iterate_binary(<<Byte:8, Rest/binary>>, Acc) ->
    % case Byte < 32 orelse Byte > 126 of
    case Byte < 32 of
        true -> 
            Result = io_lib:format("<<~p>>", [Byte]);
        false -> 
            Result = io_lib:format("~s", [binary_to_list(<<Byte>>)])
    end,
    iterate_binary(Rest, [Result | Acc]).

% ssh -fNg -L 3307:localhost:3307 root@111.231.76.107 -p
% ssh -f -N -T -R 2222:localhost:22 sourceuser@138.47.99.99

    %-f: tells the SSH to background itself after it authenticates, saving you time by not having to run something on the remote server for the tunnel to remain alive.
    
    %-N: if all you need is to create a tunnel without running any remote commands then include this option to save resources.
    
    %-T: useful to disable pseudo-tty allocation, which is fitting if you are not trying to create an interactive shell.
    
    %autossh utility,

% ssh localhost -p 2222