jchannel.java

来自「JGRoups源码」· Java 代码 · 共 1,619 行 · 第 1/5 页

    /**
     * health check.<BR>
     * throws a ChannelNotConnected exception if the channel is not connected
     */
    protected void checkNotConnected() throws ChannelNotConnectedException {
        if(!connected)
            throw new ChannelNotConnectedException();
    }

    /**
     * health check<BR>
     * throws a ChannelClosed exception if the channel is closed
     */
    protected void checkClosed() throws ChannelClosedException {
        if(closed)
            throw new ChannelClosedException();
    }



    /**
     * returns the value of the event<BR>
     * These objects will be returned<BR>
     * <PRE>
     * <B>Event Type    - Return Type</B>
     * Event.MSG           - returns a Message object
     * Event.VIEW_CHANGE   - returns a View object
     * Event.SUSPECT       - returns a SuspectEvent object
     * Event.BLOCK         - returns a new BlockEvent object
     * Event.GET_APPLSTATE - returns a GetStateEvent object
     * Event.STATE_RECEIVED- returns a SetStateEvent object
     * Event.Exit          - returns an ExitEvent object
     * All other           - return the actual Event object
     * </PRE>
     * @param   evt - the event of which you want to extract the value
     * @return the event value if it matches the select list,
     *         returns null if the event is null
     *         returns the event itself if a match (See above) can not be made of the event type
     */
    static Object getEvent(Event evt) {
        if(evt == null)
            return null; // correct ?

        switch(evt.getType()) {
            case Event.MSG:
                return evt.getArg();
            case Event.VIEW_CHANGE:
                return evt.getArg();
            case Event.SUSPECT:
                return new SuspectEvent(evt.getArg());
            case Event.BLOCK:
                return new BlockEvent();
            case Event.UNBLOCK:
                return new UnblockEvent();
            case Event.GET_APPLSTATE:
                StateTransferInfo info=(StateTransferInfo)evt.getArg();
                return new GetStateEvent(info.target, info.state_id);
            case Event.STATE_RECEIVED:
                info=(StateTransferInfo)evt.getArg();
                return new SetStateEvent(info.state, info.state_id);
            case Event.STATE_TRANSFER_OUTPUTSTREAM:
            	info = (StateTransferInfo)evt.getArg();
                return new StreamingGetStateEvent(info.outputStream,info.state_id);
            case Event.STATE_TRANSFER_INPUTSTREAM:
            	info = (StateTransferInfo)evt.getArg();
                return new StreamingSetStateEvent(info.inputStream,info.state_id);
            case Event.EXIT:
                return new ExitEvent();
            default:
                return evt;
        }
    }


    /**
     * Receives the state from the group and modifies the JChannel.state object<br>
     * This method initializes the local state variable to null, and then sends the state
     * event down the stack. It waits for a GET_STATE_OK event to bounce back
     * @param evt the get state event, has to be of type Event.GET_STATE
     * @param info Information about the state transfer, e.g. target member and timeout
     * @return true of the state was received, false if the operation timed out
     */
    private boolean _getState(Event evt, StateTransferInfo info) throws ChannelNotConnectedException, ChannelClosedException {
        checkClosed();
        checkNotConnected();
        if(!state_transfer_supported) {
            throw new IllegalStateException("fetching state will fail as state transfer is not supported. "
                    + "Add one of the STATE_TRANSFER protocols to your protocol configuration");
        }

        if(flush_supported)
           flush_unblock_promise.reset();

        state_promise.reset();
        down(evt);
        Boolean state_transfer_successfull=(Boolean)state_promise.getResult(info.timeout);

        //if FLUSH is used do not return from getState() until UNBLOCK event is received
        boolean shouldWaitForUnblock = flush_supported && receive_blocks;
        if(shouldWaitForUnblock){
           try{
              flush_unblock_promise.getResultWithTimeout(FLUSH_UNBLOCK_TIMEOUT);
           }
           catch (TimeoutException te){
              if(log.isWarnEnabled())
                 log.warn("Waiting on UNBLOCK after getState timed out");
           }
        }

        return state_transfer_successfull != null && state_transfer_successfull.booleanValue();
    }


    /**
     * Disconnects and closes the channel.
     * This method does the folloing things
     * <ol>
     * <li>Calls <code>this.disconnect</code> if the disconnect parameter is true
     * <li>Calls <code>Queue.close</code> on mq if the close_mq parameter is true
     * <li>Calls <code>ProtocolStack.stop</code> on the protocol stack
     * <li>Calls <code>ProtocolStack.destroy</code> on the protocol stack
     * <li>Sets the channel closed and channel connected flags to true and false
     * <li>Notifies any channel listener of the channel close operation
     * </ol>
     */
    protected void _close(boolean disconnect, boolean close_mq) {
        if(closed)
            return;

        if(disconnect)
            disconnect();                     // leave group if connected

        if(close_mq) {
            try {
                if(mq != null)
                    mq.close(false);              // closes and removes all messages
            }
            catch(Exception e) {
                if(log.isErrorEnabled()) log.error("exception: " + e);
            }
        }

        if(prot_stack != null) {
            try {
                prot_stack.stopStack();
                prot_stack.destroy();
            }
            catch(Exception e) {
                if(log.isErrorEnabled()) log.error("failed destroying the protocol stack", e);
            }
        }
        closed=true;
        connected=false;
        notifyChannelClosed(this);
        init(); // sets local_addr=null; changed March 18 2003 (bela) -- prevented successful rejoining
    }


    public final void closeMessageQueue(boolean flush_entries) {
        if(mq != null)
            mq.close(flush_entries);
    }


    /**
     * Creates a separate thread to close the protocol stack.
     * This is needed because the thread that called JChannel.up() with the EXIT event would
     * hang waiting for up() to return, while up() actually tries to kill that very thread.
     * This way, we return immediately and allow the thread to terminate.
     */
    private void handleExit(Event evt) {
        notifyChannelShunned();
        if(closer != null && !closer.isAlive())
            closer=null;
        if(closer == null) {
            if(log.isInfoEnabled())
                log.info("received an EXIT event, will leave the channel");
            closer=new CloserThread(evt);
            closer.start();
        }
    }

    public boolean flushSupported() {
        return flush_supported;
    }

    /**
     * Will berform a flush of the system, ie. all pending messages are flushed out of the system and all members ack
     * their reception. After this call return, no member will be sending any messages until {@link #stopFlush()} is
     * called.
     * @param timeout
     * @param automatic_resume Call {@link #stopFlush()} after the flush
     * @return true if FLUSH completed within the timeout
     */
    public boolean startFlush(long timeout, boolean automatic_resume) {
        if(!flush_supported) {
            throw new IllegalStateException("Flush is not supported, add pbcast.FLUSH protocol to your configuration");
        }

        boolean successfulFlush=false;
        down(new Event(Event.SUSPEND));
        try {
            flush_promise.reset();
            flush_promise.getResultWithTimeout(timeout);
            successfulFlush=true;
        }
        catch(TimeoutException e) {
        }

        if(automatic_resume)
            stopFlush();

        return successfulFlush;
    }

    public void stopFlush() {
        if(!flush_supported) {
            throw new IllegalStateException("Flush is not supported, add pbcast.FLUSH protocol to your configuration");
        }
        down(new Event(Event.RESUME));
    }

    Address determineCoordinator() {
        Vector mbrs=my_view != null? my_view.getMembers() : null;
        if(mbrs == null)
            return null;
        if(mbrs.size() > 0)
            return (Address)mbrs.firstElement();
        return null;
    }

    /* ------------------------------- End of Private Methods ---------------------------------- */


    class CloserThread extends Thread {
        final Event evt;
        final Thread t=null;


        CloserThread(Event evt) {
            super(Util.getGlobalThreadGroup(), "CloserThread");
            this.evt=evt;
            setDaemon(true);
        }


        public void run() {
            try {
                String old_cluster_name=cluster_name; // remember because close() will null it
                if(log.isInfoEnabled())
                    log.info("closing the channel");
                _close(false, false); // do not disconnect before closing channel, do not close mq (yet !)

                if(up_handler != null)
                    up_handler.up(this.evt);
                else {
                    try {
                        if(receiver == null)
                            mq.add(this.evt);
                    }
                    catch(Exception ex) {
                        if(log.isErrorEnabled()) log.error("exception: " + ex);
                    }
                }

                if(mq != null) {
                    Util.sleep(500); // give the mq thread a bit of time to deliver EXIT to the application
                    try {
                        mq.close(false);
                    }
                    catch(Exception ex) {
                    }
                }

                if(auto_reconnect) {
                    try {
                        if(log.isInfoEnabled()) log.info("reconnecting to group " + old_cluster_name);
                        open();
                    }
                    catch(Exception ex) {
                        if(log.isErrorEnabled()) log.error("failure reopening channel: " + ex);
                        return;
                    }
                    try {
                        if(additional_data != null) {
                            // send previously set additional_data down the stack - other protocols (e.g. TP) use it
                            Map m=new HashMap(11);
                            m.put("additional_data", additional_data);
                            down(new Event(Event.CONFIG, m));
                        }
                        connect(old_cluster_name);
                        notifyChannelReconnected(local_addr);
                    }
                    catch(Exception ex) {
                        if(log.isErrorEnabled()) log.error("failure reconnecting to channel: " + ex);
                        return;
                    }
                }

                if(auto_getstate) {
                    if(log.isInfoEnabled())
                        log.info("fetching the state (auto_getstate=true)");
                    boolean rc=JChannel.this.getState(null, GET_STATE_DEFAULT_TIMEOUT);
                    if(log.isInfoEnabled()) {
                        if(rc)
                            log.info("state was retrieved successfully");
                        else
                            log.info("state transfer failed");
                    }
                }

            }
            catch(Exception ex) {
                if(log.isErrorEnabled()) log.error("exception: " + ex);
            }
            finally {
                closer=null;
            }
        }
    }

}