| /* |
| * Copyright (c) 2001, 2013, Oracle and/or its affiliates. All rights reserved. |
| * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. |
| * |
| * This code is free software; you can redistribute it and/or modify it |
| * under the terms of the GNU General Public License version 2 only, as |
| * published by the Free Software Foundation. Oracle designates this |
| * particular file as subject to the "Classpath" exception as provided |
| * by Oracle in the LICENSE file that accompanied this code. |
| * |
| * This code is distributed in the hope that it will be useful, but WITHOUT |
| * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or |
| * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License |
| * version 2 for more details (a copy is included in the LICENSE file that |
| * accompanied this code). |
| * |
| * You should have received a copy of the GNU General Public License version |
| * 2 along with this work; if not, write to the Free Software Foundation, |
| * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. |
| * |
| * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA |
| * or visit www.oracle.com if you need additional information or have any |
| * questions. |
| */ |
| |
| 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++) { |
| TypeCode tc = _exceptions.item(i); |
| if ( tc.id().equals(repoId) ) { |
| // Since we dont have the actual user exception |
| // class, the spec says we have to create an |
| // UnknownUserException and put it in the |
| // environment. |
| Any eany = orb.create_any(); |
| eany.read_value(is, (TypeCode)tc); |
| |
| return new UnknownUserException(eany); |
| } |
| } |
| } catch (Exception b) { |
| throw wrapper.unexpectedDiiException(b); |
| } |
| |
| // must be a truly unknown exception |
| return wrapper.unknownCorbaExc( CompletionStatus.COMPLETED_MAYBE); |
| } |
| |
| public void setDIIException(Exception exception) |
| { |
| diiRequest.env().exception(exception); |
| } |
| |
| public void handleDIIReply(InputStream inputStream) |
| { |
| if (! isDIIRequest()) { |
| return; |
| } |
| ((RequestImpl)diiRequest).unmarshalReply(inputStream); |
| } |
| |
| public Message getDispatchHeader() |
| { |
| return dispatchHeader; |
| } |
| |
| public void setDispatchHeader(Message msg) |
| { |
| dispatchHeader = msg; |
| } |
| |
| public ByteBuffer getDispatchBuffer() |
| { |
| return dispatchByteBuffer; |
| } |
| |
| public void setDispatchBuffer(ByteBuffer byteBuffer) |
| { |
| dispatchByteBuffer = byteBuffer; |
| } |
| |
| public int getThreadPoolToUse() { |
| int poolToUse = 0; |
| Message msg = getDispatchHeader(); |
| // A null msg should never happen. But, we'll be |
| // defensive just in case. |
| if (msg != null) { |
| poolToUse = msg.getThreadPoolToUse(); |
| } |
| return poolToUse; |
| } |
| |
| public byte getStreamFormatVersion() |
| { |
| // REVISIT: ContactInfo/Acceptor output object factories |
| // just use this. Maybe need to distinguish: |
| // createOutputObjectForRequest |
| // createOutputObjectForReply |
| // then do getStreamFormatVersionForRequest/ForReply here. |
| if (streamFormatVersionSet) { |
| return streamFormatVersion; |
| } |
| return getStreamFormatVersionForReply(); |
| } |
| |
| /** |
| * If the RMI-IIOP maximum stream format version service context |
| * is present, it indicates the maximum stream format version we |
| * could use for the reply. If it isn't present, the default is |
| * 2 for GIOP 1.3 or greater, 1 for lower. |
| * |
| * This is only sent on requests. Clients can find out the |
| * server's maximum by looking for a tagged component in the IOR. |
| */ |
| public byte getStreamFormatVersionForReply() { |
| |
| // NOTE: The request service contexts may indicate the max. |
| ServiceContexts svc = getRequestServiceContexts(); |
| |
| MaxStreamFormatVersionServiceContext msfvsc |
| = (MaxStreamFormatVersionServiceContext)svc.get( |
| MaxStreamFormatVersionServiceContext.SERVICE_CONTEXT_ID); |
| |
| if (msfvsc != null) { |
| byte localMaxVersion = ORBUtility.getMaxStreamFormatVersion(); |
| byte remoteMaxVersion = msfvsc.getMaximumStreamFormatVersion(); |
| |
| return (byte)Math.min(localMaxVersion, remoteMaxVersion); |
| } else { |
| // Defaults to 1 for GIOP 1.2 or less, 2 for |
| // GIOP 1.3 or higher. |
| if (getGIOPVersion().lessThan(GIOPVersion.V1_3)) |
| return ORBConstants.STREAM_FORMAT_VERSION_1; |
| else |
| return ORBConstants.STREAM_FORMAT_VERSION_2; |
| } |
| } |
| |
| public boolean isSystemExceptionReply() |
| { |
| return replyHeader.getReplyStatus() == ReplyMessage.SYSTEM_EXCEPTION; |
| } |
| |
| public boolean isUserExceptionReply() |
| { |
| return replyHeader.getReplyStatus() == ReplyMessage.USER_EXCEPTION; |
| } |
| |
| public boolean isLocationForwardReply() |
| { |
| return ( (replyHeader.getReplyStatus() == ReplyMessage.LOCATION_FORWARD) || |
| (replyHeader.getReplyStatus() == ReplyMessage.LOCATION_FORWARD_PERM) ); |
| //return replyHeader.getReplyStatus() == ReplyMessage.LOCATION_FORWARD; |
| } |
| |
| public boolean isDifferentAddrDispositionRequestedReply() |
| { |
| return replyHeader.getReplyStatus() == ReplyMessage.NEEDS_ADDRESSING_MODE; |
| } |
| |
| public short getAddrDispositionReply() |
| { |
| return replyHeader.getAddrDisposition(); |
| } |
| |
| public IOR getForwardedIOR() |
| { |
| return replyHeader.getIOR(); |
| } |
| |
| public SystemException getSystemExceptionReply() |
| { |
| return replyHeader.getSystemException(replyExceptionDetailMessage); |
| } |
| |
| //////////////////////////////////////////////////// |
| // |
| // Used by server side. |
| // |
| |
| public ObjectKey getObjectKey() |
| { |
| return getRequestHeader().getObjectKey(); |
| } |
| |
| public void setProtocolHandler(CorbaProtocolHandler protocolHandler) |
| { |
| throw wrapper.methodShouldNotBeCalled() ; |
| } |
| |
| public CorbaProtocolHandler getProtocolHandler() |
| { |
| // REVISIT: should look up in orb registry. |
| return this; |
| } |
| |
| //////////////////////////////////////////////////// |
| // |
| // ResponseHandler |
| // |
| |
| public org.omg.CORBA.portable.OutputStream createReply() |
| { |
| // Note: relies on side-effect of setting mediator output field. |
| // REVISIT - cast - need interface |
| getProtocolHandler().createResponse(this, (ServiceContexts) null); |
| return (OutputStream) getOutputObject(); |
| } |
| |
| public org.omg.CORBA.portable.OutputStream createExceptionReply() |
| { |
| // Note: relies on side-effect of setting mediator output field. |
| // REVISIT - cast - need interface |
| getProtocolHandler().createUserExceptionResponse(this, (ServiceContexts) null); |
| return (OutputStream) getOutputObject(); |
| } |
| |
| public boolean executeReturnServantInResponseConstructor() |
| { |
| return _executeReturnServantInResponseConstructor; |
| |
| } |
| |
| public void setExecuteReturnServantInResponseConstructor(boolean b) |
| { |
| _executeReturnServantInResponseConstructor = b; |
| } |
| |
| public boolean executeRemoveThreadInfoInResponseConstructor() |
| { |
| return _executeRemoveThreadInfoInResponseConstructor; |
| } |
| |
| public void setExecuteRemoveThreadInfoInResponseConstructor(boolean b) |
| { |
| _executeRemoveThreadInfoInResponseConstructor = b; |
| } |
| |
| public boolean executePIInResponseConstructor() |
| { |
| return _executePIInResponseConstructor; |
| } |
| |
| public void setExecutePIInResponseConstructor( boolean b ) |
| { |
| _executePIInResponseConstructor = b; |
| } |
| |
| private byte getStreamFormatVersionForThisRequest(IOR ior, |
| GIOPVersion giopVersion) |
| { |
| |
| byte localMaxVersion |
| = ORBUtility.getMaxStreamFormatVersion(); |
| |
| IOR effectiveTargetIOR = |
| ((CorbaContactInfo)this.contactInfo).getEffectiveTargetIOR(); |
| IIOPProfileTemplate temp = |
| (IIOPProfileTemplate)effectiveTargetIOR.getProfile().getTaggedProfileTemplate(); |
| Iterator iter = temp.iteratorById(TAG_RMI_CUSTOM_MAX_STREAM_FORMAT.value); |
| if (!iter.hasNext()) { |
| // Didn't have the max stream format version tagged |
| // component. |
| if (giopVersion.lessThan(GIOPVersion.V1_3)) |
| return ORBConstants.STREAM_FORMAT_VERSION_1; |
| else |
| return ORBConstants.STREAM_FORMAT_VERSION_2; |
| } |
| |
| byte remoteMaxVersion |
| = ((MaxStreamFormatVersionComponent)iter.next()).getMaxStreamFormatVersion(); |
| |
| return (byte)Math.min(localMaxVersion, remoteMaxVersion); |
| } |
| |
| //////////////////////////////////////////////////////////////////////// |
| //////////////////////////////////////////////////////////////////////// |
| //////////////////////////////////////////////////////////////////////// |
| |
| // REVISIT - This could be a separate implementation object looked |
| // up in a registry. However it needs some state in the message |
| // mediator so combine for now. |
| |
| |
| protected boolean isThreadDone = false; |
| |
| //////////////////////////////////////////////////// |
| // |
| // pept.protocol.ProtocolHandler |
| // |
| |
| public boolean handleRequest(MessageMediator messageMediator) |
| { |
| try { |
| dispatchHeader.callback(this); |
| } catch (IOException e) { |
| // REVISIT - this should be handled internally. |
| ; |
| } |
| return isThreadDone; |
| } |
| |
| //////////////////////////////////////////////////// |
| // |
| // iiop.messages.MessageHandler |
| // |
| |
| private void setWorkThenPoolOrResumeSelect(Message header) |
| { |
| if (getConnection().getEventHandler().shouldUseSelectThreadToWait()) { |
| resumeSelect(header); |
| } else { |
| // Leader/Follower when using reader thread. |
| // When this thread is done working it will go back in pool. |
| |
| isThreadDone = true; |
| |
| // First unregister current registration. |
| orb.getTransportManager().getSelector(0) |
| .unregisterForEvent(getConnection().getEventHandler()); |
| // Have another thread become the reader. |
| orb.getTransportManager().getSelector(0) |
| .registerForEvent(getConnection().getEventHandler()); |
| } |
| } |
| |
| private void setWorkThenReadOrResumeSelect(Message header) |
| { |
| if (getConnection().getEventHandler().shouldUseSelectThreadToWait()) { |
| resumeSelect(header); |
| } else { |
| // When using reader thread then wen this thread is |
| // done working it will continue reading. |
| isThreadDone = false; |
| } |
| } |
| |
| private void resumeSelect(Message header) |
| { |
| // NOTE: VERY IMPORTANT: |
| // Only participate in select after getting to the point |
| // that proper serialization of fragments is ensured. |
| |
| if (transportDebug()) { |
| dprint(".resumeSelect:->"); |
| // REVISIT: not-OO: |
| String requestId = "?"; |
| if (header instanceof RequestMessage) { |
| requestId = |
| new Integer(((RequestMessage)header) |
| .getRequestId()).toString(); |
| } else if (header instanceof ReplyMessage) { |
| requestId = |
| new Integer(((ReplyMessage)header) |
| .getRequestId()).toString(); |
| } else if (header instanceof FragmentMessage_1_2) { |
| requestId = |
| new Integer(((FragmentMessage_1_2)header) |
| .getRequestId()).toString(); |
| } |
| dprint(".resumeSelect: id/" |
| + requestId |
| + " " + getConnection() |
| ); |
| |
| } |
| |
| // IMPORTANT: To avoid bug (4953599), we force the Thread that does the NIO select |
| // to also do the enable/disable of Ops using SelectionKey.interestOps(Ops of Interest). |
| // Otherwise, the SelectionKey.interestOps(Ops of Interest) may block indefinitely in |
| // this thread. |
| EventHandler eventHandler = getConnection().getEventHandler(); |
| orb.getTransportManager().getSelector(0).registerInterestOps(eventHandler); |
| |
| if (transportDebug()) { |
| dprint(".resumeSelect:<-"); |
| } |
| } |
| |
| private void setInputObject() |
| { |
| // REVISIT: refactor createInputObject (and createMessageMediator) |
| // into base PlugInFactory. Get via connection (either ContactInfo |
| // or Acceptor). |
| if (getConnection().getContactInfo() != null) { |
| inputObject = (CDRInputObject) |
| getConnection().getContactInfo() |
| .createInputObject(orb, this); |
| } else if (getConnection().getAcceptor() != null) { |
| inputObject = (CDRInputObject) |
| getConnection().getAcceptor() |
| .createInputObject(orb, this); |
| } else { |
| throw new RuntimeException("CorbaMessageMediatorImpl.setInputObject"); |
| } |
| inputObject.setMessageMediator(this); |
| setInputObject(inputObject); |
| } |
| |
| private void signalResponseReceived() |
| { |
| // This will end up using the MessageMediator associated with |
| // the original request instead of the current mediator (which |
| // need to be constructed to hold the dispatchBuffer and connection). |
| connection.getResponseWaitingRoom() |
| .responseReceived((InputObject)inputObject); |
| } |
| |
| // This handles message types for which we don't create classes. |
| public void handleInput(Message header) throws IOException |
| { |
| try { |
| messageHeader = header; |
| |
| if (transportDebug()) |
| dprint(".handleInput->: " |
| + MessageBase.typeToString(header.getType())); |
| |
| setWorkThenReadOrResumeSelect(header); |
| |
| switch(header.getType()) |
| { |
| case Message.GIOPCloseConnection: |
| if (transportDebug()) { |
| dprint(".handleInput: CloseConnection: purging"); |
| } |
| connection.purgeCalls(wrapper.connectionRebind(), true, false); |
| break; |
| case Message.GIOPMessageError: |
| if (transportDebug()) { |
| dprint(".handleInput: MessageError: purging"); |
| } |
| connection.purgeCalls(wrapper.recvMsgError(), true, false); |
| break; |
| default: |
| if (transportDebug()) { |
| dprint(".handleInput: ERROR: " |
| + MessageBase.typeToString(header.getType())); |
| } |
| throw wrapper.badGiopRequestType() ; |
| } |
| releaseByteBufferToPool(); |
| } finally { |
| if (transportDebug()) { |
| dprint(".handleInput<-: " |
| + MessageBase.typeToString(header.getType())); |
| } |
| } |
| } |
| |
| public void handleInput(RequestMessage_1_0 header) throws IOException |
| { |
| try { |
| if (transportDebug()) dprint(".REQUEST 1.0->: " + header); |
| try { |
| messageHeader = requestHeader = (RequestMessage) header; |
| setInputObject(); |
| } finally { |
| setWorkThenPoolOrResumeSelect(header); |
| } |
| getProtocolHandler().handleRequest(header, this); |
| } catch (Throwable t) { |
| if (transportDebug()) |
| dprint(".REQUEST 1.0: !!ERROR!!: " + header, t); |
| // Mask the exception from thread.; |
| } finally { |
| if (transportDebug()) dprint(".REQUEST 1.0<-: " + header); |
| } |
| } |
| |
| public void handleInput(RequestMessage_1_1 header) throws IOException |
| { |
| try { |
| if (transportDebug()) dprint(".REQUEST 1.1->: " + header); |
| try { |
| messageHeader = requestHeader = (RequestMessage) header; |
| setInputObject(); |
| connection.serverRequest_1_1_Put(this); |
| } finally { |
| setWorkThenPoolOrResumeSelect(header); |
| } |
| getProtocolHandler().handleRequest(header, this); |
| } catch (Throwable t) { |
| if (transportDebug()) |
| dprint(".REQUEST 1.1: !!ERROR!!: " + header, t); |
| // Mask the exception from thread.; |
| } finally { |
| if (transportDebug()) dprint(".REQUEST 1.1<-: " + header); |
| } |
| } |
| |
| // REVISIT: this is identical to 1_0 except for fragment part. |
| public void handleInput(RequestMessage_1_2 header) throws IOException |
| { |
| try { |
| try { |
| |
| messageHeader = requestHeader = (RequestMessage) header; |
| |
| header.unmarshalRequestID(dispatchByteBuffer); |
| setInputObject(); |
| |
| if (transportDebug()) dprint(".REQUEST 1.2->: id/" |
| + header.getRequestId() |
| + ": " |
| + header); |
| |
| // NOTE: in the old code this used to be done conditionally: |
| // if (header.moreFragmentsToFollow()). |
| // Now we always put it in. We take it out when |
| // the response is done. |
| // This must happen now so if a header is fragmented the stream |
| // may be found. |
| connection.serverRequestMapPut(header.getRequestId(), this); |
| } finally { |
| // Leader/Follower. |
| // Note: This *MUST* come after putting stream in above map |
| // since the header may be fragmented and you do not want to |
| // start reading again until the map above is set. |
| setWorkThenPoolOrResumeSelect(header); |
| } |
| //inputObject.unmarshalHeader(); // done in subcontract. |
| getProtocolHandler().handleRequest(header, this); |
| } catch (Throwable t) { |
| if (transportDebug()) dprint(".REQUEST 1.2: id/" |
| + header.getRequestId() |
| + ": !!ERROR!!: " |
| + header, |
| t); |
| // Mask the exception from thread.; |
| } finally { |
| connection.serverRequestMapRemove(header.getRequestId()); |
| |
| if (transportDebug()) dprint(".REQUEST 1.2<-: id/" |
| + header.getRequestId() |
| + ": " |
| + header); |
| } |
| } |
| |
| public void handleInput(ReplyMessage_1_0 header) throws IOException |
| { |
| try { |
| try { |
| if (transportDebug()) dprint(".REPLY 1.0->: " + header); |
| messageHeader = replyHeader = (ReplyMessage) header; |
| setInputObject(); |
| |
| // REVISIT: this should be done by waiting thread. |
| inputObject.unmarshalHeader(); |
| |
| signalResponseReceived(); |
| } finally{ |
| setWorkThenReadOrResumeSelect(header); |
| } |
| } catch (Throwable t) { |
| if (transportDebug())dprint(".REPLY 1.0: !!ERROR!!: " + header, t); |
| // Mask the exception from thread.; |
| } finally { |
| if (transportDebug()) dprint(".REPLY 1.0<-: " + header); |
| } |
| } |
| |
| public void handleInput(ReplyMessage_1_1 header) throws IOException |
| { |
| try { |
| if (transportDebug()) dprint(".REPLY 1.1->: " + header); |
| messageHeader = replyHeader = (ReplyMessage) header; |
| setInputObject(); |
| |
| if (header.moreFragmentsToFollow()) { |
| |
| // More fragments are coming to complete this reply, so keep |
| // a reference to the InputStream so we can add the fragments |
| connection.clientReply_1_1_Put(this); |
| |
| // In 1.1, we can't assume that we have the request ID in the |
| // first fragment. Thus, another thread is used |
| // to be the reader while this thread unmarshals |
| // the extended header and wakes up the client thread. |
| setWorkThenPoolOrResumeSelect(header); |
| |
| // REVISIT - error handling. |
| // This must be done now. |
| inputObject.unmarshalHeader(); |
| |
| signalResponseReceived(); |
| |
| } else { |
| |
| // Not fragmented, therefore we know the request |
| // ID is here. Thus, we can unmarshal the extended header |
| // and wake up the client thread without using a third |
| // thread as above. |
| |
| // REVISIT - error handling during unmarshal. |
| // This must be done now to get the request id. |
| inputObject.unmarshalHeader(); |
| |
| signalResponseReceived(); |
| |
| setWorkThenReadOrResumeSelect(header); |
| } |
| } catch (Throwable t) { |
| if (transportDebug()) dprint(".REPLY 1.1: !!ERROR!!: " + header); |
| // Mask the exception from thread.; |
| } finally { |
| if (transportDebug()) dprint(".REPLY 1.1<-: " + header); |
| } |
| } |
| |
| public void handleInput(ReplyMessage_1_2 header) throws IOException |
| { |
| try { |
| try { |
| messageHeader = replyHeader = (ReplyMessage) header; |
| |
| // We know that the request ID is in the first fragment |
| header.unmarshalRequestID(dispatchByteBuffer); |
| |
| if (transportDebug()) { |
| dprint(".REPLY 1.2->: id/" |
| + + header.getRequestId() |
| + ": more?: " + header.moreFragmentsToFollow() |
| + ": " + header); |
| } |
| |
| setInputObject(); |
| |
| signalResponseReceived(); |
| } finally { |
| setWorkThenReadOrResumeSelect(header); |
| } |
| } catch (Throwable t) { |
| if (transportDebug()) dprint(".REPLY 1.2: id/" |
| + header.getRequestId() |
| + ": !!ERROR!!: " |
| + header, t); |
| // Mask the exception from thread.; |
| } finally { |
| if (transportDebug()) dprint(".REPLY 1.2<-: id/" |
| + header.getRequestId() |
| + ": " |
| + header); |
| } |
| } |
| |
| public void handleInput(LocateRequestMessage_1_0 header) throws IOException |
| { |
| try { |
| if (transportDebug()) |
| dprint(".LOCATE_REQUEST 1.0->: " + header); |
| try { |
| messageHeader = header; |
| setInputObject(); |
| } finally { |
| setWorkThenPoolOrResumeSelect(header); |
| } |
| getProtocolHandler().handleRequest(header, this); |
| } catch (Throwable t) { |
| if (transportDebug()) |
| dprint(".LOCATE_REQUEST 1.0: !!ERROR!!: " + header, t); |
| // Mask the exception from thread.; |
| } finally { |
| if (transportDebug()) |
| dprint(".LOCATE_REQUEST 1.0<-: " + header); |
| } |
| |
| } |
| |
| public void handleInput(LocateRequestMessage_1_1 header) throws IOException |
| { |
| try { |
| if (transportDebug()) |
| dprint(".LOCATE_REQUEST 1.1->: " + header); |
| try { |
| messageHeader = header; |
| setInputObject(); |
| } finally { |
| setWorkThenPoolOrResumeSelect(header); |
| } |
| getProtocolHandler().handleRequest(header, this); |
| } catch (Throwable t) { |
| if (transportDebug()) |
| dprint(".LOCATE_REQUEST 1.1: !!ERROR!!: " + header, t); |
| // Mask the exception from thread.; |
| } finally { |
| if (transportDebug()) |
| dprint(".LOCATE_REQUEST 1.1<-:" + header); |
| } |
| } |
| |
| public void handleInput(LocateRequestMessage_1_2 header) throws IOException |
| { |
| try { |
| try { |
| messageHeader = header; |
| |
| header.unmarshalRequestID(dispatchByteBuffer); |
| setInputObject(); |
| |
| if (transportDebug()) |
| dprint(".LOCATE_REQUEST 1.2->: id/" |
| + header.getRequestId() |
| + ": " |
| + header); |
| |
| if (header.moreFragmentsToFollow()) { |
| connection.serverRequestMapPut(header.getRequestId(),this); |
| } |
| } finally { |
| setWorkThenPoolOrResumeSelect(header); |
| } |
| getProtocolHandler().handleRequest(header, this); |
| } catch (Throwable t) { |
| if (transportDebug()) |
| dprint(".LOCATE_REQUEST 1.2: id/" |
| + header.getRequestId() |
| + ": !!ERROR!!: " |
| + header, t); |
| // Mask the exception from thread.; |
| } finally { |
| if (transportDebug()) |
| dprint(".LOCATE_REQUEST 1.2<-: id/" |
| + header.getRequestId() |
| + ": " |
| + header); |
| } |
| } |
| |
| public void handleInput(LocateReplyMessage_1_0 header) throws IOException |
| { |
| try { |
| if (transportDebug()) |
| dprint(".LOCATE_REPLY 1.0->:" + header); |
| try { |
| messageHeader = header; |
| setInputObject(); |
| inputObject.unmarshalHeader(); // REVISIT Put in subcontract. |
| signalResponseReceived(); |
| } finally { |
| setWorkThenReadOrResumeSelect(header); |
| } |
| } catch (Throwable t) { |
| if (transportDebug()) |
| dprint(".LOCATE_REPLY 1.0: !!ERROR!!: " + header, t); |
| // Mask the exception from thread.; |
| } finally { |
| if (transportDebug()) |
| dprint(".LOCATE_REPLY 1.0<-: " + header); |
| } |
| } |
| |
| public void handleInput(LocateReplyMessage_1_1 header) throws IOException |
| { |
| try { |
| if (transportDebug()) dprint(".LOCATE_REPLY 1.1->: " + header); |
| try { |
| messageHeader = header; |
| setInputObject(); |
| // Fragmented LocateReplies are not allowed in 1.1. |
| inputObject.unmarshalHeader(); |
| signalResponseReceived(); |
| } finally { |
| setWorkThenReadOrResumeSelect(header); |
| } |
| } catch (Throwable t) { |
| if (transportDebug()) |
| dprint(".LOCATE_REPLY 1.1: !!ERROR!!: " + header, t); |
| // Mask the exception from thread.; |
| } finally { |
| if (transportDebug()) dprint(".LOCATE_REPLY 1.1<-: " + header); |
| } |
| } |
| |
| public void handleInput(LocateReplyMessage_1_2 header) throws IOException |
| { |
| try { |
| try { |
| messageHeader = header; |
| |
| // No need to put in client reply map - already there. |
| header.unmarshalRequestID(dispatchByteBuffer); |
| |
| setInputObject(); |
| |
| if (transportDebug()) dprint(".LOCATE_REPLY 1.2->: id/" |
| + header.getRequestId() |
| + ": " |
| + header); |
| |
| signalResponseReceived(); |
| } finally { |
| setWorkThenPoolOrResumeSelect(header); // REVISIT |
| } |
| } catch (Throwable t) { |
| if (transportDebug()) |
| dprint(".LOCATE_REPLY 1.2: id/" |
| + header.getRequestId() |
| + ": !!ERROR!!: " |
| + header, t); |
| // Mask the exception from thread.; |
| } finally { |
| if (transportDebug()) dprint(".LOCATE_REPLY 1.2<-: id/" |
| + header.getRequestId() |
| + ": " |
| + header); |
| } |
| } |
| |
| public void handleInput(FragmentMessage_1_1 header) throws IOException |
| { |
| try { |
| if (transportDebug()) { |
| dprint(".FRAGMENT 1.1->: " |
| + "more?: " + header.moreFragmentsToFollow() |
| + ": " + header); |
| } |
| try { |
| messageHeader = header; |
| MessageMediator mediator = null; |
| CDRInputObject inputObject = null; |
| |
| if (connection.isServer()) { |
| mediator = connection.serverRequest_1_1_Get(); |
| } else { |
| mediator = connection.clientReply_1_1_Get(); |
| } |
| if (mediator != null) { |
| inputObject = (CDRInputObject) mediator.getInputObject(); |
| } |
| |
| // If no input stream available, then discard the fragment. |
| // This can happen: |
| // 1. if a fragment message is received prior to receiving |
| // the original request/reply message. Very unlikely. |
| // 2. if a fragment message is received after the |
| // reply has been sent (early replies) |
| // Note: In the case of early replies, the fragments received |
| // during the request processing (which are never unmarshaled), |
| // will eventually be discarded by the GC. |
| if (inputObject == null) { |
| if (transportDebug()) |
| dprint(".FRAGMENT 1.1: ++++DISCARDING++++: " + header); |
| // need to release dispatchByteBuffer to pool if |
| // we are discarding |
| releaseByteBufferToPool(); |
| return; |
| } |
| |
| inputObject.getBufferManager() |
| .processFragment(dispatchByteBuffer, header); |
| |
| if (! header.moreFragmentsToFollow()) { |
| if (connection.isServer()) { |
| connection.serverRequest_1_1_Remove(); |
| } else { |
| connection.clientReply_1_1_Remove(); |
| } |
| } |
| } finally { |
| // NOTE: This *must* come after queing the fragment |
| // when using the selector to ensure fragments stay in order. |
| setWorkThenReadOrResumeSelect(header); |
| } |
| } catch (Throwable t) { |
| if (transportDebug()) |
| dprint(".FRAGMENT 1.1: !!ERROR!!: " + header, t); |
| // Mask the exception from thread.; |
| } finally { |
| if (transportDebug()) dprint(".FRAGMENT 1.1<-: " + header); |
| } |
| } |
| |
| public void handleInput(FragmentMessage_1_2 header) throws IOException |
| { |
| try { |
| try { |
| messageHeader = header; |
| |
| // Note: We know it's a 1.2 fragment, we have the data, but |
| // we need the IIOPInputStream instance to unmarshal the |
| // request ID... but we need the request ID to get the |
| // IIOPInputStream instance. So we peek at the raw bytes. |
| |
| header.unmarshalRequestID(dispatchByteBuffer); |
| |
| if (transportDebug()) { |
| dprint(".FRAGMENT 1.2->: id/" |
| + header.getRequestId() |
| + ": more?: " + header.moreFragmentsToFollow() |
| + ": " + header); |
| } |
| |
| MessageMediator mediator = null; |
| InputObject inputObject = null; |
| |
| if (connection.isServer()) { |
| mediator = |
| connection.serverRequestMapGet(header.getRequestId()); |
| } else { |
| mediator = |
| connection.clientRequestMapGet(header.getRequestId()); |
| } |
| if (mediator != null) { |
| inputObject = mediator.getInputObject(); |
| } |
| // See 1.1 comments. |
| if (inputObject == null) { |
| if (transportDebug()) { |
| dprint(".FRAGMENT 1.2: id/" |
| + header.getRequestId() |
| + ": ++++DISCARDING++++: " |
| + header); |
| } |
| // need to release dispatchByteBuffer to pool if |
| // we are discarding |
| releaseByteBufferToPool(); |
| return; |
| } |
| ((CDRInputObject)inputObject) |
| .getBufferManager().processFragment( |
| dispatchByteBuffer, header); |
| |
| // REVISIT: but if it is a server don't you have to remove the |
| // stream from the map? |
| if (! connection.isServer()) { |
| /* REVISIT |
| * No need to do anything. |
| * Should we mark that last was received? |
| if (! header.moreFragmentsToFollow()) { |
| // Last fragment. |
| } |
| */ |
| } |
| } finally { |
| // NOTE: This *must* come after queing the fragment |
| // when using the selector to ensure fragments stay in order. |
| setWorkThenReadOrResumeSelect(header); |
| } |
| } catch (Throwable t) { |
| if (transportDebug()) |
| dprint(".FRAGMENT 1.2: id/" |
| + header.getRequestId() |
| + ": !!ERROR!!: " |
| + header, t); |
| // Mask the exception from thread.; |
| } finally { |
| if (transportDebug()) dprint(".FRAGMENT 1.2<-: id/" |
| + header.getRequestId() |
| + ": " |
| + header); |
| } |
| } |
| |
| public void handleInput(CancelRequestMessage header) throws IOException |
| { |
| try { |
| try { |
| messageHeader = header; |
| setInputObject(); |
| |
| // REVISIT: Move these two to subcontract. |
| inputObject.unmarshalHeader(); |
| |
| if (transportDebug()) dprint(".CANCEL->: id/" |
| + header.getRequestId() + ": " |
| + header.getGIOPVersion() + ": " |
| + header); |
| |
| processCancelRequest(header.getRequestId()); |
| releaseByteBufferToPool(); |
| } finally { |
| setWorkThenReadOrResumeSelect(header); |
| } |
| } catch (Throwable t) { |
| if (transportDebug()) dprint(".CANCEL: id/" |
| + header.getRequestId() |
| + ": !!ERROR!!: " |
| + header, t); |
| // Mask the exception from thread.; |
| } finally { |
| if (transportDebug()) dprint(".CANCEL<-: id/" |
| + header.getRequestId() + ": " |
| + header.getGIOPVersion() + ": " |
| + header); |
| } |
| } |
| |
| private void throwNotImplemented() |
| { |
| isThreadDone = false; |
| throwNotImplemented(""); |
| } |
| |
| private void throwNotImplemented(String msg) |
| { |
| throw new RuntimeException("CorbaMessageMediatorImpl: not implemented " + msg); |
| } |
| |
| private void dprint(String msg, Throwable t) |
| { |
| dprint(msg); |
| t.printStackTrace(System.out); |
| } |
| |
| private void dprint(String msg) |
| { |
| ORBUtility.dprint("CorbaMessageMediatorImpl", msg); |
| } |
| |
| protected String opAndId(CorbaMessageMediator mediator) |
| { |
| return ORBUtility.operationNameAndRequestId(mediator); |
| } |
| |
| private boolean transportDebug() |
| { |
| return orb.transportDebugFlag; |
| } |
| |
| // REVISIT: move this to subcontract (but both client and server need it). |
| private final void processCancelRequest(int cancelReqId) { |
| |
| // The GIOP version of CancelRequest does not matter, since |
| // CancelRequest_1_0 could be sent to cancel a request which |
| // has a different GIOP version. |
| |
| /* |
| * CancelRequest processing logic : |
| * |
| * - find the request with matching requestId |
| * |
| * - call cancelProcessing() in BufferManagerRead [BMR] |
| * |
| * - the hope is that worker thread would call BMR.underflow() |
| * to wait for more fragments to come in. When BMR.underflow() is |
| * called, if a CancelRequest had already arrived, |
| * the worker thread would throw ThreadDeath, |
| * else the thread would wait to be notified of the |
| * arrival of a new fragment or CancelRequest. Upon notification, |
| * the woken up thread would check to see if a CancelRequest had |
| * arrived and if so throw a ThreadDeath or it will continue to |
| * process the received fragment. |
| * |
| * - if all the fragments had been received prior to CancelRequest |
| * then the worker thread would never block in BMR.underflow(). |
| * So, setting the abort flag in BMR has no effect. The request |
| * processing will complete normally. |
| * |
| * - in the case where the server has received enough fragments to |
| * start processing the request and the server sends out |
| * an early reply. In such a case if the CancelRequest arrives |
| * after the reply has been sent, it has no effect. |
| */ |
| |
| if (!connection.isServer()) { |
| return; // we do not support bi-directional giop yet, ignore. |
| } |
| |
| // Try to get hold of the InputStream buffer. |
| // In the case of 1.0 requests there is no way to get hold of |
| // InputStream. Try out the 1.1 and 1.2 cases. |
| |
| // was the request 1.2 ? |
| MessageMediator mediator = connection.serverRequestMapGet(cancelReqId); |
| int requestId ; |
| if (mediator == null) { |
| // was the request 1.1 ? |
| mediator = connection.serverRequest_1_1_Get(); |
| if (mediator == null) { |
| // XXX log this! |
| // either the request was 1.0 |
| // or an early reply has already been sent |
| // or request processing is over |
| // or its a spurious CancelRequest |
| return; // do nothing. |
| } |
| |
| requestId = ((CorbaMessageMediator) mediator).getRequestId(); |
| |
| if (requestId != cancelReqId) { |
| // A spurious 1.1 CancelRequest has been received. |
| // XXX log this! |
| return; // do nothing |
| } |
| |
| if (requestId == 0) { // special case |
| // XXX log this |
| // this means that |
| // 1. the 1.1 requests' requestId has not been received |
| // i.e., a CancelRequest was received even before the |
| // 1.1 request was received. The spec disallows this. |
| // 2. or the 1.1 request has a requestId 0. |
| // |
| // It is a little tricky to distinguish these two. So, be |
| // conservative and do not cancel the request. Downside is that |
| // 1.1 requests with requestId of 0 will never be cancelled. |
| return; // do nothing |
| } |
| } else { |
| requestId = ((CorbaMessageMediator) mediator).getRequestId(); |
| } |
| |
| Message msg = ((CorbaMessageMediator)mediator).getRequestHeader(); |
| if (msg.getType() != Message.GIOPRequest) { |
| // Any mediator obtained here should only ever be for a GIOP |
| // request. |
| wrapper.badMessageTypeForCancel() ; |
| } |
| |
| // At this point we have a valid message mediator that contains |
| // a valid requestId. |
| |
| // at this point we have chosen a request to be cancelled. But we |
| // do not know if the target object's method has been invoked or not. |
| // Request input stream being available simply means that the request |
| // processing is not over yet. simply set the abort flag in the |
| // BMRS and hope that the worker thread would notice it (this can |
| // happen only if the request stream is being unmarshalled and the |
| // target's method has not been invoked yet). This guarantees |
| // that the requests which have been dispatched to the |
| // target's method will never be cancelled. |
| |
| BufferManagerReadStream bufferManager = (BufferManagerReadStream) |
| ((CDRInputObject)mediator.getInputObject()).getBufferManager(); |
| bufferManager.cancelProcessing(cancelReqId); |
| } |
| |
| //////////////////////////////////////////////////// |
| // |
| // spi.protocol.CorbaProtocolHandler |
| // |
| |
| public void handleRequest(RequestMessage msg, |
| CorbaMessageMediator messageMediator) |
| { |
| try { |
| beginRequest(messageMediator); |
| try { |
| handleRequestRequest(messageMediator); |
| if (messageMediator.isOneWay()) { |
| return; |
| } |
| } catch (Throwable t) { |
| if (messageMediator.isOneWay()) { |
| return; |
| } |
| handleThrowableDuringServerDispatch( |
| messageMediator, t, CompletionStatus.COMPLETED_MAYBE); |
| } |
| sendResponse(messageMediator); |
| } catch (Throwable t) { |
| dispatchError(messageMediator, "RequestMessage", t); |
| } finally { |
| endRequest(messageMediator); |
| } |
| } |
| |
| public void handleRequest(LocateRequestMessage msg, |
| CorbaMessageMediator messageMediator) |
| { |
| try { |
| beginRequest(messageMediator); |
| try { |
| handleLocateRequest(messageMediator); |
| } catch (Throwable t) { |
| handleThrowableDuringServerDispatch( |
| messageMediator, t, CompletionStatus.COMPLETED_MAYBE); |
| } |
| sendResponse(messageMediator); |
| } catch (Throwable t) { |
| dispatchError(messageMediator, "LocateRequestMessage", t); |
| } finally { |
| endRequest(messageMediator); |
| } |
| } |
| |
| private void beginRequest(CorbaMessageMediator messageMediator) |
| { |
| ORB orb = (ORB) messageMediator.getBroker(); |
| if (orb.subcontractDebugFlag) { |
| dprint(".handleRequest->:"); |
| } |
| connection.serverRequestProcessingBegins(); |
| } |
| |
| private void dispatchError(CorbaMessageMediator messageMediator, |
| String msg, Throwable t) |
| { |
| if (orb.subcontractDebugFlag) { |
| dprint(".handleRequest: " + opAndId(messageMediator) |
| + ": !!ERROR!!: " |
| + msg, |
| t); |
| } |
| // REVISIT - this makes hcks sendTwoObjects fail |
| // messageMediator.getConnection().close(); |
| } |
| |
| private void sendResponse(CorbaMessageMediator messageMediator) |
| { |
| if (orb.subcontractDebugFlag) { |
| dprint(".handleRequest: " + opAndId(messageMediator) |
| + ": sending response"); |
| } |
| // REVISIT - type and location |
| CDROutputObject outputObject = (CDROutputObject) |
| messageMediator.getOutputObject(); |
| if (outputObject != null) { |
| // REVISIT - can be null for TRANSIENT below. |
| outputObject.finishSendingMessage(); |
| } |
| } |
| |
| private void endRequest(CorbaMessageMediator messageMediator) |
| { |
| ORB orb = (ORB) messageMediator.getBroker(); |
| if (orb.subcontractDebugFlag) { |
| dprint(".handleRequest<-: " + opAndId(messageMediator)); |
| } |
| |
| // release NIO ByteBuffers to ByteBufferPool |
| |
| try { |
| OutputObject outputObj = messageMediator.getOutputObject(); |
| if (outputObj != null) { |
| outputObj.close(); |
| } |
| InputObject inputObj = messageMediator.getInputObject(); |
| if (inputObj != null) { |
| inputObj.close(); |
| } |
| } catch (IOException ex) { |
| // Given what close() does, this catch shouldn't ever happen. |
| // See CDRInput/OutputObject.close() for more info. |
| // It also won't result in a Corba error if an IOException happens. |
| if (orb.subcontractDebugFlag) { |
| dprint(".endRequest: IOException:" + ex.getMessage(), ex); |
| } |
| } finally { |
| ((CorbaConnection)messageMediator.getConnection()).serverRequestProcessingEnds(); |
| } |
| } |
| |
| protected void handleRequestRequest(CorbaMessageMediator messageMediator) |
| { |
| // Does nothing if already unmarshaled. |
| ((CDRInputObject)messageMediator.getInputObject()).unmarshalHeader(); |
| |
| ORB orb = (ORB)messageMediator.getBroker(); |
| synchronized (orb) { |
| orb.checkShutdownState(); |
| } |
| |
| ObjectKey okey = messageMediator.getObjectKey(); |
| if (orb.subcontractDebugFlag) { |
| ObjectKeyTemplate oktemp = okey.getTemplate() ; |
| dprint( ".handleRequest: " + opAndId(messageMediator) |
| + ": dispatching to scid: " + oktemp.getSubcontractId()); |
| } |
| |
| CorbaServerRequestDispatcher sc = okey.getServerRequestDispatcher(orb); |
| |
| if (orb.subcontractDebugFlag) { |
| dprint(".handleRequest: " + opAndId(messageMediator) |
| + ": dispatching to sc: " + sc); |
| } |
| |
| if (sc == null) { |
| throw wrapper.noServerScInDispatch() ; |
| } |
| |
| // NOTE: |
| // This is necessary so mediator can act as ResponseHandler |
| // and pass necessary info to response constructors located |
| // in the subcontract. |
| // REVISIT - same class right now. |
| //messageMediator.setProtocolHandler(this); |
| |
| try { |
| orb.startingDispatch(); |
| sc.dispatch(messageMediator); |
| } finally { |
| orb.finishedDispatch(); |
| } |
| } |
| |
| protected void handleLocateRequest(CorbaMessageMediator messageMediator) |
| { |
| ORB orb = (ORB)messageMediator.getBroker(); |
| LocateRequestMessage msg = (LocateRequestMessage) |
| messageMediator.getDispatchHeader(); |
| IOR ior = null; |
| LocateReplyMessage reply = null; |
| short addrDisp = -1; |
| |
| try { |
| ((CDRInputObject)messageMediator.getInputObject()).unmarshalHeader(); |
| CorbaServerRequestDispatcher sc = |
| msg.getObjectKey().getServerRequestDispatcher( orb ) ; |
| if (sc == null) { |
| return; |
| } |
| |
| ior = sc.locate(msg.getObjectKey()); |
| |
| if ( ior == null ) { |
| reply = MessageBase.createLocateReply( |
| orb, msg.getGIOPVersion(), |
| msg.getEncodingVersion(), |
| msg.getRequestId(), |
| LocateReplyMessage.OBJECT_HERE, null); |
| |
| } else { |
| reply = MessageBase.createLocateReply( |
| orb, msg.getGIOPVersion(), |
| msg.getEncodingVersion(), |
| msg.getRequestId(), |
| LocateReplyMessage.OBJECT_FORWARD, ior); |
| } |
| // REVISIT: Should we catch SystemExceptions? |
| |
| } catch (AddressingDispositionException ex) { |
| |
| // create a response containing the expected target |
| // addressing disposition. |
| |
| reply = MessageBase.createLocateReply( |
| orb, msg.getGIOPVersion(), |
| msg.getEncodingVersion(), |
| msg.getRequestId(), |
| LocateReplyMessage.LOC_NEEDS_ADDRESSING_MODE, null); |
| |
| addrDisp = ex.expectedAddrDisp(); |
| |
| } catch (RequestCanceledException ex) { |
| |
| return; // no need to send reply |
| |
| } catch ( Exception ex ) { |
| |
| // REVISIT If exception is not OBJECT_NOT_EXIST, it should |
| // have a different reply |
| |
| // This handles OBJECT_NOT_EXIST exceptions thrown in |
| // the subcontract or obj manager. Send back UNKNOWN_OBJECT. |
| |
| reply = MessageBase.createLocateReply( |
| orb, msg.getGIOPVersion(), |
| msg.getEncodingVersion(), |
| msg.getRequestId(), |
| LocateReplyMessage.UNKNOWN_OBJECT, null); |
| } |
| |
| CDROutputObject outputObject = |
| createAppropriateOutputObject(messageMediator, |
| msg, reply); |
| messageMediator.setOutputObject(outputObject); |
| outputObject.setMessageMediator(messageMediator); |
| |
| reply.write(outputObject); |
| // outputObject.setMessage(reply); // REVISIT - not necessary |
| if (ior != null) { |
| ior.write(outputObject); |
| } |
| if (addrDisp != -1) { |
| AddressingDispositionHelper.write(outputObject, addrDisp); |
| } |
| } |
| |
| private CDROutputObject createAppropriateOutputObject( |
| CorbaMessageMediator messageMediator, |
| Message msg, LocateReplyMessage reply) |
| { |
| CDROutputObject outputObject; |
| |
| if (msg.getGIOPVersion().lessThan(GIOPVersion.V1_2)) { |
| // locate msgs 1.0 & 1.1 :=> grow, |
| outputObject = sun.corba.OutputStreamFactory.newCDROutputObject( |
| (ORB) messageMediator.getBroker(), |
| this, |
| GIOPVersion.V1_0, |
| (CorbaConnection) messageMediator.getConnection(), |
| reply, |
| ORBConstants.STREAM_FORMAT_VERSION_1); |
| } else { |
| // 1.2 :=> stream |
| outputObject = sun.corba.OutputStreamFactory.newCDROutputObject( |
| (ORB) messageMediator.getBroker(), |
| messageMediator, |
| reply, |
| ORBConstants.STREAM_FORMAT_VERSION_1); |
| } |
| return outputObject; |
| } |
| |
| public void handleThrowableDuringServerDispatch( |
| CorbaMessageMediator messageMediator, |
| Throwable throwable, |
| CompletionStatus completionStatus) |
| { |
| if (((ORB)messageMediator.getBroker()).subcontractDebugFlag) { |
| dprint(".handleThrowableDuringServerDispatch: " |
| + opAndId(messageMediator) + ": " |
| + throwable); |
| } |
| |
| // If we haven't unmarshaled the header, we probably don't |
| // have enough information to even send back a reply. |
| |
| // REVISIT |
| // Cannot do this check. When target addressing disposition does |
| // not match (during header unmarshaling) it throws an exception |
| // to be handled here. |
| /* |
| if (! ((CDRInputObject)messageMediator.getInputObject()) |
| .unmarshaledHeader()) { |
| return; |
| } |
| */ |
| handleThrowableDuringServerDispatch(messageMediator, |
| throwable, |
| completionStatus, |
| 1); |
| } |
| |
| |
| // REVISIT - catch and ignore RequestCanceledException. |
| |
| protected void handleThrowableDuringServerDispatch( |
| CorbaMessageMediator messageMediator, |
| Throwable throwable, |
| CompletionStatus completionStatus, |
| int iteration) |
| { |
| if (iteration > 10) { |
| if (((ORB)messageMediator.getBroker()).subcontractDebugFlag) { |
| dprint(".handleThrowableDuringServerDispatch: " |
| + opAndId(messageMediator) |
| + ": cannot handle: " |
| + throwable); |
| } |
| |
| // REVISIT - should we close connection? |
| RuntimeException rte = |
| new RuntimeException("handleThrowableDuringServerDispatch: " + |
| "cannot create response."); |
| rte.initCause(throwable); |
| throw rte; |
| } |
| |
| try { |
| if (throwable instanceof ForwardException) { |
| ForwardException fex = (ForwardException)throwable ; |
| createLocationForward( messageMediator, fex.getIOR(), null ) ; |
| return; |
| } |
| |
| if (throwable instanceof AddressingDispositionException) { |
| handleAddressingDisposition( |
| messageMediator, |
| (AddressingDispositionException)throwable); |
| return; |
| } |
| |
| // Else. |
| |
| SystemException sex = |
| convertThrowableToSystemException(throwable, completionStatus); |
| |
| createSystemExceptionResponse(messageMediator, sex, null); |
| return; |
| |
| } catch (Throwable throwable2) { |
| |
| // User code (e.g., postinvoke, interceptors) may change |
| // the exception, so we end up back here. |
| // Report the changed exception. |
| |
| handleThrowableDuringServerDispatch(messageMediator, |
| throwable2, |
| completionStatus, |
| iteration + 1); |
| return; |
| } |
| } |
| |
| protected SystemException convertThrowableToSystemException( |
| Throwable throwable, |
| CompletionStatus completionStatus) |
| { |
| if (throwable instanceof SystemException) { |
| return (SystemException)throwable; |
| } |
| |
| if (throwable instanceof RequestCanceledException) { |
| // Reporting an exception response causes the |
| // poa current stack, the interceptor stacks, etc. |
| // to be balanced. It also notifies interceptors |
| // that the request was cancelled. |
| |
| return wrapper.requestCanceled( throwable ) ; |
| } |
| |
| // NOTE: We do not trap ThreadDeath above Throwable. |
| // There is no reason to stop the thread. It is |
| // just a worker thread. The ORB never throws |
| // ThreadDeath. Client code may (e.g., in ServantManagers, |
| // interceptors, or servants) but that should not |
| // effect the ORB threads. So it is just handled |
| // generically. |
| |
| // |
| // Last resort. |
| // If user code throws a non-SystemException report it generically. |
| // |
| |
| return wrapper.runtimeexception( CompletionStatus.COMPLETED_MAYBE, throwable ) ; |
| } |
| |
| protected void handleAddressingDisposition( |
| CorbaMessageMediator messageMediator, |
| AddressingDispositionException ex) |
| { |
| |
| short addrDisp = -1; |
| |
| // from iiop.RequestProcessor. |
| |
| // Respond with expected target addressing disposition. |
| |
| switch (messageMediator.getRequestHeader().getType()) { |
| case Message.GIOPRequest : |
| ReplyMessage replyHeader = MessageBase.createReply( |
| (ORB)messageMediator.getBroker(), |
| messageMediator.getGIOPVersion(), |
| messageMediator.getEncodingVersion(), |
| messageMediator.getRequestId(), |
| ReplyMessage.NEEDS_ADDRESSING_MODE, |
| null, null); |
| // REVISIT: via acceptor factory. |
| CDROutputObject outputObject = |
| sun.corba.OutputStreamFactory.newCDROutputObject( |
| (ORB)messageMediator.getBroker(), |
| this, |
| messageMediator.getGIOPVersion(), |
| (CorbaConnection)messageMediator.getConnection(), |
| replyHeader, |
| ORBConstants.STREAM_FORMAT_VERSION_1); |
| messageMediator.setOutputObject(outputObject); |
| outputObject.setMessageMediator(messageMediator); |
| replyHeader.write(outputObject); |
| AddressingDispositionHelper.write(outputObject, |
| ex.expectedAddrDisp()); |
| return; |
| |
| case Message.GIOPLocateRequest : |
| LocateReplyMessage locateReplyHeader = MessageBase.createLocateReply( |
| (ORB)messageMediator.getBroker(), |
| messageMediator.getGIOPVersion(), |
| messageMediator.getEncodingVersion(), |
| messageMediator.getRequestId(), |
| LocateReplyMessage.LOC_NEEDS_ADDRESSING_MODE, |
| null); |
| |
| addrDisp = ex.expectedAddrDisp(); |
| |
| // REVISIT: via acceptor factory. |
| outputObject = |
| createAppropriateOutputObject(messageMediator, |
| messageMediator.getRequestHeader(), |
| locateReplyHeader); |
| messageMediator.setOutputObject(outputObject); |
| outputObject.setMessageMediator(messageMediator); |
| locateReplyHeader.write(outputObject); |
| IOR ior = null; |
| if (ior != null) { |
| ior.write(outputObject); |
| } |
| if (addrDisp != -1) { |
| AddressingDispositionHelper.write(outputObject, addrDisp); |
| } |
| return; |
| } |
| } |
| |
| public CorbaMessageMediator createResponse( |
| CorbaMessageMediator messageMediator, |
| ServiceContexts svc) |
| { |
| // REVISIT: ignore service contexts during framework transition. |
| // They are set in SubcontractResponseHandler to the wrong connection. |
| // Then they would be set again here and a duplicate contexts |
| // exception occurs. |
| return createResponseHelper( |
| messageMediator, |
| getServiceContextsForReply(messageMediator, null)); |
| } |
| |
| public CorbaMessageMediator createUserExceptionResponse( |
| CorbaMessageMediator messageMediator, ServiceContexts svc) |
| { |
| // REVISIT - same as above |
| return createResponseHelper( |
| messageMediator, |
| getServiceContextsForReply(messageMediator, null), |
| true); |
| } |
| |
| public CorbaMessageMediator createUnknownExceptionResponse( |
| CorbaMessageMediator messageMediator, UnknownException ex) |
| { |
| // NOTE: This service context container gets augmented in |
| // tail call. |
| ServiceContexts contexts = null; |
| SystemException sys = new UNKNOWN( 0, |
| CompletionStatus.COMPLETED_MAYBE); |
| contexts = new ServiceContexts( (ORB)messageMediator.getBroker() ); |
| UEInfoServiceContext uei = new UEInfoServiceContext(sys); |
| contexts.put( uei ) ; |
| return createSystemExceptionResponse(messageMediator, sys, contexts); |
| } |
| |
| public CorbaMessageMediator createSystemExceptionResponse( |
| CorbaMessageMediator messageMediator, |
| SystemException ex, |
| ServiceContexts svc) |
| { |
| if (messageMediator.getConnection() != null) { |
| // It is possible that fragments of response have already been |
| // sent. Then an error may occur (e.g. marshaling error like |
| // non serializable object). In that case it is too late |
| // to send the exception. We just return the existing fragmented |
| // stream here. This will cause an incomplete last fragment |
| // to be sent. Then the other side will get a marshaling error |
| // when attempting to unmarshal. |
| |
| // REVISIT: Impl - make interface method to do the following. |
| CorbaMessageMediatorImpl mediator = (CorbaMessageMediatorImpl) |
| ((CorbaConnection)messageMediator.getConnection()) |
| .serverRequestMapGet(messageMediator.getRequestId()); |
| |
| OutputObject existingOutputObject = null; |
| if (mediator != null) { |
| existingOutputObject = mediator.getOutputObject(); |
| } |
| |
| // REVISIT: need to think about messageMediator containing correct |
| // pointer to output object. |
| if (existingOutputObject != null && |
| mediator.sentFragment() && |
| ! mediator.sentFullMessage()) |
| { |
| return mediator; |
| } |
| } |
| |
| // Only do this if interceptors have been initialized on this request |
| // and have not completed their lifecycle (otherwise the info stack |
| // may be empty or have a different request's entry on top). |
| if (messageMediator.executePIInResponseConstructor()) { |
| // REVISIT: not necessary in framework now? |
| // Inform Portable Interceptors of the SystemException. This is |
| // required to be done here because the ending interception point |
| // is called in the when creating the response below |
| // but we do not currently write the SystemException into the |
| // response until after the ending point is called. |
| ((ORB)messageMediator.getBroker()).getPIHandler().setServerPIInfo( ex ); |
| } |
| |
| if (((ORB)messageMediator.getBroker()).subcontractDebugFlag && |
| ex != null) |
| { |
| dprint(".createSystemExceptionResponse: " |
| + opAndId(messageMediator), |
| ex); |
| } |
| |
| ServiceContexts serviceContexts = |
| getServiceContextsForReply(messageMediator, svc); |
| |
| // NOTE: We MUST add the service context before creating |
| // the response since service contexts are written to the |
| // stream when the response object is created. |
| |
| addExceptionDetailMessage(messageMediator, ex, serviceContexts); |
| |
| CorbaMessageMediator response = |
| createResponseHelper(messageMediator, serviceContexts, false); |
| |
| // NOTE: From here on, it is too late to add more service contexts. |
| // They have already been serialized to the stream (and maybe fragments |
| // sent). |
| |
| ORBUtility.writeSystemException( |
| ex, (OutputStream)response.getOutputObject()); |
| |
| return response; |
| } |
| |
| private void addExceptionDetailMessage(CorbaMessageMediator mediator, |
| SystemException ex, |
| ServiceContexts serviceContexts) |
| { |
| ByteArrayOutputStream baos = new ByteArrayOutputStream(); |
| PrintWriter pw = new PrintWriter(baos); |
| ex.printStackTrace(pw); |
| pw.flush(); // NOTE: you must flush or baos will be empty. |
| EncapsOutputStream encapsOutputStream = |
| sun.corba.OutputStreamFactory.newEncapsOutputStream((ORB)mediator.getBroker()); |
| encapsOutputStream.putEndian(); |
| encapsOutputStream.write_wstring(baos.toString()); |
| UnknownServiceContext serviceContext = |
| new UnknownServiceContext(ExceptionDetailMessage.value, |
| encapsOutputStream.toByteArray()); |
| serviceContexts.put(serviceContext); |
| } |
| |
| public CorbaMessageMediator createLocationForward( |
| CorbaMessageMediator messageMediator, IOR ior, ServiceContexts svc) |
| { |
| ReplyMessage reply |
| = MessageBase.createReply( |
| (ORB)messageMediator.getBroker(), |
| messageMediator.getGIOPVersion(), |
| messageMediator.getEncodingVersion(), |
| messageMediator.getRequestId(), |
| ReplyMessage.LOCATION_FORWARD, |
| getServiceContextsForReply(messageMediator, svc), |
| ior); |
| |
| return createResponseHelper(messageMediator, reply, ior); |
| } |
| |
| protected CorbaMessageMediator createResponseHelper( |
| CorbaMessageMediator messageMediator, ServiceContexts svc) |
| { |
| ReplyMessage message = |
| MessageBase.createReply( |
| (ORB)messageMediator.getBroker(), |
| messageMediator.getGIOPVersion(), |
| messageMediator.getEncodingVersion(), |
| messageMediator.getRequestId(), |
| ReplyMessage.NO_EXCEPTION, |
| svc, |
| null); |
| return createResponseHelper(messageMediator, message, null); |
| } |
| |
| protected CorbaMessageMediator createResponseHelper( |
| CorbaMessageMediator messageMediator, ServiceContexts svc,boolean user) |
| { |
| ReplyMessage message = |
| MessageBase.createReply( |
| (ORB)messageMediator.getBroker(), |
| messageMediator.getGIOPVersion(), |
| messageMediator.getEncodingVersion(), |
| messageMediator.getRequestId(), |
| user ? ReplyMessage.USER_EXCEPTION : |
| ReplyMessage.SYSTEM_EXCEPTION, |
| svc, |
| null); |
| return createResponseHelper(messageMediator, message, null); |
| } |
| |
| // REVISIT - IOR arg is ignored. |
| protected CorbaMessageMediator createResponseHelper( |
| CorbaMessageMediator messageMediator, ReplyMessage reply, IOR ior) |
| { |
| // REVISIT - these should be invoked from subcontract. |
| runServantPostInvoke(messageMediator); |
| runInterceptors(messageMediator, reply); |
| runRemoveThreadInfo(messageMediator); |
| |
| if (((ORB)messageMediator.getBroker()).subcontractDebugFlag) { |
| dprint(".createResponseHelper: " |
| + opAndId(messageMediator) + ": " |
| + reply); |
| } |
| |
| messageMediator.setReplyHeader(reply); |
| |
| OutputObject replyOutputObject; |
| // REVISIT = do not use null. |
| // |
| if (messageMediator.getConnection() == null) { |
| replyOutputObject = |
| sun.corba.OutputStreamFactory.newCDROutputObject(orb, |
| messageMediator, messageMediator.getReplyHeader(), |
| messageMediator.getStreamFormatVersion(), |
| BufferManagerFactory.GROW); |
| } else { |
| replyOutputObject = messageMediator.getConnection().getAcceptor() |
| .createOutputObject(messageMediator.getBroker(), messageMediator); |
| } |
| messageMediator.setOutputObject(replyOutputObject); |
| messageMediator.getOutputObject().setMessageMediator(messageMediator); |
| |
| reply.write((OutputStream) messageMediator.getOutputObject()); |
| if (reply.getIOR() != null) { |
| reply.getIOR().write((OutputStream) messageMediator.getOutputObject()); |
| } |
| // REVISIT - not necessary? |
| //messageMediator.this.replyIOR = reply.getIOR(); |
| |
| // NOTE: The mediator holds onto output object so return value |
| // not really necessary. |
| return messageMediator; |
| } |
| |
| protected void runServantPostInvoke(CorbaMessageMediator messageMediator) |
| { |
| // Run ServantLocator::postinvoke. This may cause a SystemException |
| // which will throw out of the constructor and return later |
| // to construct a reply for that exception. The internal logic |
| // of returnServant makes sure that postinvoke is only called once. |
| // REVISIT: instead of instanceof, put method on all orbs. |
| ORB orb = null; |
| // This flag is to deal with BootstrapServer use of reply streams, |
| // with ServerRequestDispatcher's use of reply streams, etc. |
| if (messageMediator.executeReturnServantInResponseConstructor()) { |
| // It is possible to get marshaling errors in the skeleton after |
| // postinvoke has completed. We must set this to false so that |
| // when the error exception reply is constructed we don't try |
| // to incorrectly access poa current (which will be the wrong |
| // one or an empty stack. |
| messageMediator.setExecuteReturnServantInResponseConstructor(false); |
| messageMediator.setExecuteRemoveThreadInfoInResponseConstructor(true); |
| |
| try { |
| orb = (ORB)messageMediator.getBroker(); |
| OAInvocationInfo info = orb.peekInvocationInfo() ; |
| ObjectAdapter oa = info.oa(); |
| try { |
| oa.returnServant() ; |
| } catch (Throwable thr) { |
| wrapper.unexpectedException( thr ) ; |
| |
| if (thr instanceof Error) |
| throw (Error)thr ; |
| else if (thr instanceof RuntimeException) |
| throw (RuntimeException)thr ; |
| } finally { |
| oa.exit(); |
| } |
| } catch (EmptyStackException ese) { |
| throw wrapper.emptyStackRunServantPostInvoke( ese ) ; |
| } |
| } |
| } |
| |
| protected void runInterceptors(CorbaMessageMediator messageMediator, |
| ReplyMessage reply) |
| { |
| if( messageMediator.executePIInResponseConstructor() ) { |
| // Invoke server request ending interception points (send_*): |
| // Note: this may end up with a SystemException or an internal |
| // Runtime ForwardRequest |
| ((ORB)messageMediator.getBroker()).getPIHandler(). |
| invokeServerPIEndingPoint( reply ); |
| |
| // Note this will be executed even if a ForwardRequest or |
| // SystemException is thrown by a Portable Interceptors ending |
| // point since we end up in this constructor again anyway. |
| ((ORB)messageMediator.getBroker()).getPIHandler(). |
| cleanupServerPIRequest(); |
| |
| // See createSystemExceptionResponse for why this is necesary. |
| messageMediator.setExecutePIInResponseConstructor(false); |
| } |
| } |
| |
| protected void runRemoveThreadInfo(CorbaMessageMediator messageMediator) |
| { |
| // Once you get here then the final reply is available (i.e., |
| // postinvoke and interceptors have completed. |
| if (messageMediator.executeRemoveThreadInfoInResponseConstructor()) { |
| messageMediator.setExecuteRemoveThreadInfoInResponseConstructor(false); |
| ((ORB)messageMediator.getBroker()).popInvocationInfo() ; |
| } |
| } |
| |
| protected ServiceContexts getServiceContextsForReply( |
| CorbaMessageMediator messageMediator, ServiceContexts contexts) |
| { |
| CorbaConnection c = (CorbaConnection) messageMediator.getConnection(); |
| |
| if (((ORB)messageMediator.getBroker()).subcontractDebugFlag) { |
| dprint(".getServiceContextsForReply: " |
| + opAndId(messageMediator) |
| + ": " + c); |
| } |
| |
| if (contexts == null) { |
| contexts = new ServiceContexts(((ORB)messageMediator.getBroker())); |
| } |
| |
| // NOTE : We only want to send the runtime context the first time |
| |
| if (c != null && !c.isPostInitialContexts()) { |
| c.setPostInitialContexts(); |
| SendingContextServiceContext scsc = |
| new SendingContextServiceContext( |
| ((ORB)messageMediator.getBroker()).getFVDCodeBaseIOR()) ; |
| |
| if (contexts.get( scsc.getId() ) != null) |
| throw wrapper.duplicateSendingContextServiceContext() ; |
| |
| contexts.put( scsc ) ; |
| |
| if ( ((ORB)messageMediator.getBroker()).subcontractDebugFlag) |
| dprint(".getServiceContextsForReply: " |
| + opAndId(messageMediator) |
| + ": added SendingContextServiceContext" ) ; |
| } |
| |
| // send ORBVersion servicecontext as part of the Reply |
| |
| ORBVersionServiceContext ovsc |
| = new ORBVersionServiceContext(ORBVersionFactory.getORBVersion()); |
| |
| if (contexts.get( ovsc.getId() ) != null) |
| throw wrapper.duplicateOrbVersionServiceContext() ; |
| |
| contexts.put( ovsc ) ; |
| |
| if ( ((ORB)messageMediator.getBroker()).subcontractDebugFlag) |
| dprint(".getServiceContextsForReply: " |
| + opAndId(messageMediator) |
| + ": added ORB version service context"); |
| |
| return contexts; |
| } |
| |
| // REVISIT - this method should be migrated to orbutil.ORBUtility |
| // since all locations that release ByteBuffers use |
| // very similar logic and debug information. |
| private void releaseByteBufferToPool() { |
| if (dispatchByteBuffer != null) { |
| orb.getByteBufferPool().releaseByteBuffer(dispatchByteBuffer); |
| if (transportDebug()) { |
| int bbId = System.identityHashCode(dispatchByteBuffer); |
| StringBuffer sb = new StringBuffer(); |
| sb.append(".handleInput: releasing ByteBuffer (" + bbId + |
| ") to ByteBufferPool"); |
| dprint(sb.toString()); |
| } |
| } |
| } |
| } |
| |
| // End of file. |
