orber_iiop_outproxy.erl

来自「OTP是开放电信平台的简称」· ERL 代码 · 共 530 行 · 第 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_outproxy.erl
%% 
%% Description:
%%    This file contains the IIOP "proxy" for outgoing connections
%%
%% Creation date: 990425
%%
%%-----------------------------------------------------------------
-module(orber_iiop_outproxy).

-behaviour(gen_server).

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

%%-----------------------------------------------------------------
%% External exports
%%-----------------------------------------------------------------
-export([start/0, start/1, request/5, cancel/2, cancel/3]).

%%-----------------------------------------------------------------
%% Internal exports
%%-----------------------------------------------------------------
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
	 code_change/3, terminate/2, stop/2, stop/1, checkheaders/1]).

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

-record(state, {stype, socket, db, timeout, client_timeout, host, port, parent,
		error_reason = {'EXCEPTION', #'COMM_FAILURE'
				{completion_status=?COMPLETED_MAYBE}}}).

%%-----------------------------------------------------------------
%% External interface functions
%%-----------------------------------------------------------------
start() ->
    ignore.

start(Opts) ->
    gen_server:start_link(orber_iiop_outproxy, Opts, []).

request(Pid, true, Timeout, Msg, RequestId) ->
    %% Why not simply use gen_server:call? We must be able to receive
    %% more than one reply (i.e. fragmented messages).
    MRef = erlang:monitor(process, Pid),
    gen_server:cast(Pid, {request, Timeout, Msg, RequestId, self(), MRef}),
    receive
	{MRef, Reply} ->
	    erlang:demonitor(MRef),
            receive 
                {'DOWN', MRef, _, _, _} -> 
                    Reply
            after 0 -> 
                    Reply
            end;
	{'DOWN', MRef, _, Pid, _Reason} when pid(Pid) ->
            receive
		%% Clear EXIT message from queue
                {'EXIT', _Pid, _What} -> 
                    corba:raise(#'COMM_FAILURE'{completion_status=?COMPLETED_MAYBE})
            after 0 ->
                    corba:raise(#'COMM_FAILURE'{completion_status=?COMPLETED_MAYBE})
            end;
	{fragmented, GIOPHdr, Bytes, RequestId, MRef} ->
	    collect_fragments(GIOPHdr, [], Bytes, Pid, RequestId, MRef)
    end;
request(Pid, _, _, Msg, _RequestId) ->
    %% No response expected
    gen_server:cast(Pid, {oneway_request, Msg}).

cancel(Pid, RequestId) ->
    gen_server:cast(Pid, {cancel, RequestId}).

cancel(Pid, RequestId, MRef) ->
    gen_server:cast(Pid, {cancel, RequestId, MRef, self()}).

%%-----------------------------------------------------------------
%% Internal interface functions
%%-----------------------------------------------------------------
%%-----------------------------------------------------------------
%% Func: stop/2
%%-----------------------------------------------------------------
stop(Pid, Timeout) ->
    gen_server:call(Pid, stop, Timeout).
stop(Pid) ->
    gen_server:cast(Pid, stop).


%%-----------------------------------------------------------------
%% Server functions
%%-----------------------------------------------------------------
%%-----------------------------------------------------------------
%% Func: init/1
%%-----------------------------------------------------------------
init({connect, Host, Port, SocketType, SocketOptions, Parent, Key, NewKey}) ->
    process_flag(trap_exit, true), 
    case catch orber_socket:connect(SocketType, Host, Port, SocketOptions) of
	{'EXCEPTION', _E} ->
	    ignore;
	%% We used to reply the below but since this would generate a CRASH REPORT
	%% if '-boot start_sasl' used. Due to a request to change this behaviour
	%% we did.
	%% {stop, {'EXCEPTION', E}};
	Socket ->
	    SockData = orber_socket:sockdata(SocketType, Socket),
	    orber_iiop_pm:add_connection(Key, NewKey, SockData),
	    Timeout = orber:iiop_connection_timeout(),
	    {ok, #state{stype = SocketType, socket = Socket,
			db = ets:new(orber_outgoing_requests, [set]),
			timeout = Timeout, client_timeout = orber:iiop_timeout(),
			host = Host, port = Port, parent = Parent}, Timeout}
    end.

%%-----------------------------------------------------------------
%% Func: terminate/2
%%-----------------------------------------------------------------
terminate(_Reason, #state{db = OutRequests, error_reason = ER}) ->
    %% Kill all proxies and delete table before terminating
    notify_clients(OutRequests, ets:first(OutRequests), ER),
    ets:delete(OutRequests),
    ok.

notify_clients(_, '$end_of_table', _ER) ->
    ok;
notify_clients(OutRequests, Key, ER) ->
    case ets:lookup(OutRequests, Key) of
	[{_, Pid, TRef, MRef}] ->
	    cancel_timer(TRef),
	    Pid ! {MRef, ER},
	    notify_clients(OutRequests, ets:next(OutRequests, Key), ER)
    end.

%%-----------------------------------------------------------------
%% Func: handle_call/3
%%-----------------------------------------------------------------
handle_call(stop, _From, State) ->
    {stop, normal, ok, State};
handle_call(X, From, State) ->
    orber:dbg("[~p] orber_iiop_outproxy:handle_call(~p);~n"
	      "Un-recognized call from ~p", [?LINE, X, From], ?DEBUG_LEVEL),
    {noreply, State, State#state.timeout}.

%%-----------------------------------------------------------------
%% Func: handle_cast/2
%%-----------------------------------------------------------------
handle_cast({request, Timeout, Msg, RequestId, From, MRef}, 
	    #state{client_timeout = DefaultTimeout} = State) ->
    orber_socket:write(State#state.stype, State#state.socket, Msg),
    true = ets:insert(State#state.db, {RequestId, From, 
				       start_timer(Timeout, DefaultTimeout, RequestId),
				       MRef}),
    {noreply, State, State#state.timeout};
handle_cast({oneway_request, Msg}, State) ->
    orber_socket:write(State#state.stype, State#state.socket, Msg),
    {noreply, State, State#state.timeout};
handle_cast({cancel, ReqId}, State) ->
    case ets:lookup(State#state.db, ReqId) of
	[{ReqId, _From, TRef, _MRef}] ->
	    cancel_timer(TRef),
	    ets:delete(State#state.db, ReqId),
	    orber:dbg("[~p] orber_iiop_outproxy:handle_info(~p);~n"
		      "Request cancelled", [?LINE, State], ?DEBUG_LEVEL),
	    {noreply, State, State#state.timeout};
	_ ->
	    {noreply, State, State#state.timeout}
    end;
handle_cast({cancel, ReqId, MRef, From}, State) ->
    case ets:lookup(State#state.db, ReqId) of
	[{ReqId, From, TRef, MRef}] ->
	    cancel_timer(TRef),
	    ets:delete(State#state.db, ReqId),
	    From ! {MRef, ReqId, cancelled},
	    orber:dbg("[~p] orber_iiop_outproxy:handle_info(~p); 
Request cancelled", [?LINE, State], ?DEBUG_LEVEL),
	    {noreply, State, State#state.timeout};
	_ ->
	    From ! {MRef, ReqId, cancelled},
	    {noreply, State, State#state.timeout}
    end;
handle_cast(stop, State) ->
    {stop, normal, State};
handle_cast(X, State) ->
    orber:dbg("[~p] orber_iiop_outproxy:handle_cast(~p); 
Un-recognized cast.", [?LINE, X], ?DEBUG_LEVEL),
    {noreply, State, State#state.timeout}.

%%-----------------------------------------------------------------
%% Func: handle_info/2
%%-----------------------------------------------------------------
handle_info({tcp, _Socket, Bytes}, State) ->
    handle_reply(Bytes, State);
handle_info({ssl, _Socket, Bytes}, State) ->
    handle_reply(Bytes, State);
handle_info({tcp_closed, _Socket}, State) ->
    {stop, normal, State}; 
handle_info({ssl_closed, _Socket}, State) ->
    {stop, normal, State};
handle_info({tcp_error, Socket, Reason}, #state{socket = Socket, host = Host,
						port = Port} = State) ->
    orber:error("[~p] IIOP proxy received the TCP error message: ~p~n"
		"The server-side ORB is located at '~p:~p'~n"
		"See the gen_tcp/inet documentation for more information.", 
		[?LINE, Reason, Host, Port], ?DEBUG_LEVEL),
    {stop, normal, State};
handle_info({ssl_error, Socket, Reason}, #state{socket = Socket, host = Host,
						port = Port} = State) ->
    orber:error("[~p] IIOP proxy received the SSL error message: ~p~n"
		"The server-side ORB is located at '~p:~p'~n"
		"See the SSL-application documentation for more information.", 
		[?LINE, Reason, Host, Port], ?DEBUG_LEVEL),
    {stop, normal, State};
handle_info({timeout, _TRef, ReqId}, State) ->
    case ets:lookup(State#state.db, ReqId) of
	[{ReqId, Pid, _, MRef}] ->
	    ets:delete(State#state.db, ReqId),
	    Pid ! {MRef, {'EXCEPTION', #'TIMEOUT'{completion_status=?COMPLETED_MAYBE}}},
	    orber:dbg("[~p] orber_iiop_outproxy:handle_info(~p, ~p);~n"
		      "Request timed out", 
		      [?LINE, State#state.host, State#state.port], ?DEBUG_LEVEL),
	    {noreply, State, State#state.timeout};
	_ ->
	    {noreply, State, State#state.timeout}
    end;
handle_info(stop, State) ->
    {stop, normal, State};
handle_info(timeout, State) ->
    case ets:info(State#state.db, size) of
	0 ->
	    orber:dbg("[~p] orber_iiop_outproxy:handle_info(~p, ~p);~n"
		      "Outgoing connection timed out after ~p msec", 
		      [?LINE, State#state.host, State#state.port,
		       State#state.timeout], ?DEBUG_LEVEL),
	    {stop, normal, State};
	_Amount ->
	    %% Still pending request, cannot close the connection.
	    {noreply, State, State#state.timeout}
    end;
handle_info({'EXIT', Parent, Reason}, #state{parent = Parent} = State) ->
    orber:dbg("[~p] orber_iiop_outproxy:handle_info(~p);~nParent terminated.", 
	      [?LINE, Reason], ?DEBUG_LEVEL),
    {stop, normal, State};
handle_info({reconfigure, _Options}, State) ->
    %% Currently there are no parameters that can be changed.