jchannel.java

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

        }
        catch(QueueClosedException queue_closed) {
            throw new ChannelClosedException();
        }
        catch(TimeoutException t) {
            throw t;
        }
        catch(Exception e) {
            if(log.isErrorEnabled()) log.error("exception: " + e);
            return null;
        }
    }


    /**
     * Just peeks at the next message, view or block. Does <em>not</em> install
     * new view if view is received<BR>
     * Does the same thing as JChannel.receive but doesn't remove the object from the
     * receiver queue
     */
    public Object peek(long timeout) throws ChannelNotConnectedException, ChannelClosedException, TimeoutException {

        checkClosed();
        checkNotConnected();

        try {
            Event evt=(timeout <= 0)? (Event)mq.peek() : (Event)mq.peek(timeout);
            Object retval=getEvent(evt);
            evt=null;
            return retval;
        }
        catch(QueueClosedException queue_closed) {
            if(log.isErrorEnabled()) log.error("exception: " + queue_closed);
            return null;
        }
        catch(TimeoutException t) {
            return null;
        }
        catch(Exception e) {
            if(log.isErrorEnabled()) log.error("exception: " + e);
            return null;
        }
    }




    /**
     * Returns the current view.
     * <BR>
     * If the channel is not connected or if it is closed it will return null.
     * <BR>
     * @return returns the current group view, or null if the channel is closed or disconnected
     */
    public View getView() {
        return closed || !connected ? null : my_view;
    }


    /**
     * returns the local address of the channel
     * returns null if the channel is closed
     */
    public Address getLocalAddress() {
        return closed ? null : local_addr;
    }


    /**
     * returns the name of the channel
     * if the channel is not connected or if it is closed it will return null
     * @deprecated Use {@link #getClusterName()} instead
     */
    public String getChannelName() {
        return closed ? null : !connected ? null : cluster_name;
    }

    public String getClusterName() {
        return cluster_name;
    }


    /**
     * Sets a channel option.  The options can be one of the following:
     * <UL>
     * <LI>    Channel.BLOCK
     * <LI>    Channel.LOCAL
     * <LI>    Channel.AUTO_RECONNECT
     * <LI>    Channel.AUTO_GETSTATE
     * </UL>
     * <P>
     * There are certain dependencies between the options that you can set,
     * I will try to describe them here.
     * <P>
     * Option: Channel.BLOCK<BR>
     * Value:  java.lang.Boolean<BR>
     * Result: set to true will set setOpt(VIEW, true) and the JChannel will receive BLOCKS and VIEW events<BR>
     *<BR>
     * Option: LOCAL<BR>
     * Value:  java.lang.Boolean<BR>
     * Result: set to true the JChannel will receive messages that it self sent out.<BR>
     *<BR>
     * Option: AUTO_RECONNECT<BR>
     * Value:  java.lang.Boolean<BR>
     * Result: set to true and the JChannel will try to reconnect when it is being closed<BR>
     *<BR>
     * Option: AUTO_GETSTATE<BR>
     * Value:  java.lang.Boolean<BR>
     * Result: set to true, the AUTO_RECONNECT will be set to true and the JChannel will try to get the state after a close and reconnect happens<BR>
     * <BR>
     *
     * @param option the parameter option Channel.VIEW, Channel.SUSPECT, etc
     * @param value the value to set for this option
     *
     */
    public void setOpt(int option, Object value) {
        if(closed) {
            if(log.isWarnEnabled()) log.warn("channel is closed; option not set !");
            return;
        }

        switch(option) {
            case VIEW:
                if(log.isWarnEnabled())
                    log.warn("option VIEW has been deprecated (it is always true now); this option is ignored");
                break;
            case SUSPECT:
                if(log.isWarnEnabled())
                    log.warn("option SUSPECT has been deprecated (it is always true now); this option is ignored");
                break;
            case BLOCK:
                if(value instanceof Boolean)
                    receive_blocks=((Boolean)value).booleanValue();
                else
                    if(log.isErrorEnabled()) log.error("option " + Channel.option2String(option) +
                                                     " (" + value + "): value has to be Boolean");
                break;

            case GET_STATE_EVENTS:
                if(log.isWarnEnabled())
                    log.warn("option GET_STATE_EVENTS has been deprecated (it is always true now); this option is ignored");
                break;

            case LOCAL:
                if(value instanceof Boolean)
                    receive_local_msgs=((Boolean)value).booleanValue();
                else
                    if(log.isErrorEnabled()) log.error("option " + Channel.option2String(option) +
                                                     " (" + value + "): value has to be Boolean");
                break;

            case AUTO_RECONNECT:
                if(value instanceof Boolean)
                    auto_reconnect=((Boolean)value).booleanValue();
                else
                    if(log.isErrorEnabled()) log.error("option " + Channel.option2String(option) +
                                                     " (" + value + "): value has to be Boolean");
                break;

            case AUTO_GETSTATE:
                if(value instanceof Boolean) {
                    auto_getstate=((Boolean)value).booleanValue();
                    if(auto_getstate)
                        auto_reconnect=true;
                }
                else
                    if(log.isErrorEnabled()) log.error("option " + Channel.option2String(option) +
                                                     " (" + value + "): value has to be Boolean");
                break;

            default:
                if(log.isErrorEnabled()) log.error("option " + Channel.option2String(option) + " not known");
                break;
        }
    }


    /**
     * returns the value of an option.
     * @param option the option you want to see the value for
     * @return the object value, in most cases java.lang.Boolean
     * @see JChannel#setOpt
     */
    public Object getOpt(int option) {
        switch(option) {
            case VIEW:
            	return Boolean.TRUE;
            case BLOCK:
            	return receive_blocks ? Boolean.TRUE : Boolean.FALSE;
            case SUSPECT:
            	return Boolean.TRUE;
            case AUTO_RECONNECT:
                return auto_reconnect ? Boolean.TRUE : Boolean.FALSE;
            case AUTO_GETSTATE:
                return auto_getstate ? Boolean.TRUE : Boolean.FALSE;
            case GET_STATE_EVENTS:
                return Boolean.TRUE;
            case LOCAL:
            	return receive_local_msgs ? Boolean.TRUE : Boolean.FALSE;
            default:
                if(log.isErrorEnabled()) log.error("option " + Channel.option2String(option) + " not known");
                return null;
        }
    }


    /**
     * Called to acknowledge a block() (callback in <code>MembershipListener</code> or
     * <code>BlockEvent</code> received from call to <code>receive()</code>).
     * After sending blockOk(), no messages should be sent until a new view has been received.
     * Calling this method on a closed channel has no effect.
     */
    public void blockOk() {
        down(new Event(Event.BLOCK_OK));
        down(new Event(Event.START_QUEUEING));
    }


    /**
     * Retrieves the current group state. Sends GET_STATE event down to STATE_TRANSFER layer.
     * Blocks until STATE_TRANSFER sends up a GET_STATE_OK event or until <code>timeout</code>
     * milliseconds have elapsed. The argument of GET_STATE_OK should be a single object.
     * @param target the target member to receive the state from. if null, state is retrieved from coordinator
     * @param timeout the number of milliseconds to wait for the operation to complete successfully. 0 waits until
     * the state has been received
     * @return true of the state was received, false if the operation timed out
     */
    public boolean getState(Address target, long timeout) throws ChannelNotConnectedException, ChannelClosedException {
        return getState(target,null,timeout);
    }

    /**
     * Retrieves a substate (or partial state) from the target.
     * @param target State provider. If null, coordinator is used
     * @param state_id The ID of the substate. If null, the entire state will be transferred
     * @param timeout the number of milliseconds to wait for the operation to complete successfully. 0 waits until
     * the state has been received
     * @return
     * @throws ChannelNotConnectedException
     * @throws ChannelClosedException
     */
    public boolean getState(Address target, String state_id, long timeout) throws ChannelNotConnectedException, ChannelClosedException {
        if(target == null)
            target=determineCoordinator();
        if(target != null && local_addr != null && target.equals(local_addr)) {
            if(log.isTraceEnabled())
                log.trace("cannot get state from myself (" + target + "): probably the first member");
            return false;
        }

        StateTransferInfo info=new StateTransferInfo(target, state_id, timeout);
        boolean rc=_getState(new Event(Event.GET_STATE, info), info);
        if(rc == false)
            down(new Event(Event.RESUME_STABLE));
        return rc;
    }


    /**
     * Retrieves the current group state. Sends GET_STATE event down to STATE_TRANSFER layer.
     * Blocks until STATE_TRANSFER sends up a GET_STATE_OK event or until <code>timeout</code>
     * milliseconds have elapsed. The argument of GET_STATE_OK should be a vector of objects.
     * @param targets - the target members to receive the state from ( an Address list )
     * @param timeout - the number of milliseconds to wait for the operation to complete successfully
     * @return true of the state was received, false if the operation timed out
     * @deprecated Not really needed - we always want to get the state from a single member,
     * use {@link #getState(org.jgroups.Address, long)} instead
     */
    public boolean getAllStates(Vector targets, long timeout) throws ChannelNotConnectedException, ChannelClosedException {
        throw new UnsupportedOperationException("use getState() instead");
    }


    /**
     * Called by the application is response to receiving a <code>getState()</code> object when
     * calling <code>receive()</code>.
     * When the application receives a getState() message on the receive() method,
     * it should call returnState() to reply with the state of the application
     * @param state The state of the application as a byte buffer
     *              (to send over the network).
     */
    public void returnState(byte[] state) {
        StateTransferInfo info=new StateTransferInfo(null, null, 0L, state);
        down(new Event(Event.GET_APPLSTATE_OK, info));
    }

    /**
     * Returns a substate as indicated by state_id
     * @param state
     * @param state_id
     */
    public void returnState(byte[] state, String state_id) {
        StateTransferInfo info=new StateTransferInfo(null, state_id, 0L, state);
        down(new Event(Event.GET_APPLSTATE_OK, info));
    }





    /**
     * Callback method <BR>
     * Called by the ProtocolStack when a message is received.
     * It will be added to the message queue from which subsequent
     * <code>Receive</code>s will dequeue it.
     * @param evt the event carrying the message from the protocol stack
     */
    public void up(Event evt) {
        int     type=evt.getType();
        Message msg;


        switch(type) {

        case Event.MSG:
            msg=(Message)evt.getArg();
            if(!receive_local_msgs) {  // discard local messages (sent by myself to me)
                if(local_addr != null && msg.getSrc() != null)
                    if(local_addr.equals(msg.getSrc()))
                        return;
            }
            break;

        case Event.VIEW_CHANGE: