orber_iiop_inproxy.erl

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

    Timeout = orber_env:iiop_ssl_accept_timeout(),
    case catch orber_socket:post_accept(Type, Socket, Timeout) of
	ok ->
	    {noreply, State};
	_Failed ->
	    orber_socket:close(Type, Socket),
	    {stop, normal, State}
    end;
handle_cast(stop, State) ->
    {stop, normal, State};
handle_cast(_, State) ->
    {noreply, State, State#state.timeout}.

%%-----------------------------------------------------------------
%% Func: handle_info/2
%%-----------------------------------------------------------------
%% Normal invocation
handle_info({tcp, Socket, Bytes}, State) ->
    handle_msg(normal, Socket, Bytes, State);
handle_info({ssl, Socket, Bytes}, State) ->
    handle_msg(ssl, Socket, Bytes, State);
%% Errors, closed connection
handle_info({tcp_closed, _Socket}, State) ->
    {stop, normal, State};
handle_info({tcp_error, _Socket, _Reason}, State) ->
    {stop, normal, State};
handle_info({ssl_closed, _Socket}, State) ->
    {stop, normal, State};
handle_info({ssl_error, _Socket, _Reason}, State) ->
    {stop, normal, State};
%% Servant termination.
handle_info({'EXIT', Pid, normal}, State) ->
    ets:delete(State#state.db, Pid),
    {noreply, decrease_counter(State), State#state.timeout};
handle_info({message_error, _Pid, ReqId}, State) ->
    ets:delete(State#state.db, ReqId),
    {noreply, State, State#state.timeout};
handle_info(timeout, State) ->
    case ets:info(State#state.db, size) of
	0 ->
	    %% No pending requests, close the connection.
	    {stop, normal, State};
	_Amount ->
	    %% Still pending request, cannot close the connection.
	    {noreply, State, State#state.timeout}
    end;
handle_info({reconfigure, Options}, State) ->
    {noreply, update_state(State, Options), State#state.timeout};
handle_info(_X,State) ->
    {noreply, State, State#state.timeout}.

handle_msg(Type, Socket, Bytes, #state{stype = Type, socket = Socket, 
				       giop_env = Env} = State) ->
    case catch cdr_decode:dec_giop_message_header(Bytes) of
	%% Only when using IIOP-1.2 may the client send this message. 
	%% Introduced in CORBA-2.6
	#giop_message{message_type = ?GIOP_MSG_CLOSE_CONNECTION, 
		      giop_version = {1,2}} ->
	    {stop, normal, State};
	#giop_message{message_type = ?GIOP_MSG_CLOSE_CONNECTION} ->
	    {noreply, State, State#state.timeout};
	#giop_message{message_type = ?GIOP_MSG_CANCEL_REQUEST} = GIOPHdr ->
	    ReqId = cdr_decode:peek_request_id(GIOPHdr#giop_message.byte_order,
					       GIOPHdr#giop_message.message),
	    case ets:lookup(State#state.db, ReqId) of
		[{RId, PPid}] ->
		    ets:delete(State#state.db, RId),
		    PPid ! {self(), cancel_request_header};
		[] ->
		    send_msg_error(Type, Socket, Bytes, 
				   Env#giop_env{version = 
						GIOPHdr#giop_message.giop_version},
				   "No such request id")
	    end,
	    {noreply, State, State#state.timeout};
	%% A fragment; we must have received a Request or LocateRequest
	%% with fragment-flag set to true.
	%% We need to decode the header to get the request-id.
	#giop_message{message_type = ?GIOP_MSG_FRAGMENT,
		      giop_version = {1,2}} = GIOPHdr ->
	    ReqId = cdr_decode:peek_request_id(GIOPHdr#giop_message.byte_order,
					       GIOPHdr#giop_message.message),
	    case ets:lookup(State#state.db, ReqId) of
		[{_RId, PPid}] when GIOPHdr#giop_message.fragments == true ->
		    PPid ! {self(), GIOPHdr};
		[{RId, PPid}] ->
		    ets:delete(State#state.db, RId),
		    PPid ! {self(), GIOPHdr};
		[] ->
		    send_msg_error(Type, Socket, Bytes, 
				   Env#giop_env{version = 
						GIOPHdr#giop_message.giop_version},
				   "No such fragment id")
	    end,
	    {noreply, State, State#state.timeout};
	%% Must be a Request or LocateRequest which have been fragmented.
	%% We need to decode the header to get the request-id.
	#giop_message{fragments = true,
		      giop_version = {1,2}} = GIOPHdr ->
	    ReqId = cdr_decode:peek_request_id(GIOPHdr#giop_message.byte_order,
					       GIOPHdr#giop_message.message),
	    Pid = 
		orber_iiop_inrequest:
		start_fragment_collector(GIOPHdr, Bytes, 
					 Type, Socket, 
					 ReqId, self(),
					 State#state.max_fragments,
					 Env#giop_env{version = {1,2},
						      request_id = ReqId}),
	    ets:insert(State#state.db, {Pid, ReqId}),
	    ets:insert(State#state.db, {ReqId, Pid}),
	    {noreply, increase_counter(State), State#state.timeout};
	GIOPHdr when record(GIOPHdr, giop_message) ->
	    Pid = orber_iiop_inrequest:start(GIOPHdr, Bytes, Type, Socket, 
					     Env#giop_env{version = 
							  GIOPHdr#giop_message.giop_version}),
	    ets:insert(State#state.db, {Pid, undefined}),
	    {noreply, increase_counter(State), State#state.timeout};
	{'EXIT', message_error} ->
	    send_msg_error(Type, Socket, Bytes, 
			   Env#giop_env{version = orber_env:giop_version()},
			   "Unable to decode the GIOP-header"),
	    {noreply, State, State#state.timeout}
    end;
handle_msg(Type, _, Bytes, State) ->
    orber:dbg("[~p] orber_iiop_inproxy:handle_msg(~p);~n"
	      "Received a message from a socket of a different type.~n"
	      "Should be ~p but was ~p.", 
	      [?LINE, Bytes, State#state.stype, Type], ?DEBUG_LEVEL),
    {noreply, State, State#state.timeout}.

send_msg_error(Type, Socket, Data, Env, Msg) ->
    orber:dbg("[~p] orber_iiop_inproxy:handle_msg(~p); ~p.", 
	      [?LINE, Data, Msg], ?DEBUG_LEVEL),
    Reply = cdr_encode:enc_message_error(Env),
    orber_socket:write(Type, Socket, Reply).

increase_counter(#state{max_requests = infinity} = State) ->
    State;
increase_counter(#state{max_requests = Max, 
			request_counter = Counter} = State) when Max > Counter ->
    orber_socket:setopts(State#state.stype, State#state.socket, [{active, once}]),
    State#state{request_counter = Counter + 1};
increase_counter(State) ->
    State#state{request_counter = State#state.request_counter + 1}.

decrease_counter(#state{max_requests = infinity} = State) ->
    State;
decrease_counter(#state{max_requests = Max, 
			request_counter = Counter} = State) when Max =< Counter ->
    orber_socket:setopts(State#state.stype, State#state.socket, [{active, once}]),
    State#state{request_counter = Counter - 1};
decrease_counter(State) ->
    State#state{request_counter = State#state.request_counter - 1}.

update_state(#state{giop_env = Env} = State, 
	     [{interceptors, false}|Options]) ->
    update_state(State#state{giop_env = 
			     Env#giop_env{interceptors = false}}, Options);
update_state(#state{giop_env = #giop_env{interceptors = false, host = [SH], 
					 iiop_port = SP} = Env, 
		    peer = {PH, PP}, stype = normal} = State, 
	     [{interceptors, {native, LPIs}}|Options]) ->
    %% No Interceptor(s). Add the same Ref used by the built in interceptors.
    update_state(State#state{giop_env = 
			     Env#giop_env{interceptors = 
					  {native, {PH, PP, SH, SP}, LPIs}}},
		 Options);
update_state(#state{giop_env = #giop_env{interceptors = false, host = [SH], 
					 iiop_ssl_port = SP} = Env, 
		    peer = {PH, PP}, stype = ssl} = State, 
	     [{interceptors, {native, LPIs}}|Options]) ->
    %% No Interceptor(s). Add the same Ref used by the built in interceptors.
    update_state(State#state{giop_env = 
			     Env#giop_env{interceptors = 
					  {native, {PH, PP, SH, SP}, LPIs}}},
		 Options);
update_state(#state{giop_env = #giop_env{interceptors = {native, Ref, _}} = Env} = 
	     State, 
	     [{interceptors, {native, LPIs}}|Options]) ->
    %% Interceptor(s) already in use. We must use the same Ref as before.
    update_state(State#state{giop_env = 
			     Env#giop_env{interceptors = {native, Ref, LPIs}}},
		 Options);
update_state(State, [H|T]) ->
    orber:dbg("[~p] orber_iiop_inproxy:update_state(~p, ~p)~n"
	      "Couldn't change the state.", 
	      [?LINE, H, State], ?DEBUG_LEVEL),
    update_state(State, T);
update_state(State, []) ->
    State.

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