orber_iiop_pm.erl

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

%%--------------------------------------------------------------------
%% ``The contents of this file are subject to the Erlang Public License,
%% Version 1.1, (the "License"); you may not use this file except in
%% compliance with the License. You should have received a copy of the
%% Erlang Public License along with this software. If not, it can be
%% retrieved via the world wide web at http://www.erlang.org/.
%% 
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
%% the License for the specific language governing rights and limitations
%% under the License.
%% 
%% The Initial Developer of the Original Code is Ericsson Utvecklings AB.
%% Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings
%% AB. All Rights Reserved.''
%% 
%%     $Id$
%%
%%-----------------------------------------------------------------
%% File: orber_iiop_pm.erl
%% Description:
%%    This file contains the mapping of addresses on the format {Host, Port} 
%%    to a proxy pid.
%%
%% Creation date: 990615
%%
%%-----------------------------------------------------------------
-module(orber_iiop_pm).

-behaviour(gen_server).

-include_lib("orber/src/orber_iiop.hrl").
-include_lib("orber/include/corba.hrl").
-include_lib("kernel/include/inet.hrl").

%%-----------------------------------------------------------------
%% External exports
%%-----------------------------------------------------------------
-export([start/0, start/1]).

%%-----------------------------------------------------------------
%% Internal exports
%%-----------------------------------------------------------------
-export([connect/7, 
	 close_connection/1, close_connection/2,
	 list_existing_connections/0, 
	 list_setup_connections/0, 
	 list_all_connections/0,
	 init/1, handle_call/3, handle_cast/2, handle_info/2,
	 code_change/3, terminate/2, stop/0, setup_connection/8,
	 reconfigure/1, reconfigure/3, reconfigure/4, add_connection/3,
	 sockname2peername/2, peername2sockname/2]).

%%-----------------------------------------------------------------
%% Macros/Defines
%%-----------------------------------------------------------------
-define(DEBUG_LEVEL, 7).

-define(PM_CONNECTION_DB, orber_iiop_pm_db).

-record(state, {connections, queue}).

-record(connection, {hp, child, interceptors, slave, 
		     flags = 0, alias = 0, socketdata = {"Unavailable", 0}}).

%%-----------------------------------------------------------------
%% External interface functions
%%-----------------------------------------------------------------
start() ->
    ignore.
start(Opts) ->
    gen_server:start_link({local, 'orber_iiop_pm'}, ?MODULE, Opts, []).


