orber_iiop_pm.erl

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

	ok ->
	    {reply, ok, State};
	_What ->
	    {reply, {error, "Unable to change configuration"}, State}
    end;
handle_call(stop, _From, State) ->
    {stop, normal, ok, State};
handle_call(_, _, State) ->
    {noreply, State}.

update_db('$end_of_table', _) ->
    ok;
update_db(Key, Options) ->
    [Connection] = ets:lookup(?PM_CONNECTION_DB, Key),
    NewConnection = update_connection(Connection, Options),
    ets:insert(?PM_CONNECTION_DB, NewConnection),
    update_db(ets:next(?PM_CONNECTION_DB, Key), Options).


update_connection(Connection, [{interceptors, false}|Options]) ->
    update_connection(Connection#connection{interceptors = false}, Options);
update_connection(#connection{interceptors = false,
			      hp = {PH, PP, _},
			      socketdata = {SH, SP}} = Connection, 
		  [{interceptors, {native, LPIs}}|Options]) ->
    %% No Interceptor(s). Add the same Ref used by the built in interceptors.
    update_connection(Connection#connection{interceptors = 
					    {native, {PH, PP, SH, SP}, LPIs}},
		      Options);
update_connection(#connection{interceptors = {native, Ref, _}} = Connection, 
		  [{interceptors, {native, LPIs}}|Options]) ->
    %% Interceptor(s) already in use. We must use the same Ref as before.
    update_connection(Connection#connection{interceptors = 
					    {native, Ref, LPIs}},
		      Options);
update_connection(Connection, [H|T]) ->
    orber:dbg("[~p] orber_iiop_pm:update_connection(~p, ~p)~n"
	      "Unable to update the connection.~n", 
	      [?LINE, Connection, H], ?DEBUG_LEVEL),
    update_connection(Connection, T);
update_connection(Connection, []) ->
    Connection.

do_disconnect([], _Interface, _State) ->
    ok;
