votingadapter.java

来自「JGRoups源码」· Java 代码 · 共 505 行 · 第 1/2 页

        }

        // it is always interesting to know the class that caused failure...
        if (!(object instanceof VoteResult)) {
            String faultClass = object.getClass().getName();

            // ...but we do not handle byzantine faults
            throw new ChannelException("Node " + response.getSender() +
                       " generated fault (class " + faultClass + ')');
        }

        // what if we received the response from faulty node?
        if (object instanceof FailureVoteResult) {

            if(log.isErrorEnabled()) log.error(((FailureVoteResult)object).getReason());

            return PROCESS_BREAK;
        }

        // everything is fine :)
        return PROCESS_CONTINUE;
    }

    /**
     * Callback for notification about the new view of the group.
     */
    public void viewAccepted(View newView) {

        // clean nodes that were suspected but still exist in new view
        Iterator iterator = suspectedNodes.iterator();
        while(iterator.hasNext()) {
            Address suspectedNode = (Address)iterator.next();
            if (newView.containsMember(suspectedNode))
                iterator.remove();
        }

        for(Iterator it=membership_listeners.iterator(); it.hasNext();) {
            MembershipListener listener=(MembershipListener)it.next();
            try {
                listener.viewAccepted(newView);
            }
            catch(Throwable t) {
                if(log.isErrorEnabled())
                    log.error("failed calling viewAccepted() on " + listener, t);
            }
        }
    }

    /**
     * Callback for notification that one node is suspected
     */
    public void suspect(Address suspected) {
        suspectedNodes.add(suspected);
        for(Iterator it=membership_listeners.iterator(); it.hasNext();) {
            MembershipListener listener=(MembershipListener)it.next();
            try {
                listener.suspect(suspected);
            }
            catch(Throwable t) {
                if(log.isErrorEnabled())
                    log.error("failed calling suspect() on " + listener, t);
            }
        }
    }

    /**
     * Blocks the channel until the ViewAccepted is invoked.
     */
    public void block() {
        for(Iterator it=membership_listeners.iterator(); it.hasNext();) {
            MembershipListener listener=(MembershipListener)it.next();
            try {
                listener.block();
            }
            catch(Throwable t) {
                if(log.isErrorEnabled())
                    log.error("failed calling block() on " + listener, t);
            }
        }
    }

    /**
     * Get the channel state.
     *
     * @return always <code>null</code>, we do not have any group-shared
     * state.
     */
    public byte[] getState() {
        return null;
    }

    /**
     * Receive the message. All messages are ignored.
     *
     * @param msg message to check.
     */
    public void receive(org.jgroups.Message msg) {
        // do nothing
    }

    /**
     * Set the channel state. We do nothing here.
     */
    public void setState(byte[] state) {
        // ignore the state, we do not have any.
    }

    private final Set voteListeners = new HashSet();
    private VotingListener[] listeners;

    /**
     * Vote on the specified decree requiring all nodes to vote.
     * 
     * @param decree decree on which nodes should vote.
     * @param timeout time during which nodes can vote.
     * 
     * @return <code>true</code> if nodes agreed on a decree, otherwise 
     * <code>false</code>
     * 
     * @throws ChannelException if something went wrong.
     */
    public boolean vote(Object decree, long timeout) throws ChannelException {
        return vote(decree, timeout, null);
    }

    /**
     * Vote on the specified decree requiring all nodes to vote.
     * 
     * @param decree decree on which nodes should vote.
     * @param timeout time during which nodes can vote.
     * @param voteResponseProcessor processor which will be called for every response that is received.
     * 
     * @return <code>true</code> if nodes agreed on a decree, otherwise 
     * <code>false</code>
     * 
     * @throws ChannelException if something went wrong.
     */
    public boolean vote(Object decree, long timeout, VoteResponseProcessor voteResponseProcessor) throws ChannelException {
        return vote(decree, VOTE_ALL, timeout, voteResponseProcessor);
    }

    /**
     * Adds voting listener.
     */
    public void addVoteListener(VotingListener listener) {
        voteListeners.add(listener);
        listeners = (VotingListener[])voteListeners.toArray(
                                new VotingListener[voteListeners.size()]);
    }

    /**
     * Removes voting listener.
     */
    public void removeVoteListener(VotingListener listener) {
        voteListeners.remove(listener);

        listeners = (VotingListener[])voteListeners.toArray(
                                new VotingListener[voteListeners.size()]);
    }

    /**
     * This method performs voting on the specific decree between all
     * local voteListeners.
     */
    public VoteResult localVote(Object decree) {

        VoteResult voteResult = new VoteResult();

        for(int i = 0; i < listeners.length; i++) {
            VotingListener listener = listeners[i];

            try {
                voteResult.addVote(listener.vote(decree));
            } catch (VoteException vex) {
                // do nothing here.
            } catch(RuntimeException ex) {

                if(log.isErrorEnabled()) log.error(ex.toString());

                // if we are here, then listener 
                // had thrown a RuntimeException
                return new FailureVoteResult(ex.getMessage());
            }
        }


            if(log.isDebugEnabled()) log.debug("Voting on decree " + decree.toString() + " : " +
            voteResult.toString());

        return voteResult;
    }

    /**
     * Convert consensus type into string representation. This method is 
     * useful for debugginf.
     * 
     * @param consensusType type of the consensus.
     * 
     * @return string representation of the consensus type.
     */
    public static String getConsensusStr(int consensusType) {
        switch(consensusType) {
    case VotingAdapter.VOTE_ALL : return "VOTE_ALL";
    case VotingAdapter.VOTE_ANY : return "VOTE_ANY";
    case VotingAdapter.VOTE_MAJORITY : return "VOTE_MAJORITY";
    default : return "UNKNOWN";
        }
    }

    /**
     * This class represents the result of local voting. It contains a 
     * number of positive and negative votes collected during local voting.
     */
    public static class VoteResult implements Serializable {
        private int positiveVotes = 0;
        private int negativeVotes = 0;
        private static final long serialVersionUID = 2868605599965196746L;

        public void addVote(boolean vote) {
            if (vote)
                positiveVotes++;
            else
                negativeVotes++;
        }

        public int getPositiveVotes() { return positiveVotes; }

        public int getNegativeVotes() { return negativeVotes; }

        public String toString() {
            return "VoteResult: up=" + positiveVotes +
                ", down=" + negativeVotes;
        }
    }

    /**
     * Class that represents a result of local voting on the failed node.
     */
    public static class FailureVoteResult extends VoteResult {
        private final String reason;

        public FailureVoteResult(String reason) {
            this.reason = reason;
        }

        public String getReason() {
            return reason;
        }
    }

}