import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
/**
 * Returns the address of the node we think is the leader.
 */
protected InetSocketAddress findLeader() {
    InetSocketAddress addr = null;
    // Find the leader by id
    Vote current = self.getCurrentVote();
    for (QuorumServer s : self.getView().values()) {
        if (s.id == current.getId()) {
            // Ensure we have the leader's correct IP address before
            // attempting to connect.
            s.recreateSocketAddresses();
            addr = s.addr;
            break;
        }
    }
    if (addr == null) {
        LOG.warn("Couldn't find the leader with id = "
                + current.getId());
    }
    return addr;
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
private void fakeConfigureServers(int count) throws IOException {
    peers = new HashMap<Long,QuorumServer>();
    qps = new QuorumPeer[count];
    qpsDirs = new File[count];
    for(int i = 1; i <= count; i++) {
        peers.put(Long.valueOf(i), new QuorumServer(
            i, "127.0.0.1", fakeBasePort + i, serverCount + fakeBasePort + i, null));
    }
    StringBuilder sb = new StringBuilder();
    for(int i = 0; i < count; i++) {
        //make that testData exists otherwise it fails on windows
        testData.mkdirs();
        qpsDirs[i] = File.createTempFile("sysTest", ".tmp", testData);
        qpsDirs[i].delete();
        qpsDirs[i].mkdir();
        int port = fakeBasePort+10+i;
        if (sb.length() > 0) {
            sb.append(',');
        }
        sb.append("localhost:");
        sb.append(port);
    }
    serverHostPort = sb.toString();
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
@Before
public void setUp() throws Exception {

    this.count = 3;
    this.peers = new HashMap<Long,QuorumServer>(count); 
    peerTmpdir = new File[count];
    peerQuorumPort = new int[count];
    peerClientPort = new int[count];
    
    for(int i = 0; i < count; i++) {
        peerQuorumPort[i] = PortAssignment.unique();
        peerClientPort[i] = PortAssignment.unique();
        peers.put(Long.valueOf(i),
                  new QuorumServer(i, "0.0.0.0",
                                   peerQuorumPort[i],
                                   PortAssignment.unique(), null));
        peerTmpdir[i] = ClientBase.createTmpDir();
    }
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
/**
 * MockServer plays the role of peer C. Respond to two requests for votes
 * with vote for self and then Assert.fail. 
 */
void mockServer() throws InterruptedException, IOException {          
    byte b[] = new byte[36];
    ByteBuffer responseBuffer = ByteBuffer.wrap(b);
    DatagramPacket packet = new DatagramPacket(b, b.length);
    QuorumServer server = peers.get(Long.valueOf(2));
    DatagramSocket udpSocket = new DatagramSocket(server.addr.getPort());
    LOG.info("In MockServer");
    mockLatch.countDown();
    Vote current = new Vote(2, 1);
    for (int i=0;i<2;++i) {
        udpSocket.receive(packet);
        responseBuffer.rewind();
        LOG.info("Received " + responseBuffer.getInt() + " " + responseBuffer.getLong() + " " + responseBuffer.getLong());
        LOG.info("From " + packet.getSocketAddress());
        responseBuffer.clear();
        responseBuffer.getInt(); // Skip the xid
        responseBuffer.putLong(2);
        
        responseBuffer.putLong(current.getId());
        responseBuffer.putLong(current.getZxid());
        packet.setData(b);
        udpSocket.send(packet);
    }
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
    count = 3;

    peers = new HashMap<Long,QuorumServer>(count);
    threads = new ArrayList<LEThread>(count);
    tmpdir = new File[count];
    port = new int[count];

    round = new int[3];
    round[0] = 0;
    round[1] = 0;
    round[2] = 0;

    start0 = new Semaphore(0);
    finish0 = new Semaphore(0);
    finish3 = new Semaphore(0);
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
public boolean equals(Object o) {
    if (!(o instanceof QuorumMaj)) {
        return false;
    }
    QuorumMaj qm = (QuorumMaj) o;
    if (qm.getVersion() == version)
        return true;
    if (allMembers.size() != qm.getAllMembers().size())
        return false;
    for (QuorumServer qs : allMembers.values()) {
        QuorumServer qso = qm.getAllMembers().get(qs.id);
        if (qso == null || !qs.equals(qso))
            return false;
    }
    return true;
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
public QuorumMaj(Properties props) throws ConfigException {
    for (Entry<Object, Object> entry : props.entrySet()) {
        String key = entry.getKey().toString();
        String value = entry.getValue().toString();

        if (key.startsWith("server.")) {
            int dot = key.indexOf('.');
            long sid = Long.parseLong(key.substring(dot + 1));
            QuorumServer qs = new QuorumServer(sid, value);
            allMembers.put(Long.valueOf(sid), qs);
            if (qs.type == LearnerType.PARTICIPANT)
                votingMembers.put(Long.valueOf(sid), qs);
            else {
                observingMembers.put(Long.valueOf(sid), qs);
            }
        } else if (key.equals("version")) {
            version = Long.parseLong(value, 16);
        }
    }
    half = votingMembers.size() / 2;
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
public String toString() {
    StringBuilder sw = new StringBuilder();

    for (QuorumServer member : getAllMembers().values()) {
        String key = "server." + member.id;
        String value = member.toString();
        sw.append(key);
        sw.append('=');
        sw.append(value);
        sw.append('\n');
    }
    String hexVersion = Long.toHexString(version);
    sw.append("version=");
    sw.append(hexVersion);
    return sw.toString();
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
/**
 * Returns the address of the node we think is the leader.
 */
protected InetSocketAddress findLeader() {
    InetSocketAddress addr = null;
    // Find the leader by id
    Vote current = self.getCurrentVote();
    for (QuorumServer s : self.getView().values()) {
        if (s.id == current.getId()) {
            addr = s.addr;
            break;
        }
    }
    if (addr == null) {
        LOG.warn("Couldn't find the leader with id = "
                + current.getId());
    }
    return addr;
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
private void setupClientPort() throws ConfigException {
    if (serverId == UNSET_SERVERID) {
        return;
    }
    QuorumServer qs = quorumVerifier.getAllMembers().get(serverId);
    if (clientPortAddress != null && qs != null && qs.clientAddr != null) {
        if ((!clientPortAddress.getAddress().isAnyLocalAddress()
                && !clientPortAddress.equals(qs.clientAddr)) ||
                (clientPortAddress.getAddress().isAnyLocalAddress()
                        && clientPortAddress.getPort() != qs.clientAddr.getPort()))
            throw new ConfigException("client address for this server (id = " + serverId +
                    ") in static config file is " + clientPortAddress +
                    " is different from client address found in dynamic file: " + qs.clientAddr);
    }
    if (qs != null && qs.clientAddr != null) clientPortAddress = qs.clientAddr;
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
/**
 * Send notifications to all peers upon a change in our vote
 */
private void sendNotifications() {
    for (QuorumServer server : self.getVotingView().values()) {
        long sid = server.id;

        ToSend notmsg = new ToSend(ToSend.mType.notification,
                proposedLeader,
                proposedZxid,
                logicalclock,
                QuorumPeer.ServerState.LOOKING,
                sid,
                proposedEpoch);
        if(LOG.isDebugEnabled()){
            LOG.debug("Sending Notification: " + proposedLeader + " (n.leader), 0x"  +
                  Long.toHexString(proposedZxid) + " (n.zxid), 0x" + Long.toHexString(logicalclock)  +
                  " (n.round), " + sid + " (recipient), " + self.getId() +
                  " (myid), 0x" + Long.toHexString(proposedEpoch) + " (n.peerEpoch)");
        }
        sendqueue.offer(notmsg);
    }
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
/**
 * Returns the address of the node we think is the leader.
 */
protected QuorumServer findLeader() {
    QuorumServer leaderServer = null;
    // Find the leader by id
    Vote current = self.getCurrentVote();
    for (QuorumServer s : self.getView().values()) {
        if (s.id == current.getId()) {
            // Ensure we have the leader's correct IP address before
            // attempting to connect.
            s.recreateSocketAddresses();
            leaderServer = s;
            break;
        }
    }
    if (leaderServer == null) {
        LOG.warn("Couldn't find the leader with id = "
                + current.getId());
    }
    return leaderServer;
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
    this.count = 3;
    this.peers = new HashMap<Long,QuorumServer>(count);
    peerQuorumPort = new int[count];
    peerClientPort = new int[count];
    authzHosts = new HashSet<String>();
    for(int i = 0; i < count; i++) {
        peerQuorumPort[i] = PortAssignment.unique();
        peerClientPort[i] = PortAssignment.unique();
        QuorumServer qs = new QuorumServer(i, "0.0.0.0",
                peerQuorumPort[i], PortAssignment.unique(), null);
        peers.put(Long.valueOf(i), qs);
        authzHosts.add(qs.hostname);
    }
    executor = new ThreadPoolExecutor(3, 10,
            60, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
private void sendNotifications() {
    for (QuorumServer server : self.getView().values()) {

        ToSend notmsg = new ToSend(ToSend.mType.notification,
                AuthFastLeaderElection.sequencer++, proposedLeader,
                proposedZxid, logicalclock, QuorumPeer.ServerState.LOOKING,
                self.getView().get(server.id).electionAddr);

        sendqueue.offer(notmsg);
    }
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
/**
 * Send notifications to all peers upon a change in our vote
 */
private void sendNotifications() {
	for (QuorumServer server : self.getVotingView().values()) {
		long sid = server.id;

		ToSend notmsg = new ToSend(ToSend.mType.notification, proposedLeader, proposedZxid, logicalclock,
				QuorumPeer.ServerState.LOOKING, sid, proposedEpoch);
		if (LOG.isDebugEnabled()) {
			LOG.debug("Sending Notification: " + proposedLeader + " (n.leader), 0x" + Long.toHexString(proposedZxid)
					+ " (n.zxid), 0x" + Long.toHexString(logicalclock) + " (n.round), " + sid + " (recipient), "
					+ self.getId() + " (myid), 0x" + Long.toHexString(proposedEpoch) + " (n.peerEpoch)");
		}
		sendqueue.offer(notmsg);
	}
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
@Before
public void setUp()
throws Exception {
    tmpdir = ClientBase.createTmpDir();
    peers = new HashMap<Long,QuorumServer>();
    for(int i = 0; i < 5; i++) {
        peers.put(Long.valueOf(i),
                  new QuorumServer(Long.valueOf(i), "127.0.0.1", PortAssignment.unique(), 0, null));
    }
    peer = new QuorumPeer(peers,
            tmpdir,
            tmpdir,
            PortAssignment.unique(),
            3, 3, 1000, 2, 2);
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
    count = 3;

    peers = new HashMap<Long,QuorumServer>(count);
    tmpdir = new File[count];
    port = new int[count];
    cnxManagers = new QuorumCnxManager[count - 1];
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
@Test
public void testCnxManagerTimeout() throws Exception {
    Random rand = new Random();
    byte b = (byte) rand.nextInt();
    int finalOctet = b & 0xFF;
    int deadPort = PortAssignment.unique();
    String deadAddress = new String("192.0.2." + finalOctet);
        
    LOG.info("This is the dead address I'm trying: " + deadAddress);
        
    peers.put(Long.valueOf(2),
              new QuorumServer(2, deadAddress, deadPort, PortAssignment.unique(), null));
    peerTmpdir[2] = ClientBase.createTmpDir();

    QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[1], peerTmpdir[1], peerClientPort[1], 3, 1, 1000, 2, 2);
    QuorumCnxManager cnxManager = new QuorumCnxManager(peer);
    QuorumCnxManager.Listener listener = cnxManager.listener;
    if(listener != null){
        listener.start();
    } else {
        LOG.error("Null listener when initializing cnx manager");
    }

    long begin = System.currentTimeMillis();
    cnxManager.toSend(new Long(2), createMsg(ServerState.LOOKING.ordinal(), 1, -1, 1));
    long end = System.currentTimeMillis();
        
    if((end - begin) > 6000) Assert.fail("Waited more than necessary");
    
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
    count = 3;
    peers = new HashMap<Long,QuorumServer>(count);
    tmpdir = new File[count];
    port = new int[count];
}
 

import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer; //导入依赖的package包/类
void populate()
throws Exception {
    for (int i = 0; i < count; i++) {
        peers.put(Long.valueOf(i),
                  new QuorumServer(i, "0.0.0.0",
                                   PortAssignment.unique(),
                                   PortAssignment.unique(), null));
        tmpdir[i] = ClientBase.createTmpDir();
        port[i] = PortAssignment.unique();
    }
}