corbaresponsewaitingroomimpl.java

来自「JAVA 所有包」· Java 代码 · 共 305 行

/*
 * @(#)CorbaResponseWaitingRoomImpl.java	1.30 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.transport;

import java.util.Hashtable;

import org.omg.CORBA.CompletionStatus;
import org.omg.CORBA.SystemException;

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.spi.logging.CORBALogDomains;
import com.sun.corba.se.spi.orb.ORB;
import com.sun.corba.se.spi.protocol.CorbaMessageMediator;
import com.sun.corba.se.spi.transport.CorbaConnection;
import com.sun.corba.se.spi.transport.CorbaResponseWaitingRoom;

import com.sun.corba.se.impl.encoding.BufferManagerReadStream;
import com.sun.corba.se.impl.encoding.CDRInputObject;
import com.sun.corba.se.impl.logging.ORBUtilSystemException;
import com.sun.corba.se.impl.orbutil.ORBUtility;
import com.sun.corba.se.impl.protocol.giopmsgheaders.LocateReplyOrReplyMessage;
import com.sun.corba.se.impl.protocol.giopmsgheaders.ReplyMessage;

/**
 * @author Harold Carr
 */
public class CorbaResponseWaitingRoomImpl
    implements
	CorbaResponseWaitingRoom
{
    final static class OutCallDesc
    {
        java.lang.Object done = new java.lang.Object();
        Thread thread;
	MessageMediator messageMediator;
        SystemException exception;
        InputObject inputObject;
    }

    private ORB orb;
    private ORBUtilSystemException wrapper ;

    private CorbaConnection connection;
    // Maps requestId to an OutCallDesc.
    private Hashtable out_calls = null; // REVISIT - use int hastable/map

    public CorbaResponseWaitingRoomImpl(ORB orb, CorbaConnection connection)
    {
	this.orb = orb;
	wrapper = ORBUtilSystemException.get( orb, 
	    CORBALogDomains.RPC_TRANSPORT ) ;
	this.connection = connection;
        out_calls = new Hashtable();
    }

    ////////////////////////////////////////////////////
    //
    // pept.transport.ResponseWaitingRoom
    //

    public void registerWaiter(MessageMediator mediator)
    {
	CorbaMessageMediator messageMediator = (CorbaMessageMediator) mediator;

	if (orb.transportDebugFlag) {
	    dprint(".registerWaiter: " + opAndId(messageMediator));
	}

	Integer requestId = messageMediator.getRequestIdInteger();
        
	OutCallDesc call = new OutCallDesc();
	call.thread = Thread.currentThread();
	call.messageMediator = messageMediator;
	out_calls.put(requestId, call);
    }

    public void unregisterWaiter(MessageMediator mediator)
    {
	CorbaMessageMediator messageMediator = (CorbaMessageMediator) mediator;

	if (orb.transportDebugFlag) {
	    dprint(".unregisterWaiter: " + opAndId(messageMediator));
	}

	Integer requestId = messageMediator.getRequestIdInteger();

        out_calls.remove(requestId);
    }

    public InputObject waitForResponse(MessageMediator mediator)
    {
      CorbaMessageMediator messageMediator = (CorbaMessageMediator) mediator;

      try {

        InputObject returnStream = null;

	if (orb.transportDebugFlag) {
	    dprint(".waitForResponse->: " + opAndId(messageMediator));
	}

	Integer requestId = messageMediator.getRequestIdInteger();

        if (messageMediator.isOneWay()) {
            // The waiter is removed in releaseReply in the same
            // way as a normal request.

	    if (orb.transportDebugFlag) {
		dprint(".waitForResponse: one way - not waiting: "
		       + opAndId(messageMediator));
	    }

            return null;
        }

        OutCallDesc call = (OutCallDesc)out_calls.get(requestId);
        if (call == null) {
	    throw wrapper.nullOutCall(CompletionStatus.COMPLETED_MAYBE);
	}

        synchronized(call.done) {

            while (call.inputObject == null && call.exception == null) {
                // Wait for the reply from the server.
                // The ReaderThread reads in the reply IIOP message
                // and signals us.
                try {
		    if (orb.transportDebugFlag) {
			dprint(".waitForResponse: waiting: "
			       + opAndId(messageMediator));
		    }
                    call.done.wait();
                } catch (InterruptedException ie) {};
            }

            if (call.exception != null) {
		if (orb.transportDebugFlag) {
		    dprint(".waitForResponse: exception: " 
			   + opAndId(messageMediator));
		}
                throw call.exception;
            }

            returnStream = call.inputObject;
        }

	// REVISIT -- exceptions from unmarshaling code will
        // go up through this client thread!

        if (returnStream != null) {
	    // On fragmented streams the header MUST be unmarshaled here
	    // (in the client thread) in case it blocks.
	    // If the header was already unmarshaled, this won't
	    // do anything
	    // REVISIT: cast - need interface method.
	    ((CDRInputObject)returnStream).unmarshalHeader();
	}

        return returnStream;

      } finally {
	if (orb.transportDebugFlag) {
	    dprint(".waitForResponse<-: " + opAndId(messageMediator));
	}
      }
    }

    public void responseReceived(InputObject is) 
    {
	CDRInputObject inputObject = (CDRInputObject) is;
	LocateReplyOrReplyMessage header = (LocateReplyOrReplyMessage)
	    inputObject.getMessageHeader();
        Integer requestId = new Integer(header.getRequestId());
        OutCallDesc call = (OutCallDesc) out_calls.get(requestId);

	if (orb.transportDebugFlag) {
	    dprint(".responseReceived: id/"
		   + requestId  + ": "
		   + header);
	}

        // This is an interesting case.  It could mean that someone sent us a
        // reply message, but we don't know what request it was for.  That
        // would probably call for an error.  However, there's another case
        // that's normal and we should think about --
        //
        // If the unmarshaling thread does all of its work inbetween the time
        // the ReaderThread gives it the last fragment and gets to the
        // out_calls.get line, then it will also be null, so just return;
        if (call == null) {
	    if (orb.transportDebugFlag) {
		dprint(".responseReceived: id/" 
		       + requestId
		       + ": no waiter: "
		       + header);
	    }
            return;
	}

        // Set the reply InputObject and signal the client thread
        // that the reply has been received.
        // The thread signalled will remove outcall descriptor if appropriate.
        // Otherwise, it'll be removed when last fragment for it has been put on
        // BufferManagerRead's queue.
        synchronized (call.done) {
	    CorbaMessageMediator messageMediator = (CorbaMessageMediator)
		call.messageMediator;

	    if (orb.transportDebugFlag) {
		dprint(".responseReceived: "
		       + opAndId(messageMediator)
		       + ": notifying waiters");
	    }

	    messageMediator.setReplyHeader(header);
	    messageMediator.setInputObject(is);
	    inputObject.setMessageMediator(messageMediator);
            call.inputObject = is;
            call.done.notify();
        }
    }

    public int numberRegistered()
    {
        // Note: Hashtable.size() is not synchronized
	return out_calls.size();
    }

    //////////////////////////////////////////////////
    //
    // CorbaResponseWaitingRoom
    //

    public void signalExceptionToAllWaiters(SystemException systemException)
    {

	if (orb.transportDebugFlag) {
	    dprint(".signalExceptionToAllWaiters: " + systemException);
	}

        OutCallDesc call;
        java.util.Enumeration e = out_calls.elements();
        while(e.hasMoreElements()) {
            call = (OutCallDesc) e.nextElement();
        
            synchronized(call.done){
                // anything waiting for BufferManagerRead's fragment queue
                // needs to be cancelled
                CorbaMessageMediator corbaMsgMediator =
                             (CorbaMessageMediator)call.messageMediator;
                CDRInputObject inputObject =
                           (CDRInputObject)corbaMsgMediator.getInputObject();
                // IMPORTANT: If inputObject is null, then no need to tell
                //            BufferManagerRead to cancel request processing. 
                if (inputObject != null) {
                    BufferManagerReadStream bufferManager =
                        (BufferManagerReadStream)inputObject.getBufferManager();
                    int requestId = corbaMsgMediator.getRequestId();
                    bufferManager.cancelProcessing(requestId);
                }
                call.inputObject = null;
                call.exception = systemException;
                call.done.notify();
            }
        }
    }

    public MessageMediator getMessageMediator(int requestId)
    {
        Integer id = new Integer(requestId);
        OutCallDesc call = (OutCallDesc) out_calls.get(id);
	if (call == null) {
	    // This can happen when getting early reply fragments for a
	    // request which has completed (e.g., client marshaling error).
	    return null;
	}
	return call.messageMediator;
    }

    ////////////////////////////////////////////////////
    //
    // Implementation.
    //

    protected void dprint(String msg)
    {
	ORBUtility.dprint("CorbaResponseWaitingRoomImpl", msg);
    }

    protected String opAndId(CorbaMessageMediator mediator)
    {
	return ORBUtility.operationNameAndRequestId(mediator);
    }
}

// End of file.