orber_iiop_outproxy.erl

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

    {noreply, State, State#state.timeout};
handle_info(X, State) ->
    orber:dbg("[~p] orber_iiop_outproxy:handle_info(~p);~nUn-recognized info.", 
	      [?LINE, X], ?DEBUG_LEVEL),
    {noreply, State, State#state.timeout}.


handle_reply(Bytes, State) ->
    %% Check IIOP headers and fetch request id
    case catch checkheaders(cdr_decode:dec_giop_message_header(Bytes)) of
	{'reply', ReplyHeader, Rest, Len, ByteOrder} ->
	    case ets:lookup(State#state.db, ReplyHeader#reply_header.request_id) of
		[{_, Pid, TRef, MRef}] ->
		    %% Send reply to the correct request process
		    cancel_timer(TRef),
		    Pid ! {MRef, {reply, ReplyHeader, Rest, Len, ByteOrder, Bytes}},
		    ets:delete(State#state.db, ReplyHeader#reply_header.request_id),
		    {noreply, State, State#state.timeout};
		_ ->
		    {noreply, State, State#state.timeout}
	    end;
	{'locate_reply', LocateReplyHeader, LocateRest, LocateLen, LocateByteOrder} ->
	    case ets:lookup(State#state.db, 
			    LocateReplyHeader#locate_reply_header.request_id) of
		[{_, Pid, TRef, MRef}] ->
		    %% Send reply to the correct request process
		    cancel_timer(TRef),
		    Pid ! {MRef, {locate_reply, LocateReplyHeader, 
				  LocateRest, LocateLen, LocateByteOrder}},
		    ets:delete(State#state.db, 
			       LocateReplyHeader#locate_reply_header.request_id),
		    {noreply, State, State#state.timeout};
		_ ->
		    {noreply, State, State#state.timeout}
	    end;
	{fragment, GIOPHdr, ReqId, false} ->
	    %% Last fragment, cancel timer and remove from DB.
	    case ets:lookup(State#state.db, ReqId) of
		[{_, Pid, TRef, MRef}] ->
		    cancel_timer(TRef),
		    Pid ! {fragment, GIOPHdr, ReqId, MRef}, 
		    ets:delete(State#state.db, ReqId),
		    {noreply, State, State#state.timeout};
		_ ->
		    %% Probably cancelled
		    {noreply, State, State#state.timeout}
	    end;
	{fragment, GIOPHdr, ReqId, _} ->
	    %% More fragments expected
	    case ets:lookup(State#state.db, ReqId) of
		[{_, Pid, _, MRef}] ->
		    Pid ! {fragment, GIOPHdr, ReqId, MRef}, 
		    {noreply, State, State#state.timeout};
		_ ->
		    %% Probably cancelled
		    {noreply, State, State#state.timeout}
	    end;
	{fragmented, GIOPHdr, ReqId} ->
	    %% This the initial message (i.e. a LocateReply or Reply).
	    case ets:lookup(State#state.db, ReqId) of
		[{_, Pid, _TRef, MRef}] ->
		    Pid ! {fragmented, GIOPHdr, Bytes, ReqId, MRef},
		    {noreply, State, State#state.timeout};
		_ ->
		    {noreply, State, State#state.timeout}
	    end;
	{'EXCEPTION', DecodeException} ->
	    orber:dbg("[~p] orber_iiop_outproxy:handle_reply(~p); decode exception(~p).", 
		      [?LINE, Bytes, DecodeException], ?DEBUG_LEVEL),
	    {noreply, State, State#state.timeout};
	{'EXIT', message_error} ->
	    orber:dbg("[~p] orber_iiop_outproxy:handle_reply(~p); message error.", 
		      [?LINE, Bytes], ?DEBUG_LEVEL),
	    ME = cdr_encode:enc_message_error(#giop_env{version = 
							orber:giop_version()}),
	    orber_socket:write(State#state.stype, State#state.socket, ME),
	    {noreply, State, State#state.timeout};
	{'EXIT', R} ->
	    orber:dbg("[~p] orber_iiop_outproxy:handle_reply(~p); got exit(~p)", 
		      [?LINE, Bytes, R], ?DEBUG_LEVEL),
	    {noreply, State, State#state.timeout};
	close_connection ->
	    orber:dbg("[~p] orber_iiop_outproxy:handle_reply();
The Server-side ORB closed the connection.", [?LINE], ?DEBUG_LEVEL),
	    {stop, normal, State};
	{error, no_reply} ->
	    {noreply, State, State#state.timeout};
	X ->
	    orber:dbg("[~p] orber_iiop_outproxy:handle_reply(~p); message error(~p).", 
		      [?LINE, Bytes, X], ?DEBUG_LEVEL),
	    {noreply, State, State#state.timeout}
    end.


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

%%-----------------------------------------------------------------
%% Internal functions
%%-----------------------------------------------------------------
checkheaders(#giop_message{message_type = ?GIOP_MSG_CLOSE_CONNECTION}) ->
    close_connection;
checkheaders(#giop_message{message_type = ?GIOP_MSG_FRAGMENT,
			   giop_version = {1,2},
			   fragments = MoreFrag} = GIOPHdr) ->
    %% 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.
    ReqId = cdr_decode:peek_request_id(GIOPHdr#giop_message.byte_order,
				       GIOPHdr#giop_message.message),
    {fragment, GIOPHdr, ReqId, MoreFrag};
checkheaders(#giop_message{fragments = true,
			   giop_version = {1,2}} = GIOPHdr) ->
    %% Must be a Reply or LocateReply which have been fragmented.
    %% We need to decode the header to get the request-id.
    ReqId = cdr_decode:peek_request_id(GIOPHdr#giop_message.byte_order,
				       GIOPHdr#giop_message.message),
    {fragmented, GIOPHdr, ReqId};
checkheaders(#giop_message{fragments = false, 
			   message_type = ?GIOP_MSG_REPLY} = GIOPHdr) ->
    {ReplyHeader, Rest, Len} = 
	cdr_decode:dec_reply_header(GIOPHdr#giop_message.giop_version,
				    GIOPHdr#giop_message.message, 
				    ?GIOP_HEADER_SIZE,
				    GIOPHdr#giop_message.byte_order),
    {'reply', ReplyHeader, Rest, Len, GIOPHdr#giop_message.byte_order};
checkheaders(#giop_message{fragments = false, 
			   message_type = ?GIOP_MSG_LOCATE_REPLY} = GIOPHdr) ->
    {LocateReplyHeader, Rest, Len} = 
	cdr_decode:dec_locate_reply_header(GIOPHdr#giop_message.giop_version,
					   GIOPHdr#giop_message.message, 
					   ?GIOP_HEADER_SIZE,
					   GIOPHdr#giop_message.byte_order),
    {'locate_reply', LocateReplyHeader, Rest, Len, GIOPHdr#giop_message.byte_order};
checkheaders(What) ->
    orber:dbg("[~p] orber_iiop_outproxy:checkheaders(~p)
Un-recognized GIOP header.", [?LINE, What], ?DEBUG_LEVEL),
    {error, no_reply}.


cancel_timer(infinity) -> 
    ok;
cancel_timer(TRef) ->
    erlang:cancel_timer(TRef).

start_timer(infinity, infinity, _) -> 
    infinity;
start_timer(infinity, Timeout, RequestId) ->
    erlang:start_timer(Timeout, self(), RequestId);
start_timer(Timeout, _, RequestId) ->
    erlang:start_timer(Timeout, self(), RequestId).



collect_fragments(GIOPHdr1, InBuffer, Bytes, Proxy, RequestId, MRef) ->
    receive
	%% There are more framents to come; just collect this message and wait for
	%% the rest.
	{fragment, #giop_message{byte_order = _ByteOrder,
				 message    = Message,
				 fragments  = true} = GIOPHdr2, RequestId, MRef} ->
	    case catch cdr_decode:dec_message_header(null, GIOPHdr2, Message) of
		{_, #fragment_header{}, FragBody, _, _} ->
		    collect_fragments(GIOPHdr1, [FragBody|InBuffer], 
				      Bytes, Proxy, RequestId, MRef);
		Other ->
		    cancel(Proxy, RequestId, MRef),
		    clear_queue(Proxy, RequestId, MRef),
		    orber:dbg("[~p] orber_iiop:collect_fragments(~p)", 
			      [?LINE, Other], ?DEBUG_LEVEL),
		    corba:raise(#'MARSHAL'{minor=(?ORBER_VMCID bor 18), 
					   completion_status=?COMPLETED_YES})
	    end;
	%% This is the last fragment. Now we can but together the fragments, decode
	%% the reply and send it to the client.
	{fragment, #giop_message{byte_order = ByteOrder,
				 message    = Message} = GIOPHdr2, RequestId, MRef} ->
	    erlang:demonitor(MRef),
            receive 
                {'DOWN', MRef, _, _, _} -> 
		    ok
            after 0 -> 
		    ok
            end,
	    case catch cdr_decode:dec_message_header(null, GIOPHdr2, Message) of
		{_, #fragment_header{}, FragBody, _, _} ->
		    %% This buffer is all the fragments concatenated.
		    Buffer = lists:reverse([FragBody|InBuffer]),
		    
		    %% Create a GIOP-message which is exactly as if hadn't been fragmented.
		    NewGIOP = GIOPHdr1#giop_message
				{message = list_to_binary([GIOPHdr1#giop_message.message|Buffer]),
				 fragments = false},
		    case checkheaders(NewGIOP) of
			{'reply', ReplyHeader, Rest, Len, ByteOrder} ->
			    %% We must keep create a copy of all bytes, as if the 
			    %% message wasn't fragmented, to be able handle TypeCode
			    %% indirection.
			    {'reply', ReplyHeader, Rest, Len, ByteOrder, 
			     list_to_binary([Bytes|Buffer])};
			{'locate_reply', ReplyHdr, Rest, Len, ByteOrder} ->
			    {'locate_reply', ReplyHdr, Rest, Len, ByteOrder};
			Error ->
			    orber:dbg("[~p] orber_iiop:collect_fragments(~p, ~p);
Unable to decode Reply or LocateReply header",[?LINE, NewGIOP, Error], ?DEBUG_LEVEL),
			    corba:raise(#'MARSHAL'{minor=(?ORBER_VMCID bor 18),
						   completion_status=?COMPLETED_YES})
		    end;
		Other ->
		    orber:dbg("[~p] orber_iiop:collect_fragments(~p);", 
			      [?LINE, Other], ?DEBUG_LEVEL),
		    corba:raise(#'MARSHAL'{minor=(?ORBER_VMCID bor 18),
					   completion_status=?COMPLETED_YES})
	    end;
	{MRef, {'EXCEPTION', E}} ->
	    orber:dbg("[~p] orber_iiop:collect_fragments(~p);", 
		      [?LINE, E], ?DEBUG_LEVEL),
	    erlang:demonitor(MRef),
            receive 
                {'DOWN', MRef, _, _, _} -> 
		    corba:raise(E)
            after 0 -> 
		    corba:raise(E)
            end;
	{'DOWN', MRef, _, Proxy, Reason} when pid(Proxy) ->
	    orber:dbg("[~p] orber_iiop:collect_fragments(~p);~n"
		      "Monitor generated a DOWN message.", 
		      [?LINE, Reason], ?DEBUG_LEVEL),
            receive
		%% Clear EXIT message from queue
                {'EXIT', _Proxy, _What} -> 
                    corba:raise(#'COMM_FAILURE'{completion_status=?COMPLETED_MAYBE})
            after 0 ->
                    corba:raise(#'COMM_FAILURE'{completion_status=?COMPLETED_MAYBE})
            end
    end.

clear_queue(Proxy, RequestId, MRef) ->
    receive 
	{fragment, _, RequestId, MRef} ->
	    clear_queue(Proxy, RequestId, MRef);
	{MRef, RequestId, cancelled} ->
	    %% This is the last message that the proxy will send
	    %% after we've cancelled the request.
	    erlang:demonitor(MRef),
	    receive 
		{'DOWN', MRef, _, _, _} -> 
		    ok
	    after 0 -> 
		    ok
	    end;
	{'DOWN', MRef, _, Proxy, _Reason} ->
	    %% The proxy terminated. Clear EXIT message from queue
            receive
                {'EXIT', Proxy, _What} -> 
                    ok
            after 0 ->
                    ok
            end
    end.