protocol.java

JGRoups源码

     * This method is called on a {@link org.jgroups.Channel#disconnect()}. Stops work (e.g. by closing multicast socket).
     * Will be called <em>from top to bottom</em>. This means that at the time of the method invocation the
     * neighbor protocol below is still working. This method will replace the
     * <b>STOP</b>, <b>STOP_OK</b>, <b>CLEANUP</b> and <b>CLEANUP_OK</b> events. The ProtocolStack guarantees that
     * when this method is called all messages in the down queue will have been flushed
     */
    public void stop() {
    }


    /**
     * This method is called on a {@link org.jgroups.Channel#close()}.
     * Does some cleanup; after the call the VM will terminate
     */
    public void destroy() {
    }


    public Queue getUpQueue() {
        return up_queue;
    }    // used by Debugger (ProtocolView)

    public Queue getDownQueue() {
        return down_queue;
    }  // used by Debugger (ProtocolView)


    /** List of events that are required to be answered by some layer above.
     @return Vector (of Integers) */
    public Vector requiredUpServices() {
        return null;
    }

    /** List of events that are required to be answered by some layer below.
     @return Vector (of Integers) */
    public Vector requiredDownServices() {
        return null;
    }

    /** List of events that are provided to layers above (they will be handled when sent down from
     above).
     @return Vector (of Integers) */
    public Vector providedUpServices() {
        return null;
    }

    /** List of events that are provided to layers below (they will be handled when sent down from
     below).
     @return Vector (of Integers) */
    public Vector providedDownServices() {
        return null;
    }


    public abstract String getName();   // all protocol names have to be unique !

    public Protocol getUpProtocol() {
        return up_prot;
    }

    public Protocol getDownProtocol() {
        return down_prot;
    }

    public void setUpProtocol(Protocol up_prot) {
        this.up_prot=up_prot;
    }

    public void setDownProtocol(Protocol down_prot) {
        this.down_prot=down_prot;
    }

    public void setProtocolStack(ProtocolStack stack) {
        this.stack=stack;
    }


    /** Used internally. If overridden, call this method first. Only creates the up_handler thread
     if down_thread is true */
    public void startUpHandler() {
        if(up_thread) {
            if(up_handler == null) {
                up_handler=new UpHandler(up_queue, this, observer);
                if(up_thread_prio >= 0) {
                    try {
                        up_handler.setPriority(up_thread_prio);
                    }
                    catch(Throwable t) {
                        if(log.isErrorEnabled()) log.error("priority " + up_thread_prio +
                                " could not be set for thread", t);
                    }
                }
                up_handler.start();
            }
        }
    }


    /** Used internally. If overridden, call this method first. Only creates the down_handler thread
     if down_thread is true */
    public void startDownHandler() {
        if(down_thread) {
            if(down_handler == null) {
                down_handler=new DownHandler(down_queue, this, observer);
                if(down_thread_prio >= 0) {
                    try {
                        down_handler.setPriority(down_thread_prio);
                    }
                    catch(Throwable t) {
                        if(log.isErrorEnabled()) log.error("priority " + down_thread_prio +
                                " could not be set for thread", t);
                    }
                }
                down_handler.start();
            }
        }
    }


    /** Used internally. If overridden, call parent's method first */
    public void stopInternal() {
        up_queue.close(false);  // this should terminate up_handler thread

        if(up_handler != null && up_handler.isAlive()) {
            try {
                up_handler.join(THREAD_JOIN_TIMEOUT);
            }
            catch(Exception ex) {
            }
            if(up_handler != null && up_handler.isAlive()) {
                up_handler.interrupt();  // still alive ? let's just kill it without mercy...
                try {
                    up_handler.join(THREAD_JOIN_TIMEOUT);
                }
                catch(Exception ex) {
                }
                if(up_handler != null && up_handler.isAlive())
                    if(log.isErrorEnabled()) log.error("up_handler thread for " + getName() +
                                                           " was interrupted (in order to be terminated), but is still alive");
            }
        }
        up_handler=null;

        down_queue.close(false); // this should terminate down_handler thread
        if(down_handler != null && down_handler.isAlive()) {
            try {
                down_handler.join(THREAD_JOIN_TIMEOUT);
            }
            catch(Exception ex) {
            }
            if(down_handler != null && down_handler.isAlive()) {
                down_handler.interrupt(); // still alive ? let's just kill it without mercy...
                try {
                    down_handler.join(THREAD_JOIN_TIMEOUT);
                }
                catch(Exception ex) {
                }
                if(down_handler != null && down_handler.isAlive())
                    if(log.isErrorEnabled()) log.error("down_handler thread for " + getName() +
                                                           " was interrupted (in order to be terminated), but is is still alive");
            }
        }
        down_handler=null;
    }


    /**
     * Internal method, should not be called by clients. Used by ProtocolStack. I would have
     * used the 'friends' modifier, but this is available only in C++ ... If the up_handler thread
     * is not available (down_thread == false), then directly call the up() method: we will run on the
     * caller's thread (e.g. the protocol layer below us).
     */
    protected void receiveUpEvent(Event evt) {
        if(up_handler == null) {
            if(observer != null) {                               // call debugger hook (if installed)
                if(observer.up(evt, up_queue.size()) == false) {  // false means discard event
                    return;
                }
            }
            up(evt);
            return;
        }
        try {
            up_queue.add(evt);
        }
        catch(Exception e) {
            if(log.isWarnEnabled()) log.warn("exception: " + e);
        }
    }

    /**
     * Internal method, should not be called by clients. Used by ProtocolStack. I would have
     * used the 'friends' modifier, but this is available only in C++ ... If the down_handler thread
     * is not available (down_thread == false), then directly call the down() method: we will run on the
     * caller's thread (e.g. the protocol layer above us).
     */
    protected void receiveDownEvent(Event evt) {
        if(down_handler == null) {
            if(observer != null) {                                    // call debugger hook (if installed)
                if(observer.down(evt, down_queue.size()) == false) {  // false means discard event
                    return;
                }
            }
            int type=evt.getType();
            if(type == Event.START || type == Event.STOP) {
                if(handleSpecialDownEvent(evt) == false)
                    return;
            }
            down(evt);
            return;
        }
        try {
            down_queue.add(evt);
        }
        catch(Exception e) {
            if(log.isWarnEnabled()) log.warn("exception: " + e);
        }
    }

    /**
     * Causes the event to be forwarded to the next layer up in the hierarchy. Typically called
     * by the implementation of <code>Up</code> (when done).
     */
    public void passUp(Event evt) {
        if(observer != null) {                   // call debugger hook (if installed)
            if(observer.passUp(evt) == false) {  // false means don't pass up (=discard) event
                return;
            }
        }
        up_prot.receiveUpEvent(evt);
    }

    /**
     * Causes the event to be forwarded to the next layer down in the hierarchy.Typically called
     * by the implementation of <code>Down</code> (when done).
     */
    public void passDown(Event evt) {
        if(observer != null) {                     // call debugger hook (if installed)
            if(observer.passDown(evt) == false) {  // false means don't pass down (=discard) event
                return;
            }
        }
        down_prot.receiveDownEvent(evt);
    }


    /**
     * An event was received from the layer below. Usually the current layer will want to examine
     * the event type and - depending on its type - perform some computation
     * (e.g. removing headers from a MSG event type, or updating the internal membership list
     * when receiving a VIEW_CHANGE event).
     * Finally the event is either a) discarded, or b) an event is sent down
     * the stack using <code>passDown()</code> or c) the event (or another event) is sent up
     * the stack using <code>passUp()</code>.
     */
    public void up(Event evt) {
        passUp(evt);
    }

    /**
     * An event is to be sent down the stack. The layer may want to examine its type and perform
     * some action on it, depending on the event's type. If the event is a message MSG, then
     * the layer may need to add a header to it (or do nothing at all) before sending it down
     * the stack using <code>passDown()</code>. In case of a GET_ADDRESS event (which tries to
     * retrieve the stack's address from one of the bottom layers), the layer may need to send
     * a new response event back up the stack using <code>passUp()</code>.
     */
    public void down(Event evt) {
        passDown(evt);
    }


    /**  These are special internal events that should not be handled by protocols
     * @return boolean True: the event should be passed further down the stack. False: the event should
     * be discarded (not passed down the stack)
     */
    protected boolean handleSpecialDownEvent(Event evt) {
        switch(evt.getType()) {
            case Event.START:
                try {
                    start();

                    // if we're the transport protocol, reply with a START_OK up the stack
                    if(down_prot == null) {
                        passUp(new Event(Event.START_OK, Boolean.TRUE));
                        return false; // don't pass down the stack
                    }
                    else
                        return true; // pass down the stack
                }
                catch(Exception e) {
                    passUp(new Event(Event.START_OK, new Exception("exception caused by " + getName() + ".start()", e)));
                    return false;
                }
            case Event.STOP:
                stop();
                if(down_prot == null) {
                    passUp(new Event(Event.STOP_OK, Boolean.TRUE));
                    return false; // don't pass down the stack
                }
                else
                    return true; // pass down the stack
            default:
                return true; // pass down by default
        }
    }
}