iiopconnection.java

来自「java jdk 1.4的源码」· Java 代码 · 共 1,000 行 · 第 1/3 页

    {
        mediator.processRequest();
    }

    /**
     * Signal the client thread that the given request has been received,
     * and set the input stream on its out call descriptor.
     */
    void signalReplyReceived(int requestId, IIOPInputStream is) 
    {
        Integer id = new Integer(requestId);
        OutCallDesc call = (OutCallDesc) out_calls.get(id);

        // 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)
            return;

        // Set the reply IIOPInputStream 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) {
            call.s = is;
            call.done.notify();
        }
    }

    /**
     * Wake up the outstanding requests on the connection, and hand them
     * COMM_FAILURE exception with a given minor code. Also, delete connection
     * from connection table and stop the reader thread. Note that this should only
     * ever be called by the Reader thread for this connection.
     * @param minor_code The minor code for the COMM_FAILURE major code.
     * @param die Kill the reader thread (this thread) before exiting.
     */
    void purge_calls(int minor_code, boolean die, boolean lockHeld)
    {
        OutCallDesc call;

        if (orb.transportDebugFlag) {
            dprint("purge_calls: starting: code = " + minor_code
                   + " die = " + die);
	}

        //
        // If this invocation is a result of ThreadDeath caused
        // by a previous execution of this routine, just exit.
        //
        synchronized ( stateEvent ){
            if ((state == ABORT) || (state == CLOSE_RECVD)) {
                if (orb.transportDebugFlag)
                    dprint("purge_calls: exiting duplicate invocation");
                return;
            }
        }

        //
        // Grab the writeLock (freeze the calls)
        //
        try {
            if (!lockHeld)
                writeLock();
        } catch (SystemException ex) {
            if (orb.transportDebugFlag)
                dprint("purge_calls: caught exception " + ex + "; continuing");
        }

        //
        // Mark the state of the connection and determine the request status
        //
        org.omg.CORBA.CompletionStatus completion_status;

        synchronized ( stateEvent ){
            if (minor_code ==  Connection.CONN_REBIND) {
                state = CLOSE_RECVD;
                completion_status = CompletionStatus.COMPLETED_NO;
            } else {
                state = ABORT;
                completion_status = CompletionStatus.COMPLETED_MAYBE;
            }
            stateEvent.notifyAll();
        }
    

        //
        // Close the socket (if its not already closed)
        //
        try {
            // if theres no socket/connection, this is just ignored
            inputStream.close();
            outputStream.close();
            socket.close();
        } catch (Exception ex) {
        }

        SystemException comm_failure_exc =
            new COMM_FAILURE(minor_code, completion_status);

        // Signal all threads with outstanding requests on this
        // connection and give them the COMM_FAILURE exception.
        java.util.Enumeration e = out_calls.elements();
        while(e.hasMoreElements()) {
            call = (OutCallDesc) e.nextElement();
        
            synchronized(call.done){
                call.s = null;
                call.exc = comm_failure_exc;
                call.done.notify();
            }
        }

        //
        // delete connection from cache and stop the reader thread
        //
        connectionTable.deleteConn(endpoint);

        //
        // Signal all the waiters of the writeLock.
        // There are 4 types of writeLock waiters:
        // 1. Send waiters:
        // 2. SendReply waiters:
        // 3. cleanUp waiters:
        // 4. purge_call waiters:
        //
        writeUnlock();

    }


    /**
     * Sets up an established connection
     */
    public void setConnection(Socket _socket, ConnectionTable ctab)
        throws Exception
    {
        socket = _socket;
        inputStream = socket.getInputStream();
        outputStream = socket.getOutputStream();
        connectionTable = ctab;

	
        synchronized ( stateEvent ){
            state = ESTABLISHED;
        
            // Catch exceptions since setDaemon can cause a
            // security exception to be thrown under netscape
            // in the Applet mode
            try {
                AccessController.doPrivileged(new PrivilegedAction() {
                    public java.lang.Object run() {
                        reader.setDaemon(true);
                        return null;
                    }
                });
            }
            catch (Exception e) {}
        
            reader.start();
            stateEvent.notifyAll();
        }

    }

    /**
     * Changes state of connection to aborted, notifying waiters
     */
    public void abortConnection()
    {
        synchronized ( stateEvent ){
        
            state = ABORT;
            ((ReaderThread) reader).shutdown();
            stateEvent.notifyAll();
        }
    
    }

    /**
     * Sets the writeLock for this connection.
     * If the writeLock is already set by someone else, block till the
     * writeLock is released and can set by us.
     * IMPORTANT: this connection's lock must be acquired before
     * setting the writeLock and must be unlocked after setting the writeLock.
     */
    protected boolean writeLock()
    {

        // Keep looping till we can set the writeLock.
        while ( true ) {
            synchronized ( stateEvent ){
                switch ( state ) {
                
                case OPENING:
                    try {
                        stateEvent.wait();
                    } catch (InterruptedException ie) {};
                    // Loop back
                    break;
                
                case ESTABLISHED:
                    synchronized (writeEvent) {
                        if (!writeLocked) {
                            writeLocked = true;
                            return true;
                        }
                    
                        try {
                            writeEvent.wait();
                        } catch (InterruptedException ie) {};
                    }
                    // Loop back
                    break;
                
                    //
                    // XXX
                    // Need to distinguish between client and server roles
                    // here probably.
                    //
                case ABORT:
                    throw new COMM_FAILURE( MinorCodes.WRITE_ERROR_SEND,
                                            CompletionStatus.COMPLETED_NO);
                     
                case CLOSE_RECVD:
                    // the connection has been closed or closing
                    // ==> throw rebind exception
                
                    throw new COMM_FAILURE( MinorCodes.CONN_CLOSE_REBIND,
                                            CompletionStatus.COMPLETED_NO);
                
                default:
                    if (orb.transportDebugFlag)
                        dprint("Connection:writeLock: weird state");
                
                    delete(Connection.CONN_ABORT);
                    return false;
                }
            }
        }
    }

    /**
     * Release the write lock on this connection.
     */
    protected void writeUnlock()
    {
        synchronized (writeEvent) {
            writeLocked = false;
            writeEvent.notify(); // wake up one guy waiting to write
        }
    }


    public void delete()
    {
        delete(Connection.CONN_ABORT);
    }

    void delete(int code)
    {
        DeleteConn dc = new DeleteConn(code);
        reader.stop(dc);
    }


    /** Send a two-way IIOP message to the server.
     */
    public IIOPInputStream invoke(IIOPOutputStream s)
        throws SystemException
    {
        return send(s, false);
    }

    /**
     * In 1.1 and 1.2, it seeds the response to be
     * continued if it's not the last fragment.
     */
    IIOPInputStream getResponse(boolean isOneway, int requestID)
    {

        IIOPInputStream returnStream = null;

        Integer requestId = new Integer(requestID);

        OutCallDesc call = (OutCallDesc)out_calls.get(requestId);

        if (isOneway) {
            out_calls.remove(requestId);
            return null;
        }

        // It's very important that only the client thread
        // removes its OutCallDesc from the table.
        if (call == null)
            throw new INTERNAL(MinorCodes.NULL_OUT_CALL,
                               CompletionStatus.COMPLETED_MAYBE);

        synchronized(call.done) {

            while (call.s == null && call.exc == null) {
                // Wait for the reply from the server.
                // The ReaderThread reads in the reply IIOP message
                // and signals us.
                try {
                    call.done.wait();
                } catch (InterruptedException ie) {};
            }

            // Remove this request ID from the out call descriptor map.
            // The ReaderThread can continue to add fragments because it
            // still has access to the input stream via the clientReplyMap.
            out_calls.remove(requestId);

            if (call.exc != null) {
                throw call.exc;
            }

            returnStream = call.s;
        }

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

        // If the header was already unmarshaled, this won't
        // do anything
        if (returnStream != null)
            returnStream.unmarshalHeader();