connect(Host, Port, SocketType, Timeout, Chars, Wchars, Ctx)
  when SocketType == normal ->
    Key = create_key(Host, Port, Ctx),
    case ets:lookup(?PM_CONNECTION_DB, Key) of
	[#connection{child = connecting}] ->
	    gen_server:call(orber_iiop_pm, {connect, Host, Port, SocketType, 
					    [], Chars, Wchars, Key}, Timeout);
	[] ->
	    gen_server:call(orber_iiop_pm, {connect, Host, Port, SocketType, 
					    [], Chars, Wchars, Key}, Timeout);
	[#connection{hp = {_, _, 0}, child = P, interceptors = I}] ->
	    {ok, P, [], I, 0};
	[#connection{hp = {_, _, Interface}, child = P, interceptors = I}] ->
	    {ok, P, [], I, [Interface]}
    end;
connect(Host, Port, SocketType, Timeout, Chars, Wchars, Ctx) 
  when SocketType == ssl ->
    Key = create_key(Host, Port, Ctx),
    case ets:lookup(?PM_CONNECTION_DB, Key) of
	[#connection{child = connecting}] ->
	    SocketOptions = get_ssl_socket_options(),
	    gen_server:call(orber_iiop_pm, {connect, Host, Port, SocketType, 
					    SocketOptions, Chars, Wchars, Key}, 
			    Timeout);
	[] ->
	    SocketOptions = get_ssl_socket_options(),
	    gen_server:call(orber_iiop_pm, {connect, Host, Port, SocketType, 
					    SocketOptions, Chars, Wchars, Key}, 
			    Timeout);
	[#connection{hp = {_, _, 0}, child = P, interceptors = I}] ->
	    {ok, P, [], I, 0};
	[#connection{hp = {_, _, Interface}, child = P, interceptors = I}] ->
	    {ok, P, [], I, [Interface]}
    end.

get_ssl_socket_options() ->
    [{verify, orber:ssl_client_verify()},
     {depth, orber:ssl_client_depth()} |
     ssl_client_extra_options([{certfile, orber:ssl_client_certfile()},
			       {cacertfile, orber:ssl_client_cacertfile()},
			       {password, orber:ssl_client_password()},
			       {keyfile, orber:ssl_client_keyfile()},
			       {ciphers, orber:ssl_client_ciphers()},
			       {cachetimeout, orber:ssl_client_cachetimeout()}], [])].

ssl_client_extra_options([], Acc) ->
    Acc;
ssl_client_extra_options([{_Type, []}|T], Acc) ->
    ssl_client_extra_options(T, Acc);
ssl_client_extra_options([{_Type, infinity}|T], Acc) ->
    ssl_client_extra_options(T, Acc);
ssl_client_extra_options([{Type, Value}|T], Acc) ->
    ssl_client_extra_options(T, [{Type, Value}|Acc]).

add_connection(Key, Key, SockData) ->
    case ets:lookup(?PM_CONNECTION_DB, Key) of
	[Connection] ->
	    ets:insert(?PM_CONNECTION_DB, 
		       Connection#connection{socketdata = SockData});
	[] ->
	    ets:insert(?PM_CONNECTION_DB, 
		       #connection{hp= Key, child = connecting,
				   socketdata = SockData})
    end;
add_connection(Key, NewKey, SockData) ->
    add_connection(Key, Key, SockData),
    add_connection(NewKey, NewKey, SockData).

get_socket_data(Key) ->
    case ets:lookup(?PM_CONNECTION_DB, Key) of
	[#connection{socketdata = SockData}] ->
	    SockData;
	_ ->
	    {"Unable to extract socket information", 0}
    end.

sockname2peername(SockHost, SockPort) ->
    orber_tb:unique(
      do_select([{#connection{hp = {'$1', '$2', '_'},
			      socketdata = {match_type(SockHost), 
					    match_type(SockPort)}, 
			      _='_'}, [], [{{'$1', '$2'}}]}])).


peername2sockname(PeerHost, PeerPort) ->
    orber_tb:unique(
      do_select([{#connection{hp = {match_type(PeerHost), 
				    match_type(PeerPort),
				    '_'},
			      socketdata = '$1', 
			      _='_'}, [], ['$1']}])).

match_type(0) -> 
    %% Wildcard port number
    '_';
match_type("") -> 
    %% Wildcard host
    '_';
match_type(Key) -> 
    %% Wildcard not used.
    Key.

create_key(Host, Port, []) ->
    {Host, Port, 0};
create_key(Host, Port, 
	   [#'IOP_ServiceContext'
	    {context_id=?ORBER_GENERIC_CTX_ID, 
	     context_data = {interface, Interface}}|_]) when list(Interface) ->
    {Host, Port, Interface};
create_key(Host, Port, 
	   [#'IOP_ServiceContext'
	    {context_id=?ORBER_GENERIC_CTX_ID, 
	     context_data = {interface, Interface}}|_]) ->
    orber:dbg("[~p] orber_iiop_pm:create_key(~p, ~p);~n"
	      "The supplied interface must be a string.",
	      [?LINE, Host, Port, Interface], ?DEBUG_LEVEL),
    corba:raise(#'BAD_CONTEXT'{completion_status=?COMPLETED_NO});
create_key(Host, Port, [_|T]) ->
    create_key(Host, Port, T).

reconfigure(Options) ->
    {Local, Proxy} = check_options(Options, [], []),
    reconfigure_local(Local),
    reconfigure_proxy(Proxy).


reconfigure(Options, Host, Port) ->
    reconfigure(Options, Host, Port, 0).
reconfigure(Options, Host, Port, Interface) ->
    case ets:lookup(?PM_CONNECTION_DB, {Host, Port, Interface}) of
	[#connection{child = P}] when pid(P) ->
	    case check_options(Options, [], []) of
		{[], Proxy} ->
		    reconfigure_proxy(Proxy, [P]);
		{Local, Proxy} ->
		    reconfigure_proxy(Proxy, [P]),
		    gen_server:call(orber_iiop_pm, {reconfigure, Local, 
						    Host, Port, Interface}, infinity)
	    end;
	_ ->
	    {error, "No proxy matched the supplied reference"}
    end.

reconfigure_local([]) ->
    ok;
reconfigure_local(Options) ->
    gen_server:call(orber_iiop_pm, {reconfigure, Options}, infinity).

reconfigure_proxy([]) ->
    ok;
reconfigure_proxy(Options) ->
    reconfigure_proxy(Options, do_select([{#connection{child = '$1', _='_'}, 
					   [], ['$1']}])).

reconfigure_proxy(Options, [Pid|T]) ->
    Pid ! {reconfigure, Options},
    reconfigure_proxy(Options, T);
reconfigure_proxy(_Options, []) ->
    ok.


check_options([{interceptors, false}|Options], Local, Proxy) ->
    check_options(Options, [{interceptors, false}|Local], Proxy);
check_options([{interceptors, {native, LPIs}}|Options], Local, Proxy) ->
    check_options(Options, [{interceptors, {native, LPIs}}|Local], Proxy);
check_options([_|Options], Local, Proxy) ->
    check_options(Options, Local, Proxy);
check_options([], Local, Proxy) ->
    {Local, Proxy}.


close_connection(PeerData) ->
    close_connection(PeerData, 0).

close_connection(PeerData, Interface) ->
    gen_server:call(orber_iiop_pm, {disconnect, PeerData, Interface}, infinity).


list_existing_connections() ->
    transform(
      lists:sort(
	do_select([{#connection{hp = {'$2','$3','$4'}, child = '$1', _='_'}, 
		    [{is_pid, '$1'}], [{{'$1', '$2','$3','$4'}}]}])), []).

list_setup_connections() ->
    transform(
      lists:sort(
	do_select([{#connection{hp = {'$1','$2','$3'}, child = connecting, _='_'}, [], 
			     [{{'$1','$2','$3'}}]}])), []).

list_all_connections() ->
    transform(
      lists:sort(
	do_select([{#connection{hp = {'$2','$3','$4'}, child = '$1', _='_'}, [], 
		    [{{'$1','$2','$3', '$4'}}]}])), []).

%% Since the connections interface can be 0 or an ip-address we want to
%% transform those containing 0.
transform([{C, H, P, 0}, {C, H, P, I}|T], Acc) ->
    %% ACL defined interface. Drop the anonymous one.
    transform(T, [{H, P, I}|Acc]);
transform([{_C, H, P, 0}|T], Acc) ->
    %% No interface supplied. Drop the 0.
    transform(T, [{H, P}|Acc]);
transform([{_C, H, P, I}|T], Acc) ->
    %% Interface supplied. Keep it.
    transform(T, [{H, P, I}|Acc]);
transform([{H,P,0}|T], Acc) ->
    transform(T, [{H,P}|Acc]);
transform([{H,P,I}|T], Acc) ->
    transform(T, [{H,P,I}|Acc]);
transform([H|T], Acc) ->
    transform(T, [H|Acc]);
transform([], Acc) ->
    Acc.

do_select(Pattern) ->   
    case catch ets:select(?PM_CONNECTION_DB, Pattern) of
	{'EXIT', _What} ->
	    [];
	Result ->
	    Result
    end.

%%-----------------------------------------------------------------
%% Internal interface functions
%%-----------------------------------------------------------------
%%-----------------------------------------------------------------
%% Func: stop/0 (Only used for test purpose !!!!!!)
%%-----------------------------------------------------------------
stop() ->
    gen_server:call(orber_iiop_pm, stop).

%%-----------------------------------------------------------------
%% Server functions
%%-----------------------------------------------------------------
%%-----------------------------------------------------------------
%% Func: init/1
%%-----------------------------------------------------------------
init(_Opts) ->
    process_flag(trap_exit, true),
    {ok, #state{connections = ets:new(orber_iiop_pm_db, 
				      [{keypos, 2}, set, public, named_table]),
		queue = ets:new(orber_iiop_pm_queue, [bag])}}.

%%-----------------------------------------------------------------
%% Func: terminate/2
%%-----------------------------------------------------------------
terminate(_Reason, #state{queue = Q}) ->
    %% Kill all proxies and close table before terminating
    stop_all_proxies(ets:first(?PM_CONNECTION_DB)),
    ets:delete(?PM_CONNECTION_DB),
    ets:delete(Q),
    ok.

stop_all_proxies('$end_of_table') ->
    ok;
stop_all_proxies(Key) ->
    case ets:lookup(?PM_CONNECTION_DB, Key) of
	[] ->
	    ok;
	[#connection{child = connecting, interceptors = I}] ->
	    invoke_connection_closed(I);
	[#connection{child = P, interceptors = I}] ->
	    invoke_connection_closed(I),
	    catch orber_iiop_outproxy:stop(P)
    end,
    stop_all_proxies(ets:next(?PM_CONNECTION_DB, Key)).

%%-----------------------------------------------------------------
%% Func: handle_call/3
%%-----------------------------------------------------------------
handle_call({connect, Host, Port, SocketType, SocketOptions, Chars, Wchars, Key}, 
	    From, State) ->
    case ets:lookup(?PM_CONNECTION_DB, Key) of
	[#connection{child = connecting}] ->
	    %% Another client already requested a connection to the given host/port. 
	    %% Just add this client to the queue.
	    ets:insert(State#state.queue, {Key, From}),
	    {noreply, State};
	[#connection{hp = {_,_,0}, child = P, interceptors = I}] ->
	    %% This case will occur if the PortMapper completed a connection
	    %% between the client's ets:lookup and receiving this request.
	    {reply, {ok, P, [], I, 0}, State};
	[#connection{hp = {_,_,Intf}, child = P, interceptors = I}] ->
	    %% This case will occur if the PortMapper completed a connection
	    %% between the client's ets:lookup and receiving this request.
	    {reply, {ok, P, [], I, [Intf]}, State};
	[] ->
	    %% The first time a connection is requested to the given host/port.
	    case catch spawn_link(?MODULE, setup_connection, 
				  [self(), Host, Port, SocketType, 
				   SocketOptions, Chars, Wchars, Key]) of
		Slave when pid(Slave) ->
		    ets:insert(?PM_CONNECTION_DB, 
			       #connection{hp = Key, child = connecting, 
					   interceptors = false, slave = Slave}),
		    ets:insert(State#state.queue, {Key, From}),
		    {noreply, State};
		What ->
		    orber:dbg("[~p] orber_iiop_pm:handle_call(connect);~n"
			      "Unable to invoke setup_connection due to: ~n~p~n", 
			      [?LINE, What], ?DEBUG_LEVEL),
		    {reply, 
		     {'EXCEPTION', #'INTERNAL'{completion_status=?COMPLETED_NO}}, 
		     State}
	    end
    end;
handle_call({disconnect, PeerData, Interface}, _From, State) ->
    {reply, do_disconnect(PeerData, Interface, State), State};
handle_call({reconfigure, Options, Host, Port, Interface}, 
	    _From, State) ->
    case ets:lookup(?PM_CONNECTION_DB, {Host, Port, Interface}) of
	[] ->
	    {reply, {error, "No proxy matched the supplied reference"}, State};
	[Connection] ->
	    NewConnection = update_connection(Connection, Options),
	    ets:insert(?PM_CONNECTION_DB, NewConnection),
	    {reply, ok, State}    
    end;
handle_call({reconfigure, Options}, _From, State) ->
    case catch update_db(ets:first(?PM_CONNECTION_DB), Options) of