corbamessagemediatorimpl.java

来自「JAVA 所有包」· Java 代码 · 共 2,135 行 · 第 1/5 页

/*
 * @(#)CorbaMessageMediatorImpl.java	1.99 05/11/17
 * 
 * Copyright 2006 Sun Microsystems, Inc. All rights reserved.
 * SUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 */

package com.sun.corba.se.impl.protocol;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.PrintWriter;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.util.EmptyStackException;
import java.util.Iterator;

import org.omg.CORBA.Any;
import org.omg.CORBA.CompletionStatus;
import org.omg.CORBA.ExceptionList;
import org.omg.CORBA.INTERNAL;
import org.omg.CORBA.Principal;
import org.omg.CORBA.SystemException;
import org.omg.CORBA.TypeCode;
import org.omg.CORBA.UnknownUserException;
import org.omg.CORBA.UNKNOWN;
import org.omg.CORBA.portable.ResponseHandler;
import org.omg.CORBA.portable.UnknownException;
import org.omg.CORBA_2_3.portable.InputStream;
import org.omg.CORBA_2_3.portable.OutputStream;
import org.omg.IOP.ExceptionDetailMessage;
import org.omg.IOP.TAG_RMI_CUSTOM_MAX_STREAM_FORMAT;

import com.sun.corba.se.pept.broker.Broker;
import com.sun.corba.se.pept.encoding.InputObject;
import com.sun.corba.se.pept.encoding.OutputObject;
import com.sun.corba.se.pept.protocol.MessageMediator;
import com.sun.corba.se.pept.protocol.ProtocolHandler;
import com.sun.corba.se.pept.transport.ByteBufferPool;
import com.sun.corba.se.pept.transport.Connection;
import com.sun.corba.se.pept.transport.ContactInfo;
import com.sun.corba.se.pept.transport.EventHandler;

import com.sun.corba.se.spi.ior.IOR;
import com.sun.corba.se.spi.ior.ObjectKey;
import com.sun.corba.se.spi.ior.ObjectKeyTemplate;
import com.sun.corba.se.spi.ior.iiop.GIOPVersion;
import com.sun.corba.se.spi.ior.iiop.IIOPProfileTemplate;
import com.sun.corba.se.spi.ior.iiop.IIOPProfile;
import com.sun.corba.se.spi.ior.iiop.MaxStreamFormatVersionComponent;
import com.sun.corba.se.spi.oa.OAInvocationInfo;
import com.sun.corba.se.spi.oa.ObjectAdapter;
import com.sun.corba.se.spi.orb.ORB;
import com.sun.corba.se.spi.orb.ORBVersionFactory;
import com.sun.corba.se.spi.protocol.CorbaMessageMediator;
import com.sun.corba.se.spi.protocol.CorbaProtocolHandler;
import com.sun.corba.se.spi.protocol.CorbaServerRequestDispatcher;
import com.sun.corba.se.spi.protocol.ForwardException;
import com.sun.corba.se.spi.transport.CorbaConnection;
import com.sun.corba.se.spi.transport.CorbaContactInfo;
import com.sun.corba.se.spi.transport.CorbaResponseWaitingRoom;
import com.sun.corba.se.spi.logging.CORBALogDomains;

import com.sun.corba.se.spi.servicecontext.ORBVersionServiceContext;
import com.sun.corba.se.spi.servicecontext.ServiceContexts;
import com.sun.corba.se.spi.servicecontext.UEInfoServiceContext;
import com.sun.corba.se.spi.servicecontext.MaxStreamFormatVersionServiceContext;
import com.sun.corba.se.spi.servicecontext.SendingContextServiceContext;
import com.sun.corba.se.spi.servicecontext.UnknownServiceContext;

