messagemediator.java

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

/*
 * @(#)MessageMediator.java	1.13 03/01/23
 *
 * Copyright 2003 Sun Microsystems, Inc. All rights reserved.
 * SUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 */
package com.sun.corba.se.internal.iiop;

import java.io.IOException;

// to force 1.3 compiler to pickup the newer location
// - remove once we move to JDK1.4
import com.sun.corba.se.internal.iiop.messages.Message;

import com.sun.corba.se.internal.iiop.messages.*;
import com.sun.corba.se.internal.orbutil.MinorCodes;
import com.sun.corba.se.internal.orbutil.ORBUtility;
import com.sun.corba.se.internal.iiop.IIOPConnection.OutCallDesc;
import org.omg.CORBA.INTERNAL;
import org.omg.CORBA.CompletionStatus;

import com.sun.corba.se.internal.orbutil.ThreadPool;
import com.sun.corba.se.internal.orbutil.Work;

/**
 * Provides processing for messages without if/switch overhead for versions
 * by using polymorphism and callbacks (double dispatch pattern).  Separates 
 * IIOPConnection from the workings of the  iiop/message system and vice 
 * versa.  Makes it easy to see the behavior for a given {msg type, version}
 * combination.
 *
 * A single thread (the ReaderThread) accesses a MessageMediator instance.
 */
public final class MessageMediator
{
    /** 
     * With GIOP 1.1 replies, we have no guarantee that the request ID
     * was in the first fragment, so we can't signal the client thread
     * to wake up and unmarshal the extended header.  Thus, we must
     * use another thread to do it.
     */
    private static class ReplyProcessor_1_1 implements Work
    {
        private static final String name = "ReplyProcessor 1.1";

        private IIOPConnection conn;
        private IIOPInputStream reply;

        ReplyProcessor_1_1(IIOPConnection conn, IIOPInputStream reply) {
            this.conn = conn;
            this.reply = reply;
        }

        public final String getName() {
            return name;
        }
        
        public void process() 
        {
            // Needs error handling
            reply.unmarshalHeader();

            ReplyMessage msg = (ReplyMessage)reply.getMessage();

            conn.signalReplyReceived(msg.getRequestId(),
                                     reply);
        }
    }    

    public MessageMediator(IIOPConnection conn) 
    {
        this.conn = conn;
    }

    // This could be a singleton, but that would lead to more
    // parameter passing.  As is, there will be one instance per
    // IIOPConnection.
    private IIOPConnection conn;
    private byte[] buf;

    // If this weren't static, there wouldn't be any synchronization overhead
    // when threads on different connections fight for access.  The downside
    // would be that cached threads would only be available per connection.
    // Since there isn't a scheme for expiring cached threads, yet, I've
    // made it static. -eea1 REVISIT
    //
    // You can argue this should be in IIOPConnection;
    private final static ThreadPool threadPool = new ThreadPool();

    private void dprint( String msg ) {
	ORBUtility.dprint( this, msg ) ;
    }

    /**
     * Create the appropriate message type, allocate a byte buffer of the
     * appropriate size, read in the message, and use the callback on the
     * message object to do the processing.
     */
    public final void processRequest()
        throws IOException
    {
        if (conn.getORB().transportDebugFlag)
            dprint("Creating message from stream");

        // Read in the message header and create the appropriate type
        // of IIOP message.
        MessageBase msg = (MessageBase)MessageBase.createFromStream(conn.getORB(), 
                                                                    conn.getInputStream());

        // Create a buffer of the correct size.  We don't even have
        // to copy the GIOP header into it since we'll never look
        // at the bytes again.
        this.buf = new byte[msg.getSize()];

        if (conn.getORB().transportDebugFlag)
            dprint("Reading the message fully, size =" + msg.getSize());

        // Read all the data into the buffer
        MessageBase.readFully(conn.getInputStream(), 
                              buf, 
                              MessageBase.GIOPMessageHeaderLength, 
                              msg.getSize() - MessageBase.GIOPMessageHeaderLength);

        if (conn.getORB().giopDebugFlag) {

            // For debugging purposes, copy the 12 bytes of the
            // GIOP header in to the main buffer
            System.arraycopy(msg.giopHeader, 0,
                             this.buf, 0,
                             12);
            
            dprint("Received message:");
            ByteBufferWithInfo bbwi = new ByteBufferWithInfo(this.buf, 0);
            bbwi.buflen = msg.getSize();
            CDRInputStream_1_0.printBuffer(bbwi);
        }

        // Ask the message to call back to the mediator to handle
        // the request.  The mediator does the appropriate thing
        // based on the message.
    
        msg.callback(this);
    }

    // (Currently this handles message types that we don't create classes for)
    public final void handleInput(MessageBase header)
    {
        if (conn.getORB().transportDebugFlag)
            dprint("Handling other GIOP message: " + header.getType());

        switch(header.getType()) 
        {
            case Message.GIOPCloseConnection:
                if (conn.getORB().transportDebugFlag)
                    dprint("Connection.processInput: got CloseConn, purging");
                conn.purge_calls(Connection.CONN_REBIND, true, false);
                break;
            case Message.GIOPMessageError:
                if (conn.getORB().transportDebugFlag)
                    dprint("Received MessageError, purging");
                conn.purge_calls(MinorCodes.RECV_MSG_ERROR, true, false);
                break;
            default:
                if (conn.getORB().transportDebugFlag)
                    dprint("Connection: bad message type" + header.getType());

                throw new INTERNAL(MinorCodes.BAD_GIOP_REQUEST_TYPE,
                                   CompletionStatus.COMPLETED_NO);
        }
    }

    
    // Request messages -----------------------------

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

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

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

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

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

        // More fragments are coming to complete this request message
        // add stream to the serverRequestMap
        if (header.moreFragmentsToFollow())
            conn.theOnly1_1ServerRequestImpl = (ServerRequestImpl)is;

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

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

        IIOPInputStream is = new ServerRequestImpl(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));
    }

    // Reply messages ---------------------------------

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

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

        is.unmarshalHeader();

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

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

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

        // 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.theOnly1_1ClientResponseImpl = (ClientResponseImpl)is;
            
            // In 1.1, we can't assume that we have the request ID in the
            // first fragment.  Thus, another thread is used to unmarshal
            // the extended header and wake up the client thread.
            threadPool.addWork(new ReplyProcessor_1_1(conn, is));

        } else {

            // If this is the only fragment, then 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.
            is.unmarshalHeader();

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

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

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

        // We know that the request ID is in the first fragment
        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);
    }

    // Locate request messages ------------------------

    // Versions 1.0 and 1.1 cannot be fragmented, so the implementation can be
    // the same here.
    public final void handleInput(LocateRequestMessage_1_0 header)
        throws IOException 
    {
        if (conn.getORB().transportDebugFlag)
            dprint("Handling GIOP 1.0 LocateRequest");

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

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

    public final void handleInput(LocateRequestMessage_1_1 header)
        throws IOException