ssl_server.erl

来自「OTP是开放电信平台的简称」· ERL 代码 · 共 1,376 行 · 第 1/3 页

	    {noreply, St#st{cons = Cons}};
	_Other ->
	    {reply, {error, ebadf}, St}
    end;

%%
%% seed
%%
handle_call({seed, Data}, _From, St) when binary(Data) ->
    send_cmd(St#st.port, ?SET_SEED, [int32(size(Data)), Data]),
    {reply, ok, St};

handle_call({seed, Data}, From, St) ->
    case (catch list_to_binary(Data)) of
	{'EXIT', _} ->
	    {reply, {error, edata}, St};
	Bin  ->
	    handle_call({seed, Bin}, From, St)
    end;

%%
%% setnodelay
%%
handle_call({setnodelay, Broker, Fd, Boolean}, From, St) ->
    debug(St, "setnodelay: broker = ~w, fd = ~w, "
	  "boolean = ~w~n", [Broker, Fd, Boolean]),
    case replace_from_by_fd(Fd, St#st.cons, From) of 
	{ok, _, Cons} ->
	    Val = if Boolean == true -> 1; true -> 0 end,
	    send_cmd(St#st.port, ?SET_SOCK_OPT, 
		     [int32(Fd), ?SET_TCP_NODELAY, Val]),
	    %% We reply when we get IOCTL_OK or IOCTL_ERR.
	    {noreply, St#st{cons = Cons}};
	_Other ->
	    {reply, {error, ebadf}, St}
    end;

%%
%% sockname
%%
handle_call({sockname, Broker, Fd}, From, St) ->
    debug(St, "sockname: broker = ~w, fd = ~w~n",
	  [Broker, Fd]),
    case replace_from_by_fd(Fd, St#st.cons, From) of 
	{ok, _, Cons} ->
	    send_cmd(St#st.port, ?GETSOCKNAME, [int32(Fd)]),
	    %% We reply when we get GETSOCKNAME_REP or GETSOCKNAME_ERR.
	    {noreply, St#st{cons = Cons}};
	_Other ->
	    {reply, {error, ebadf}, St}
    end;

%%
%% version
%%
handle_call(version, From, St) ->
    debug(St, "version: from = ~w~n", [From]),
    {reply, {ok, {St#st.compvsn, St#st.libvsn}}, St};

%%
%% dump
%%
handle_call({dump, Broker}, _From, St) ->
    debug(St, "dump: broker = ~w", [Broker]),
    Port = St#st.port,
    send_cmd(Port, ?DUMP_CMD, []),
    {reply, ok, St};

%%
%% set_debug
%%
handle_call({set_debug, Bool, Broker}, _From, St) ->
    debug(St, "set_debug: broker = ~w", [Broker]),
   Value = case Bool of 
                true ->
                    1;
                false ->
                    0
            end,
    Port = St#st.port,
    send_cmd(Port, ?DEBUG_CMD, [Value]),
    {reply, ok, St};

%%
%% set_debugmsg
%%
handle_call({set_debugmsg, Bool, Broker}, _From, St) ->
    debug(St, "set_debugmsg: broker = ~w", [Broker]),
    Value = case Bool of 
                true ->
                    1;
                false ->
                    0
            end,
    Port = St#st.port,
    send_cmd(Port, ?DEBUGMSG_CMD, [Value]),
    {reply, ok, St};

handle_call(Request, _From, St) ->
    debug(St, "unexpected call: ~w~n", [Request]),
    Reply = {error, {badcall, Request}},
    {reply, Reply, St}.

%%
%% handle_cast(Msg, St)
%%


handle_cast(Msg, St) ->
    debug(St, "unexpected cast: ~w~n", [Msg]),
    {noreply, St}.

%%
%% handle_info(Info, St)
%%

%% Data from port
%%
handle_info({Port, {data, Bin}}, St) 
  when St#st.port == Port, binary(Bin) ->
    %% io:format("++++ ssl_server got from port: ~w~n", [Bin]),

    <<OpCode:8, _/binary>> = Bin,

    case OpCode of
	
	%%
	%% transport_accept
	%%
	?TRANSPORT_ACCEPT_ERR when size(Bin) >= 5 ->
	    {ListenFd, Reason} = decode_msg(Bin, [int32, atom]),
	    debug(St, "transport_accept_err: listenfd = ~w, "
		  "reason = ~w~n", [ListenFd, Reason]),
	    case delete_last_by_fd(ListenFd, St#st.paccepts) of
		{ok, {_, _, From}, PAccepts} ->
		    gen_server:reply(From, {error, Reason}),
		    {noreply, St#st{paccepts = PAccepts}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;
	?TRANSPORT_ACCEPT_REP when size(Bin) >= 9 ->
	    {ListenFd, Fd} = decode_msg(Bin, [int32, int32]),
	    debug(St, "transport_accept_rep: listenfd = ~w, "
		  "fd = ~w~n", [ListenFd, Fd]),
	    case delete_last_by_fd(ListenFd, St#st.paccepts) of
		{ok, {_, Broker, From}, PAccepts} ->
		    Reply = {ok, Fd, St#st.proxylsport},
		    gen_server:reply(From, Reply),
		    debug(St, "transport_accept_rep: From = ~w\n", [From]),
		    Cons = add({Fd, Broker, From}, St#st.cons),
		    {noreply, St#st{cons = Cons, paccepts = PAccepts}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;
	
	%%
	%% ssl_accept
	%%
	?SSL_ACCEPT_ERR when size(Bin) >= 5 ->
	    {Fd, Reason} = decode_msg(Bin, [int32, atom]),
	    debug(St, "ssl_accept_err: listenfd = ~w, "
		  "reason = ~w~n", [Fd, Reason]),
	    %% JC: remove this?
	    case delete_last_by_fd(Fd, St#st.cons) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, {error, Reason}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;
	?SSL_ACCEPT_REP when size(Bin) >= 5 ->
	    Fd = decode_msg(Bin, [int32]),
	    debug(St, "ssl_accept_rep: Fd = ~w\n", [Fd]),
	    case replace_from_by_fd(Fd, St#st.cons, []) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, ok),
		    {noreply, St#st{cons = Cons}};
		_ ->
		    {noreply, St}
	    end;

	%%
	%% connect
	%%
	?CONNECT_SYNC_ERR when size(Bin) >= 5 ->
	    {IntRef, Reason} = decode_msg(Bin, [int32, atom]),
	    debug(St, "connect_sync_err: intref = ~w, "
		  "reason = ~w~n", [IntRef, Reason]),
	    case delete_by_intref(IntRef, St#st.cons) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, {error, Reason}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    {noreply, St}
	    end;
	?CONNECT_WAIT when size(Bin) >= 9 ->  
	    {IntRef, Fd} = decode_msg(Bin, [int32, int32]),
	    debug(St, "connect_wait: intref = ~w, "
		  "fd = ~w~n", [IntRef, Fd]),
	    case replace_fd_by_intref(IntRef, St#st.cons, Fd) of
		{ok, _, Cons} ->
		    %% We reply when we get CONNECT_REP or CONNECT_ERR
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% We have a new Fd which must be closed
		    send_cmd(St#st.port, ?CLOSE, int32(Fd)),
		    {noreply, St}
	    end;
	?CONNECT_REP when size(Bin) >= 5 ->  
	    %% after CONNECT_WAIT
	    Fd = decode_msg(Bin, [int32]),
	    debug(St, "connect_rep: fd = ~w~n", [Fd]),
	    case replace_from_by_fd(Fd, St#st.cons, []) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, 
				     {ok, Fd, St#st.proxylsport}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    {noreply, St}
	    end;
	?CONNECT_ERR when size(Bin) >= 5 ->
	    {Fd, Reason} = decode_msg(Bin, [int32, atom]),
	    debug(St, "connect_err: fd = ~w, "
		  "reason = ~w~n", [Fd, Reason]),
	    case delete_by_fd(Fd, St#st.cons) of
		{ok, {_, _, From}, Cons} ->
		    %% Fd not yet published - hence close ourselves
		    send_cmd(St#st.port, ?CLOSE, int32(Fd)),
		    gen_server:reply(From, {error, Reason}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;

	%%
	%% connection_info
	%%
	?GETCONNINFO_REP when size(Bin) >= 5 ->
	    {Fd, Protocol, Cipher} = decode_msg(Bin, [int32, string, string]),
	    debug(St, "connection_info_rep: fd = ~w, "
		  "protcol = ~p, ip = ~p~n", [Fd, Protocol, Cipher]),
	    case replace_from_by_fd(Fd, St#st.cons, []) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, {ok, {protocol_name(Protocol),
						 Cipher}}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;
	?GETCONNINFO_ERR when size(Bin) >= 5 ->
	    {Fd, Reason} = decode_msg(Bin, [int32, atom]),
	    debug(St, "connection_info_err: fd = ~w, "
		  "reason = ~w~n", [Fd, Reason]),
	    case replace_from_by_fd(Fd, St#st.cons, []) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, {error, Reason}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;

	%%
	%% listen
	%%
	?LISTEN_SYNC_ERR when size(Bin) >= 5 ->
	    {IntRef, Reason} = decode_msg(Bin, [int32, atom]),
	    debug(St, "listen_sync_err: intref = ~w, "
		  "reason = ~w~n", [IntRef, Reason]),
	    case delete_by_intref(IntRef, St#st.cons) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, {error, Reason}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    {noreply, St}
	    end;
	?LISTEN_REP when size(Bin) >= 11 ->  
	    {IntRef, ListenFd, LPort} = decode_msg(Bin, [int32, int32, int16]),
	    debug(St, "listen_rep: intref = ~w, "
		  "listenfd = ~w, sport = ~w~n", [IntRef, ListenFd, LPort]),
	    case replace_fd_from_by_intref(IntRef, St#st.cons, 
					   ListenFd, []) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, {ok, ListenFd, LPort}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% ListenFd has to be closed.
		    send_cmd(St#st.port, ?CLOSE, int32(ListenFd)),
		    {noreply, St}
	    end;

	%%
	%% proxy join
	%%
	?PROXY_JOIN_REP when size(Bin) >= 5 -> 
	    Fd = decode_msg(Bin, [int32]),
	    debug(St, "proxy_join_rep: fd = ~w~n",
		  [Fd]),
	    case get_by_fd(Fd, St#st.cons) of
		{ok, {_, _, From}} ->
		    gen_server:reply(From, ok),
		    {noreply, St};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;
	?PROXY_JOIN_ERR when size(Bin) >= 5 -> 
	    {Fd, Reason} = decode_msg(Bin, [int32, atom]),
	    debug(St, "proxy_join_rep: fd = ~w, "
		  "reason = ~w~n", [Fd, Reason]),
	    case delete_by_fd(Fd, St#st.cons) of
		{ok, {_, _, From}, Cons} ->
		    %% Must not close Fd since it is published
		    gen_server:reply(From, {error, Reason}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;

	%%
	%% peername
	%%
	?GETPEERNAME_REP when size(Bin) >= 5 ->
	    {Fd, LPort, IPString} = decode_msg(Bin, [int32, int16, string]),
	    debug(St, "getpeername_rep: fd = ~w, "
		  "sport = ~w, ip = ~p~n", [Fd, LPort, IPString]),
	    case replace_from_by_fd(Fd, St#st.cons, []) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, {ok, {IPString, LPort}}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;
	?GETPEERNAME_ERR when size(Bin) >= 5 ->
	    {Fd, Reason} = decode_msg(Bin, [int32, atom]),
	    debug(St, "getpeername_err: fd = ~w, "
		  "reason = ~w~n", [Fd, Reason]),
	    case replace_from_by_fd(Fd, St#st.cons, []) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, {error, Reason}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;

	%%
	%% ioctl
	%%
	?IOCTL_OK when size(Bin) >= 5 ->
	    Fd = decode_msg(Bin, [int32]),
	    debug(St, "ioctl_ok: fd = ~w~n",
		  [Fd]),
	    case replace_from_by_fd(Fd, St#st.cons, []) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, ok),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;
	?IOCTL_ERR when size(Bin) >= 5 ->
	    {Fd, Reason} = decode_msg(Bin, [int32, atom]),
	    debug(St, "ioctl_err: fd = ~w, "
		  "reason = ~w~n", [Fd, Reason]),
	    case replace_from_by_fd(Fd, St#st.cons, []) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, {error, Reason}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;

	%%
	%% sockname
	%%
	?GETSOCKNAME_REP when size(Bin) >= 5 ->
	    {Fd, LPort, IPString} = decode_msg(Bin, [int32, int16, string]),
	    debug(St, "getsockname_rep: fd = ~w, "
		  "sport = ~w, ip = ~p~n", [Fd, LPort, IPString]),
	    case replace_from_by_fd(Fd, St#st.cons, []) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, {ok, {IPString, LPort}}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;
	?GETSOCKNAME_ERR when size(Bin) >= 5 ->
	    {Fd, Reason} = decode_msg(Bin, [int32, atom]),
	    debug(St, "getsockname_err: fd = ~w, "
		  "reason = ~w~n", [Fd, Reason]),
	    case replace_from_by_fd(Fd, St#st.cons, []) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, {error, Reason}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;

	%%
	%% peercert
	%%
	?GETPEERCERT_REP when size(Bin) >= 5 ->
	    {Fd, Cert} = decode_msg(Bin, [int32, bin]),
	    debug(St, "getpeercert_rep: fd = ~w~n", [Fd]),
	    case replace_from_by_fd(Fd, St#st.cons, []) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, {ok, Cert}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end;
	?GETPEERCERT_ERR when size(Bin) >= 5 ->
	    {Fd, Reason} = decode_msg(Bin, [int32, atom]),
	    debug(St, "getpeercert_err: fd = ~w, reason = ~w~n", 
		  [Fd, Reason]),
	    case replace_from_by_fd(Fd, St#st.cons, []) of
		{ok, {_, _, From}, Cons} ->
		    gen_server:reply(From, {error, Reason}),
		    {noreply, St#st{cons = Cons}};
		_Other ->
		    %% Already closed
		    {noreply, St}
	    end
    end;

%%
%% EXIT
%%
handle_info({'EXIT', Pid, Reason}, St) when pid(Pid) ->
    debug(St, "exit pid = ~w, "
	  "reason = ~w~n", [Pid, Reason]),
    case delete_by_pid(Pid, St#st.cons) of
	{ok, {{intref, _}, Pid, _}, Cons} ->
	    {noreply, St#st{cons = Cons}};
	{ok, {Fd, Pid, _}, Cons} ->
	    send_cmd(St#st.port, ?CLOSE, int32(Fd)), 
	    %% If Fd is a listen socket fd, there might be pending
	    %% accepts for that fd.
	    case delete_all_by_fd(Fd, St#st.paccepts) of
		{ok, DelAccepts, RemAccepts} ->
		    %% Reply {error, closed} to all pending accepts.
		    lists:foreach(fun({_, _, From}) ->
					  gen_server:reply(From, 
							   {error, closed}) 
				  end, DelAccepts),
		    {noreply, 
		     St#st{cons = Cons, paccepts = RemAccepts}};