import com.sun.corba.se.impl.corba.RequestImpl;
import com.sun.corba.se.impl.encoding.BufferManagerFactory;
import com.sun.corba.se.impl.encoding.BufferManagerReadStream;
import com.sun.corba.se.impl.encoding.CDRInputObject;
import com.sun.corba.se.impl.encoding.CDROutputObject;
import com.sun.corba.se.impl.encoding.EncapsOutputStream;
import com.sun.corba.se.impl.logging.ORBUtilSystemException;
import com.sun.corba.se.impl.logging.InterceptorsSystemException;
import com.sun.corba.se.impl.orbutil.ORBConstants;
import com.sun.corba.se.impl.orbutil.ORBUtility;
import com.sun.corba.se.impl.ior.iiop.JavaSerializationComponent;
import com.sun.corba.se.impl.protocol.AddressingDispositionException;
import com.sun.corba.se.impl.protocol.RequestCanceledException;
import com.sun.corba.se.impl.protocol.giopmsgheaders.AddressingDispositionHelper;
import com.sun.corba.se.impl.protocol.giopmsgheaders.CancelRequestMessage;
import com.sun.corba.se.impl.protocol.giopmsgheaders.FragmentMessage_1_1;
import com.sun.corba.se.impl.protocol.giopmsgheaders.FragmentMessage_1_2;
import com.sun.corba.se.impl.protocol.giopmsgheaders.LocateRequestMessage;
import com.sun.corba.se.impl.protocol.giopmsgheaders.LocateRequestMessage_1_0;
import com.sun.corba.se.impl.protocol.giopmsgheaders.LocateRequestMessage_1_1;
import com.sun.corba.se.impl.protocol.giopmsgheaders.LocateRequestMessage_1_2;
import com.sun.corba.se.impl.protocol.giopmsgheaders.LocateReplyOrReplyMessage;
import com.sun.corba.se.impl.protocol.giopmsgheaders.LocateReplyMessage;
import com.sun.corba.se.impl.protocol.giopmsgheaders.LocateReplyMessage_1_0;
import com.sun.corba.se.impl.protocol.giopmsgheaders.LocateReplyMessage_1_1;
import com.sun.corba.se.impl.protocol.giopmsgheaders.LocateReplyMessage_1_2;
import com.sun.corba.se.impl.protocol.giopmsgheaders.Message;
import com.sun.corba.se.impl.protocol.giopmsgheaders.MessageBase;
import com.sun.corba.se.impl.protocol.giopmsgheaders.MessageHandler;
import com.sun.corba.se.impl.protocol.giopmsgheaders.ReplyMessage;
import com.sun.corba.se.impl.protocol.giopmsgheaders.ReplyMessage_1_0;
import com.sun.corba.se.impl.protocol.giopmsgheaders.ReplyMessage_1_1;
import com.sun.corba.se.impl.protocol.giopmsgheaders.ReplyMessage_1_2;
import com.sun.corba.se.impl.protocol.giopmsgheaders.RequestMessage;
import com.sun.corba.se.impl.protocol.giopmsgheaders.RequestMessage_1_0 ;
import com.sun.corba.se.impl.protocol.giopmsgheaders.RequestMessage_1_1 ;
import com.sun.corba.se.impl.protocol.giopmsgheaders.RequestMessage_1_2 ;

// REVISIT: make sure no memory leaks in client/server request/reply maps.
// REVISIT: normalize requestHeader, replyHeader, messageHeader.

/**
 * @author Harold Carr
 */
