messagemediator.java

来自「java jdk 1.4的源码」· Java 代码 · 共 616 行 · 第 1/2 页

    {
        if (conn.getORB().transportDebugFlag)
            dprint("Handling GIOP 1.1 LocateRequest");

        IIOPInputStream is = new IIOPInputStream(conn, buf, header);

        threadPool.addWork(new RequestProcessor(conn.getServerGIOP().getRequestHandler(),
                                                conn,
                                                is));
    }

    public final void handleInput(LocateRequestMessage_1_2 header)
        throws IOException 
    {
        if (conn.getORB().transportDebugFlag)
            dprint("Handling GIOP 1.2 LocateRequest");

        IIOPInputStream is = new IIOPInputStream(conn, buf, header);

        header.unmarshalRequestID(buf);

        // More fragments are coming to complete this request message
        // add stream to the serverRequestMap
        if (header.moreFragmentsToFollow())
            conn.serverRequestMap.put(new Integer(header.getRequestId()), is);

        threadPool.addWork(new RequestProcessor(conn.getServerGIOP().getRequestHandler(),
                                                conn,
                                                is));
    }

    // Locate reply messages ------------------------

    public final void handleInput(LocateReplyMessage_1_0 header)
        throws IOException 
    {
        if (conn.getORB().transportDebugFlag)
            dprint("Handling GIOP 1.0 LocateReply");

        IIOPInputStream is = new IIOPInputStream(conn, buf, header);

        is.unmarshalHeader();

        conn.signalReplyReceived(header.getRequestId(), is);
    }

    public final void handleInput(LocateReplyMessage_1_1 header)
        throws IOException 
    {
        if (conn.getORB().transportDebugFlag)
            dprint("Handling GIOP 1.1 LocateReply");

        IIOPInputStream is = new IIOPInputStream(conn, buf, header);

        is.unmarshalHeader();

        // Fragmented LocateReplies are not allowed in 1.1
        conn.signalReplyReceived(header.getRequestId(), is);
    }

    public final void handleInput(LocateReplyMessage_1_2 header)
        throws IOException 
    {
        if (conn.getORB().transportDebugFlag)
            dprint("Handling GIOP 1.2 LocateReply");

        IIOPInputStream is = new IIOPInputStream(conn, buf, header);

        header.unmarshalRequestID(buf);

        // More fragments are coming to complete this reply, so keep
        // a reference to the InputStream so we can add the fragments
        if (header.moreFragmentsToFollow())
            conn.clientReplyMap.put(new Integer(header.getRequestId()), is);

        conn.signalReplyReceived(header.getRequestId(), is);
    }

    // Fragment messages ----------------------------

    public final void handleInput(FragmentMessage_1_1 header)
        throws IOException 
    {
        if (conn.getORB().transportDebugFlag)
            dprint("Handling GIOP 1.1 Fragment.  Last? " + header.moreFragmentsToFollow());

        IIOPInputStream is;

        if (conn.isServer())
            is = conn.theOnly1_1ServerRequestImpl;
        else
            is = conn.theOnly1_1ClientResponseImpl;

        // if there is no inputstream available, then discard the message
        // 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 (is == null) {
            return;
        }

        is.getBufferManager().processFragment(buf, header);

        if (!conn.isServer()) {

            // Is it a last fragment of a reply ?
            if (!header.moreFragmentsToFollow()) {

                // It is not the responsibility of this thread to remove
                // the OutCallDesc from out_calls -- only the client thread
                // should do that.
                conn.theOnly1_1ClientResponseImpl = null;
            }
        }
    }

    public final void handleInput(FragmentMessage_1_2 header)
        throws IOException
    {
        if (conn.getORB().transportDebugFlag)
            dprint("Handling GIOP 1.2 Fragment.  Last? " + header.moreFragmentsToFollow());

        // Unusual paradox:  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.

        header.unmarshalRequestID(buf);

        Integer requestId = new Integer(header.getRequestId());
        IIOPInputStream is;

        if (conn.isServer())
            is = (IIOPInputStream)conn.serverRequestMap.get(requestId);
        else
            is = (IIOPInputStream)conn.clientReplyMap.get(requestId);

        if (is == null) {
            return;
        }

        is.getBufferManager().processFragment(buf, header);

        if (!conn.isServer()) {

            // Is it a last fragment of a reply?
            if (!header.moreFragmentsToFollow()) {

                // It is not the responsibility of this thread to remove
                // the OutCallDesc from out_calls -- only the client thread
                // should do that.
                conn.clientReplyMap.remove(requestId);
            }
        }
    }

    // Cancel request messages -----------------------

    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 (!conn.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 ?
        IIOPInputStream is = (IIOPInputStream) conn.serverRequestMap.get(
                                                    new Integer(cancelReqId));

        if (is == null) { // was the request 1.1 ?

            is = conn.theOnly1_1ServerRequestImpl;

            if (is == null) {
                // 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.
            }

            Message msg = is.getMessage();
            if (msg.getType() != Message.GIOPRequest) {
                // this should not be true. Fragmented 1.1 messages can
                // only be request messages.
                return; // do nothing
            }

            int requestId = ((RequestMessage) msg).getRequestId();

            if (requestId == 0) { // special case
                // 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
            }

            // at this point we do have a valid requestId for the 1.1 request

            if (requestId != cancelReqId) {
                // A spurious CancelRequest has been received.
                return; // do nothing
            }
        }

        // 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)
                                                    is.getBufferManager();
        bufferManager.cancelProcessing(cancelReqId);
    }

    // Currently the same for all versions, but separate methods since
    // CancelRequestMessage is an interface.
    public final void handleInput(CancelRequestMessage_1_0 header)
        throws IOException
    {
        if (conn.getORB().transportDebugFlag) {
            dprint("Handling GIOP 1.0 CancelRequest");
        }

        IIOPInputStream is = new IIOPInputStream(conn, buf, header);
        header.read(is);

        this.processCancelRequest(header.getRequestId());
    }

    public final void handleInput(CancelRequestMessage_1_1 header)
        throws IOException
    {
        if (conn.getORB().transportDebugFlag) {
            dprint("Handling GIOP 1.1 CancelRequest");
        }

        IIOPInputStream is = new IIOPInputStream(conn, buf, header);
        header.read(is);

        this.processCancelRequest(header.getRequestId());
    }

    public final void handleInput(CancelRequestMessage_1_2 header)
        throws IOException
    {
        if (conn.getORB().transportDebugFlag) {
            dprint("Handling GIOP 1.2 CancelRequest");
        }

        IIOPInputStream is = new IIOPInputStream(conn, buf, header);
        header.read(is);
        
        this.processCancelRequest(header.getRequestId());
    }
}