do_disconnect([{Host, Port}|T], Interface, State) ->
    case ets:lookup(?PM_CONNECTION_DB, {Host, Port, Interface}) of
	[] ->
	    ok;
	[#connection{child = connecting, interceptors = I}] ->
	    ets:delete(?PM_CONNECTION_DB, {Host, Port, Interface}),
	    Exc = {'EXCEPTION',#'INTERNAL'{completion_status = ?COMPLETED_NO}},
	    send_reply_to_queue(ets:lookup(State#state.queue, 
					   {Host, Port, Interface}), Exc),
	    ets:delete(State#state.queue, {Host, Port, Interface}),
	    invoke_connection_closed(I);
	[#connection{child = P, interceptors = I}] ->
	    unlink(P),
	    catch orber_iiop_outproxy:stop(P),
	    ets:delete(?PM_CONNECTION_DB, {Host, Port, Interface}),
	    invoke_connection_closed(I)
    end,
    do_disconnect(T, Interface, State).

%%-----------------------------------------------------------------
%% Func: handle_cast/2
%%-----------------------------------------------------------------
handle_cast(stop, State) ->
    {stop, normal, State};
handle_cast(_, State) ->
    {noreply, State}.

%%-----------------------------------------------------------------
%% Func: handle_info/2
%%-----------------------------------------------------------------
%% Trapping exits 
handle_info({'EXIT', Pid, Reason}, State) ->
    %% Check the most common scenario first, i.e., a proxy terminates.
    case ets:match_object(?PM_CONNECTION_DB, #connection{child = Pid, _='_'}) of
	[#connection{hp = K, interceptors = I}] ->
	    ets:delete(?PM_CONNECTION_DB, K),
	    invoke_connection_closed(I),
	    {noreply, State};
	[#connection{hp = K, interceptors = I}, #connection{hp = K2}] ->
	    ets:delete(?PM_CONNECTION_DB, K),
	    ets:delete(?PM_CONNECTION_DB, K2),
	    invoke_connection_closed(I),
	    {noreply, State};
	[] when Reason == normal ->
	    %% This might have been a spawned 'setup_connection' which terminated
	    %% after sucessfully setting up a new connection.
	    {noreply, State};
	[] ->
	    %% Wasn't a proxy. Hence, we must test if it was a spawned
	    %% 'setup_connection' that failed.
	    case ets:match_object(?PM_CONNECTION_DB, #connection{slave = Pid, _='_'}) of
		[#connection{hp = K, child = connecting, interceptors = I}] ->
		    ets:delete(?PM_CONNECTION_DB, K),
		    invoke_connection_closed(I),
		    Exc = {'EXCEPTION',#'INTERNAL'{completion_status = ?COMPLETED_NO}},
		    send_reply_to_queue(ets:lookup(State#state.queue, K), Exc),
		    ets:delete(State#state.queue, K),
		    orber:dbg("[~p] orber_iiop_pm:handle_info(setup_failed ~p);~n"
			      "It was not possible to create a connection to the"
			      " given host/port.",
			      [?LINE, K], ?DEBUG_LEVEL),
		    {noreply, State};
		[#connection{hp = K, child = connecting, interceptors = I}, 
		 #connection{hp = K2}] ->
		    ets:delete(?PM_CONNECTION_DB, K),
		    ets:delete(?PM_CONNECTION_DB, K2),
		    invoke_connection_closed(I),
		    Exc = {'EXCEPTION',#'INTERNAL'{completion_status = ?COMPLETED_NO}},
		    send_reply_to_queue(ets:lookup(State#state.queue, K), Exc),
		    ets:delete(State#state.queue, K),
		    orber:dbg("[~p] orber_iiop_pm:handle_info(setup_failed ~p);~n"
			      "It was not possible to create a connection to the"
			      " given host/port.",
			      [?LINE, K], ?DEBUG_LEVEL),
		    {noreply, State};
		_ ->
		    {noreply, State}
	    end
    end;
handle_info({setup_failed, {Host, Port, _} = Key, Key, Exc}, State) ->
    %% Deletet the data from the connection DB first to avoid clients from
    %% trying to access it again.
    ets:delete(?PM_CONNECTION_DB, Key),
    %% Now we can send whatever exception received.
    send_reply_to_queue(ets:lookup(State#state.queue, Key), Exc),
    ets:delete(State#state.queue, Key),
    orber:dbg("[~p] orber_iiop_pm:handle_info(setup_failed ~p ~p);~n"
	      "It was not possible to create a connection to the given host/port.", 
	      [?LINE, Host, Port], ?DEBUG_LEVEL),
    {noreply, State};
handle_info({setup_failed, {Host, Port, _} = Key, NewKey, Exc}, State) ->
    %% Deletet the data from the connection DB first to avoid clients from
    %% trying to access it again.
    ets:delete(?PM_CONNECTION_DB, Key),
    ets:delete(?PM_CONNECTION_DB, NewKey),
    %% Now we can send whatever exception received.
    send_reply_to_queue(ets:lookup(State#state.queue, Key), Exc),
    ets:delete(State#state.queue, Key),
    orber:dbg("[~p] orber_iiop_pm:handle_info(setup_failed ~p ~p);~n"
	      "It was not possible to create a connection to the given host/port.", 
	      [?LINE, Host, Port], ?DEBUG_LEVEL),
    {noreply, State};
handle_info({setup_successfull, Key, Key, {Child, Ctx, Int}}, State) ->
    %% Create a link to the proxy and store it in the connection DB.
    link(Child),
    case ets:lookup(?PM_CONNECTION_DB, Key) of
	[Connection] ->
	    ets:insert(?PM_CONNECTION_DB, 
		       Connection#connection{hp = Key, child = Child, 
					     interceptors = Int, 
					     slave = undefined});
	[] ->
	    ets:insert(?PM_CONNECTION_DB, 
		       #connection{hp = Key, child = Child, 
				   interceptors = Int, 
				   slave = undefined})
    end,
    %% Send the Proxy reference to all waiting clients.
    case Key of
	{_, _, 0} ->
	    send_reply_to_queue(ets:lookup(State#state.queue, Key),
				{ok, Child, Ctx, Int, 0});
	{_, _, Interface} ->
	    send_reply_to_queue(ets:lookup(State#state.queue, Key),
				{ok, Child, Ctx, Int, [Interface]})
    end,
    %% Reset the queue.
    ets:delete(State#state.queue, Key),
    {noreply, State};
handle_info({setup_successfull, Key, NewKey, {Child, Ctx, Int}}, State) ->
    %% Create a link to the proxy and store it in the connection DB.
    link(Child),
    case ets:lookup(?PM_CONNECTION_DB, NewKey) of
	[Connection] ->
	    ets:insert(?PM_CONNECTION_DB, 
		       Connection#connection{hp = NewKey, child = Child, 
					     interceptors = Int, 
					     slave = undefined});
	[] ->
	    ets:insert(?PM_CONNECTION_DB, 
		       #connection{hp = NewKey, child = Child, 
				   interceptors = Int, 
				   slave = undefined})
    end,
    case ets:lookup(?PM_CONNECTION_DB, Key) of
	[Connection2] ->
	    ets:insert(?PM_CONNECTION_DB, 
		       Connection2#connection{hp = Key, child = Child, 
					     interceptors = Int, 
					     slave = undefined});
	[] ->
	    ets:insert(?PM_CONNECTION_DB, 
		       #connection{hp = Key, child = Child, 
				   interceptors = Int, 
				   slave = undefined})
    end,
    %% Send the Proxy reference to all waiting clients.
    case NewKey of
	{_, _, 0} ->
	    send_reply_to_queue(ets:lookup(State#state.queue, Key),
				{ok, Child, Ctx, Int, 0});
	{_, _, Interface} ->
	    send_reply_to_queue(ets:lookup(State#state.queue, Key),
				{ok, Child, Ctx, Int, [Interface]})
    end,
    %% Reset the queue.
    ets:delete(State#state.queue, Key),
    {noreply, State};
handle_info(_, State) ->
    {noreply, State}.


send_reply_to_queue([], _) ->
    ok;
send_reply_to_queue([{_, Client}|T], Reply) ->
    gen_server:reply(Client, Reply),
    send_reply_to_queue(T, Reply). 

%%-----------------------------------------------------------------
%% Func: code_change/3
%%-----------------------------------------------------------------
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%-----------------------------------------------------------------
%% Internal functions
%%-----------------------------------------------------------------
setup_connection(PMPid, Host, Port, SocketType, SocketOptions, Chars, Wchars, Key) ->
    case catch access_allowed(Host, Port, SocketType, Key) of
	ok  ->
	    do_setup_connection(PMPid, Host, Port, SocketType, SocketOptions, 
				Chars, Wchars, Key, Key);
	{ok, Interface} ->
	    do_setup_connection(PMPid, Host, Port, SocketType, 
				[{ip, Interface}|SocketOptions], 
				Chars, Wchars, Key, Key);
	{ok, Interface, NewKey} ->
	    do_setup_connection(PMPid, Host, Port, SocketType, 
				[{ip, Interface}|SocketOptions], 
				Chars, Wchars, Key, NewKey);
	false ->
	    orber_tb:info("Blocked connect attempt to ~s - ~p", [Host, Port]),
	    PMPid ! {setup_failed, Key, Key,
		     {'EXCEPTION', #'NO_PERMISSION'{completion_status=?COMPLETED_NO}}},
	    ok;
	Reason ->
	    orber:dbg("[~p] orber_iiop_pm:handle_call(connect ~p ~p); failed~n"
		      "Reason: ~p", 
		      [?LINE, Host, Port, Reason], ?DEBUG_LEVEL),
	    PMPid ! {setup_failed, Key, Key,
		     {'EXCEPTION', #'COMM_FAILURE'{completion_status=?COMPLETED_NO}}},
	    ok
    end.


do_setup_connection(PMPid, Host, Port, SocketType, SocketOptions, Chars, 
		    Wchars, Key, NewKey) ->
    case catch orber_iiop_outsup:connect(Host, Port, SocketType, 
					 SocketOptions, PMPid, Key, NewKey) of
	{error, {'EXCEPTION', E}} ->
	    orber:dbg("[~p] orber_iiop_pm:handle_call(connect ~p ~p);~n"
		      "Raised Exc: ~p", 
		      [?LINE, Host, Port, E], ?DEBUG_LEVEL),
	    PMPid ! {setup_failed, Key, NewKey, {'EXCEPTION', E}},
	    ok;
	{error, Reason} ->
	    orber:dbg("[~p] orber_iiop_pm:handle_call(connect ~p ~p);~n"
		      "Got EXIT: ~p", 
		      [?LINE, Host, Port, Reason], ?DEBUG_LEVEL),
	    PMPid ! {setup_failed, Key, NewKey, 
		     {'EXCEPTION', #'INTERNAL'{completion_status=?COMPLETED_NO}}},
	    ok;
	{ok, undefined} ->
	    orber:dbg("[~p] orber_iiop_pm:handle_call(connect ~p ~p);~n"
		      "Probably no listener on the given Node/Port or timedout.",
		      [?LINE, Host, Port], ?DEBUG_LEVEL),
	    PMPid ! {setup_failed, Key, NewKey, 
		     {'EXCEPTION', #'COMM_FAILURE'{minor=(?ORBER_VMCID bor 1),
						   completion_status=?COMPLETED_NO}}},
	    ok;
	{ok, Child} ->
	    case init_interceptors(Host, Port, get_socket_data(Key)) of
		{'EXCEPTION', E} ->
		    PMPid ! {setup_failed, Key, NewKey, {'EXCEPTION', E}},
		    ok;
		Interceptors ->
		    BiDirCtx = orber:bidir_context(),
		    Ctx = case orber:exclude_codeset_ctx() of
			      true ->
				  BiDirCtx;
			      _ ->
				  CodeSetCtx = 
				      #'CONV_FRAME_CodeSetContext'
				    {char_data =  Chars, 
				     wchar_data = Wchars},
				  [#'IOP_ServiceContext'
				   {context_id=?IOP_CodeSets, 
				    context_data = CodeSetCtx} | BiDirCtx]
			  end,
		    PMPid ! {setup_successfull, Key, NewKey, 
			     {Child, Ctx, Interceptors}},
		    ok
	    end
    end.

access_allowed(Host, Port, Type, {_,_,UserInterface}) ->
    Flags = orber:get_flags(),
    Family = orber_env:ip_version(),
    case ?ORB_FLAG_TEST(Flags, ?ORB_ENV_USE_ACL_OUTGOING) of
	false when UserInterface == 0 ->
	    get_local_interface(Type, Family);
	false ->
	    inet:getaddr(UserInterface, Family);
	true ->
	    SearchFor = 
		case Type of
		    normal ->
			tcp_out;
		    ssl ->
			ssl_out
		end,
	    {ok, Ip} = inet:getaddr(Host, Family),
	    case orber_acl:match(Ip, SearchFor, true) of
		{true, [], 0} ->
		    get_local_interface(Type, Family);
		{true, [], Port} ->
		    get_local_interface(Type, Family);
		{true, [], {Min, Max}} when Port >= Min, Port =< Max ->
		    get_local_interface(Type, Family);
		{true, [Interface], 0} ->
		    {ok, NewIp} = inet:getaddr(Interface, Family),
		    {ok, NewIp, {Host, Port, 0}};
		{true, [Interface], Port} ->
		    {ok, NewIp} = inet:getaddr(Interface, Family),
		    {ok, NewIp, {Host, Port, 0}};
		{true, [Interface], {Min, Max}} when Port >= Min, Port =< Max ->
		    {ok, NewIp} = inet:getaddr(Interface, Family),
		    {ok, NewIp, {Host, Port, 0}};
		_ ->
		    false
	    end
    end.

get_local_interface(normal, Family) ->
    case orber_env:ip_address_local() of
	[] ->
	    ok;
	[Interface] ->
	    inet:getaddr(Interface, Family)
    end;
get_local_interface(ssl, Family) ->
    case orber_env:iiop_ssl_ip_address_local() of
	[] ->
	    ok;
	[Interface] ->
	    inet:getaddr(Interface, Family)
    end.


invoke_connection_closed(false) ->
    ok;
invoke_connection_closed({native, Ref, PIs}) ->
    (catch orber_pi:closed_out_connection(PIs, Ref));
invoke_connection_closed({_Type, _PIs}) ->
    ok.


init_interceptors(Host, Port, {SHost, SPort}) ->
    case orber:get_interceptors() of
	{native, PIs} ->
	    case catch orber_pi:new_out_connection(PIs, Host, Port, SHost, SPort) of
		{'EXIT', R} ->
		    orber:dbg("[~p] orber_iiop_pm:init_interceptors(~p); Got Exit: ~p.~n"
			      "One or more Interceptor incorrect or undefined?", 
			      [?LINE, PIs, R], ?DEBUG_LEVEL),
		    {'EXCEPTION', #'COMM_FAILURE'{minor=(?ORBER_VMCID bor 2), 
						  completion_status=?COMPLETED_NO}};
		IntRef ->
		    {native, IntRef, PIs}
	    end;
	Other ->
            %% Either 'false' or {Type, PIs}.
	    Other
    end.
    

%%-----------------------------------------------------------------
%% END OF MODULE
%%-----------------------------------------------------------------