public class CorbaMessageMediatorImpl
    implements 
	CorbaMessageMediator,
	CorbaProtocolHandler,
	MessageHandler
{
    protected ORB orb;
    protected ORBUtilSystemException wrapper ;
    protected InterceptorsSystemException interceptorWrapper ;
    protected CorbaContactInfo contactInfo;
    protected CorbaConnection connection;
    protected short addrDisposition;
    protected CDROutputObject outputObject;
    protected CDRInputObject inputObject;
    protected Message messageHeader;
    protected RequestMessage requestHeader;
    protected LocateReplyOrReplyMessage replyHeader;
    protected String replyExceptionDetailMessage;
    protected IOR replyIOR;
    protected Integer requestIdInteger;
    protected Message dispatchHeader;
    protected ByteBuffer dispatchByteBuffer;
    protected byte streamFormatVersion;
    protected boolean streamFormatVersionSet = false;

    protected org.omg.CORBA.Request diiRequest;

    protected boolean cancelRequestAlreadySent = false;

    protected ProtocolHandler protocolHandler;
    protected boolean _executeReturnServantInResponseConstructor = false;
    protected boolean _executeRemoveThreadInfoInResponseConstructor = false;
    protected boolean _executePIInResponseConstructor = false;

    //
    // Client-side constructor.
    //

    public CorbaMessageMediatorImpl(ORB orb,
				    ContactInfo contactInfo,
				    Connection connection,
				    GIOPVersion giopVersion,
				    IOR ior,
				    int requestId,
				    short addrDisposition,
				    String operationName,
				    boolean isOneWay)
    {
	this( orb, connection ) ;
	    
	this.contactInfo = (CorbaContactInfo) contactInfo;
	this.addrDisposition = addrDisposition;

	streamFormatVersion =
	    getStreamFormatVersionForThisRequest(
	        ((CorbaContactInfo)this.contactInfo).getEffectiveTargetIOR(),
		giopVersion);
	streamFormatVersionSet = true;

	requestHeader = (RequestMessage) MessageBase.createRequest(
            this.orb,
	    giopVersion,
	    ORBUtility.getEncodingVersion(orb, ior),
	    requestId,
	    !isOneWay,
	    ((CorbaContactInfo)this.contactInfo).getEffectiveTargetIOR(),
	    this.addrDisposition,
	    operationName,
	    new ServiceContexts(orb),
	    null);
    }

    //
    // Acceptor constructor.
    //

    public CorbaMessageMediatorImpl(ORB orb,
				    Connection connection) 
    {
	this.orb = orb;
	this.connection = (CorbaConnection)connection;
	this.wrapper = ORBUtilSystemException.get( orb,
	    CORBALogDomains.RPC_PROTOCOL ) ;
	this.interceptorWrapper = InterceptorsSystemException.get( orb,
	    CORBALogDomains.RPC_PROTOCOL ) ;
    }

    //
    // Dispatcher constructor.
    //

    // Note: in some cases (e.g., a reply message) this message 
    // mediator will only be used for dispatch.  Then the original 
    // request side mediator will take over. 
    public CorbaMessageMediatorImpl(ORB orb,
				    CorbaConnection connection,
				    Message dispatchHeader,
				    ByteBuffer byteBuffer)
    {
	this( orb, connection ) ;
	this.dispatchHeader = dispatchHeader;
        this.dispatchByteBuffer = byteBuffer;
    }

    ////////////////////////////////////////////////////
    //
    // MessageMediator
    //

    public Broker getBroker() 
    {
	return orb;
    }

    public ContactInfo getContactInfo()
    {
	return contactInfo;
    }

    public Connection getConnection()
    {
	return connection;
    }

    public void initializeMessage() 
    {
	getRequestHeader().write(outputObject);
    }

    public void finishSendingRequest() 
    {
	// REVISIT: probably move logic in outputObject to here.
        outputObject.finishSendingMessage();
    }

    public InputObject waitForResponse()
    {
	if (getRequestHeader().isResponseExpected()) {
	    return connection.waitForResponse(this);
	}
	return null;
    }

    public void setOutputObject(OutputObject outputObject)
    {
	this.outputObject = (CDROutputObject) outputObject;
    }

    public OutputObject getOutputObject()
    {
	return outputObject;
    }

    public void setInputObject(InputObject inputObject)
    {
	this.inputObject = (CDRInputObject) inputObject;
    }

    public InputObject getInputObject()
    {
	return inputObject;
    }

    ////////////////////////////////////////////////////
    //
    // CorbaMessageMediator
    //

    public void setReplyHeader(LocateReplyOrReplyMessage header)
    {
	this.replyHeader = header;
	this.replyIOR = header.getIOR(); // REVISIT - need separate field?
    }

    public LocateReplyMessage getLocateReplyHeader()
    {
	return (LocateReplyMessage) replyHeader;
    }
    
    public ReplyMessage getReplyHeader()
    {
	return (ReplyMessage) replyHeader;
    }

    public void setReplyExceptionDetailMessage(String message)
    {
	replyExceptionDetailMessage = message;
    }
    
    public RequestMessage getRequestHeader()
    {
	return requestHeader;
    }
    
    public GIOPVersion getGIOPVersion()
    {
	if (messageHeader != null) {
	    return messageHeader.getGIOPVersion();
	}
	return getRequestHeader().getGIOPVersion();
    }

    public byte getEncodingVersion() {
	if (messageHeader != null) {
	    return messageHeader.getEncodingVersion();
	}
	return getRequestHeader().getEncodingVersion();
    }

    public int getRequestId()
    {
	return getRequestHeader().getRequestId();
    }

    public Integer getRequestIdInteger()
    {
	if (requestIdInteger == null) {
	    requestIdInteger = new Integer(getRequestHeader().getRequestId());
	}
	return requestIdInteger;
    }

    public boolean isOneWay()
    {
	return ! getRequestHeader().isResponseExpected();
    }

    public short getAddrDisposition()
    {
	return addrDisposition;
    }

    public String getOperationName()
    {
	return getRequestHeader().getOperation();
    }

    public ServiceContexts getRequestServiceContexts()
    {
	return getRequestHeader().getServiceContexts();
    }

    public ServiceContexts getReplyServiceContexts()
    {
	return getReplyHeader().getServiceContexts();
    }

    public void sendCancelRequestIfFinalFragmentNotSent()
    {
	if ((!sentFullMessage()) && sentFragment() && 
	    (!cancelRequestAlreadySent))
        {
	    try {
		if (orb.subcontractDebugFlag) {
		    dprint(".sendCancelRequestIfFinalFragmentNotSent->: " 
			   + opAndId(this));
		}
		connection.sendCancelRequestWithLock(getGIOPVersion(),
						     getRequestId());
		// Case: first a location forward, then a marshaling 
		// exception (e.g., non-serializable object).  Only
		// send cancel once.
		cancelRequestAlreadySent = true;
	    } catch (IOException e) {
		if (orb.subcontractDebugFlag) {
		    dprint(".sendCancelRequestIfFinalFragmentNotSent: !ERROR : " + opAndId(this),
			   e);
		}

		// REVISIT: we could attempt to send a final incomplete
		// fragment in this case.
		throw interceptorWrapper.ioexceptionDuringCancelRequest(
		    CompletionStatus.COMPLETED_MAYBE, e );
	    } finally {
		if (orb.subcontractDebugFlag) {
		    dprint(".sendCancelRequestIfFinalFragmentNotSent<-: "
			   + opAndId(this));
		}
	    }
	}
    }

    public boolean sentFullMessage()
    {
	return outputObject.getBufferManager().sentFullMessage();
    }

    public boolean sentFragment()
    {
	return outputObject.getBufferManager().sentFragment();
    }

    public void setDIIInfo(org.omg.CORBA.Request diiRequest)
    {
	this.diiRequest = diiRequest;
    }

    public boolean isDIIRequest()
    {
	return diiRequest != null;
    }

    public Exception unmarshalDIIUserException(String repoId, InputStream is)
    {
	if (! isDIIRequest()) {
	    return null;
	}

	ExceptionList _exceptions = diiRequest.exceptions();

	try {
	    // Find the typecode for the exception
	    for (int i=0; i<_exceptions.count() ; i++) {