import org.jgroups.protocols.PingData; //导入依赖的package包/类
protected void update(PingData data) {
	if (!ClusterDiscoveryDAO.isConfigured()) return;
	org.hibernate.Session hibSession = ClusterDiscoveryDAO.getInstance().createNewSession();
	String own_address = addressAsString(data.getAddress());
       Transaction tx = null;
       try {
       	tx = hibSession.beginTransaction();
		ClusterDiscovery cluster = ClusterDiscoveryDAO.getInstance().get(new ClusterDiscovery(own_address, cluster_name), hibSession);
		if (cluster == null)
			cluster = new ClusterDiscovery(own_address, cluster_name);
		cluster.setPingData(serializeWithoutView(data));
		cluster.setTimeStamp(new Date());
		hibSession.saveOrUpdate(cluster);
		hibSession.flush();
           if (tx != null) tx.commit();
	} catch (Exception e) {
		if (tx!=null) tx.rollback();
		log.info("Failed to update my data for cluster " + cluster_name + ": " + e.getMessage());
	} finally {
		hibSession.close();
	}
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public Object up(Event evt) {

  switch (evt.getType()) {

    case Event.FIND_MBRS:
      List<Address> missing = (List<Address>) evt.getArg();

      Responses responses = new Responses(false);
      for (Address laddr : missing) {
        try {
          if (laddr instanceof JGAddress) {
            PingData pd = new PingData(laddr, true, laddr.toString(), newIpAddress(laddr));
            responses.addResponse(pd, false);
            updateUDPCache(pd);
          }
        } catch (RuntimeException e) {
          logger.warn("Unable to create PingData response", e);
          throw e;
        }
      }
      return responses;
  }
  return up_prot.up(evt);
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
/**
 * update the logical->physical address cache in UDP, which doesn't seem to be updated by UDP when
 * processing responses from FIND_MBRS
 * 
 * @param pd
 */
private void updateUDPCache(PingData pd) {
  if (setPingData == null && !warningLogged) {
    findPingDataMethod();
  }
  if (setPingData != null) {
    try {
      setPingData.invoke(transport, new Object[] {pd});
    } catch (InvocationTargetException | IllegalAccessException e) {
      if (!warningLogged) {
        log.warn("Unable to update JGroups address cache - this may affect performance", e);
        warningLogged = true;
      }
    }
  }
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
/** Handles the case where no coord responses were received. Returns true if we became the coord
 * (caller needs to terminate the join() call), or false when the caller needs to continue */
protected boolean firstOfAllClients(final Address joiner, final Responses rsps) {
    log.trace("%s: could not determine coordinator from rsps %s", gms.local_addr, rsps);

    // so the member to become singleton member (and thus coord) is the first of all clients
    SortedSet<Address> clients=new TreeSet<>();
    clients.add(joiner); // add myself again (was removed by findInitialMembers())
    for(PingData response: rsps)
        clients.add(response.getAddress());

    log.trace("%s: nodes to choose new coord from are: %s", gms.local_addr, clients);
    Address new_coord=clients.first();
    if(new_coord.equals(joiner)) {
        log.trace("%s: I (%s) am the first of the nodes, will become coordinator", gms.local_addr, joiner);
        becomeSingletonMember(joiner);
        return true;
    }
    log.trace("%s: I (%s) am not the first of the nodes, waiting for another client to become coordinator",
              gms.local_addr, joiner);
    Util.sleep(500);
    return false;
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
/** Returns all members whose PingData is flagged as coordinator */
private static List<Address> getCoords(Iterable<PingData> mbrs) {
    if(mbrs == null)
        return null;

    List<Address> coords=null;
    for(PingData mbr: mbrs) {
        if(mbr.isCoord()) {
            if(coords == null)
                coords=new ArrayList<>();
            if(!coords.contains(mbr.getAddress()))
                coords.add(mbr.getAddress());
        }
    }
    return coords;
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
protected int[] numResponses() {
    lock.lock();
    try {
        int[] num={0,0};
        for(int i=0; i < index; i++) {
            PingData data=ping_rsps[i];
            num[0]++;
            if(data.isCoord())
                num[1]++;
        }
        return num;
    }
    finally {
        lock.unlock();
    }
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
@Test
public void testResponse() throws Exception {
    Address local_addr = pinger.getLocalAddress();
    PhysicalAddress physical_addr = (PhysicalAddress) pinger
            .down(new Event(Event.GET_PHYSICAL_ADDRESS, local_addr));
    PingData data = createPingData(local_addr, physical_addr);
    final PingHeader hdr = getPingHeader(data);
    Message msg = new Message(null).setFlag(Message.Flag.DONT_BUNDLE)
            .putHeader(pinger.getId(), hdr).setBuffer(streamableToBuffer(data));
    URL url = new URL("http://localhost:8888");
    HttpURLConnection conn = (HttpURLConnection) url.openConnection(Proxy.NO_PROXY);
    conn.addRequestProperty(Server.CLUSTER_NAME, TestBase.CLUSTER_NAME);
    conn.setDoOutput(true);
    conn.setRequestMethod("POST");

    DataOutputStream out = new DataOutputStream(conn.getOutputStream());
    msg.writeTo(out);
    out.flush();

    Assert.assertEquals(200, conn.getResponseCode());
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
protected void parsePingData(final byte[] pingBytes, final List<Address> members, final Responses responses) {
    if (pingBytes == null || pingBytes.length <= 0) {
        return;
    }
    List<PingData> list;
    try {
        list = read(new ByteArrayInputStream(pingBytes));
        if (list != null) {
            // This is a common piece of logic for all PING protocols copied from org/jgroups/protocols/FILE_PING.java:245
            // Maybe could be extracted for all PING impls to share this logic?
            for (PingData data : list) {
                if (members == null || members.contains(data.getAddress())) {
                    responses.addResponse(data, data.isCoord());
                }
                if (local_addr != null && !local_addr.equals(data.getAddress())) {
                    addDiscoveryResponseToCaches(data.getAddress(), data.getLogicalName(), data.getPhysicalAddr());
                }
            }
            // end copied block
        }
    } catch (Exception e) {
        log.error("Error unmarshalling ping data.", e);
    }
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
@Override
protected void write(final List<PingData> list, final String clustername) {
    if (list == null || clustername == null) {
        return;
    }

    String filename = addressToFilename(clustername, local_addr);
    ByteArrayOutputStream out = new ByteArrayOutputStream(STREAM_BUFFER_SIZE);

    try {
        write(list, out);
        byte[] data = out.toByteArray();

        // Upload the file
        CloudBlockBlob blob = containerReference.getBlockBlobReference(filename);
        blob.upload(new ByteArrayInputStream(data), data.length);

    } catch (Exception ex) {
        log.error("Error marshalling and uploading ping data.", ex);
    }

}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
/**
 * find and initialize the method used to update UDP's address cache
 */
private void findPingDataMethod() {
  transport = (TP) getProtocolStack().getTransport();
  try {
    setPingData = TP.class.getDeclaredMethod("setPingData", new Class<?>[] {PingData.class});
    setPingData.setAccessible(true);
  } catch (NoSuchMethodException e) {
    if (!warningLogged) {
      log.warn("Unable to update JGroups address cache - this may affect performance", e);
      warningLogged = true;
    }
  }
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
public void getMembers(final String group, Responses rsps) throws Exception {
    lock.lock();
    try {
        if(!isConnected() || input == null) throw new Exception ("not connected");
        // we might get a spurious SUSPECT message from the router, just ignore it
        if(input.available() > 0) // fixes https://jira.jboss.org/jira/browse/JGRP-1151
            input.skipBytes(input.available());

        GossipData request=new GossipData(GossipRouter.GOSSIP_GET, group, null);
        request.writeTo(output);
        output.flush();

        short num_rsps=input.readShort();
        for(int i=0; i < num_rsps; i++) {
            PingData rsp=new PingData();
            rsp.readFrom(input);
            rsps.addResponse(rsp, false);
        }
    }
    catch(Exception e) {           
        connectionStateChanged(ConnectionStatus.CONNECTION_BROKEN);
        throw new Exception("Connection to " + getGossipRouterAddress() + " broken. Could not send GOSSIP_GET request", e);
    }
    finally {
        if(lock.isHeldByCurrentThread())
            lock.unlock();
    }
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
public void addResponse(PingData rsp, boolean overwrite) {
    if(rsp == null)
        return;

    boolean is_coord_rsp=rsp.isCoord(), changed=false;
    lock.lock();
    try {
        // https://jira.jboss.org/jira/browse/JGRP-1179
        int ind=find(rsp);
        if(ind == -1) { // new addition
            add(rsp);
            changed=true;
        }
        else {
            PingData existing=ping_rsps[ind]; // cannot be null
            if(overwrite || (is_coord_rsp && !existing.isCoord())) {
                ping_rsps[ind]=rsp;
                changed=true;
            }
        }
        if(changed && ((num_expected_rsps > 0 && index >= num_expected_rsps) || break_on_coord_rsp && is_coord_rsp))
            _done();
    }
    finally {
        lock.unlock();
    }
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
public PingData findResponseFrom(Address mbr) {
    if(mbr == null) return null;
    for(int i=0; i < index; i++) {
        if(ping_rsps[i] != null && mbr.equals(ping_rsps[i].getAddress()))
            return ping_rsps[i];
    }
    return null;
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
@Deprecated
public List<PingData> get(long timeout) throws InterruptedException {
    lock.lock();
    try {
        waitFor(timeout);
        return toList();
    }
    finally {
        lock.unlock();
    }
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
public Iterator<PingData> iterator() {
    lock.lock();
    try {
        return new PingDataIterator(ping_rsps, index);
    }
    finally {
        lock.unlock();
    }
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
@GuardedBy("lock") protected int find(PingData data) {
    if(data == null) return -1;
    for(int i=0; i < index; i++) {
        if(data.equals(ping_rsps[i]))
            return i;
    }
    return -1;
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
public PingDataIterator(PingData[] data, int end_index) {
    this.data=data;
    this.end_index=end_index;
    if(data == null)
        throw new IllegalArgumentException("data cannot be null");
    if(end_index > data.length)
        throw new IndexOutOfBoundsException("index is " + end_index + ", but arrays's length is only " + data.length);
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
    String host="localhost";
    int port=12001;
    String cluster_name="draw";

    for(int i=0; i < args.length; i++) {
        if(args[i].equals("-host")) {
            host=args[++i];
            continue;
        }
        if(args[i].equals("-port")) {
            port=Integer.parseInt(args[++i]);
            continue;
        }
        if(args[i].equals("-cluster")) {
            cluster_name=args[++i];
            continue;
        }
        help();
        return;
    }

    RouterStub stub=new RouterStub(host, port, null,null);
    stub.doConnect();
    Responses rsps=new Responses(false);
    stub.getMembers(cluster_name, rsps);
    for(PingData data: rsps)
        System.out.println(data);
    stub.destroy();
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
protected void writeOwnInformation() {
    PhysicalAddress physical_addr=(PhysicalAddress)down(new Event(Event.GET_PHYSICAL_ADDRESS, local_addr));
    update(new PingData(local_addr, is_server, UUID.get(local_addr), physical_addr).coord(is_coord));
}
 

import org.jgroups.protocols.PingData; //导入依赖的package包/类
protected static String print(List<PingData> rsps) {
    StringBuilder sb=new StringBuilder();
    for(PingData rsp: rsps)
        sb.append(rsp.getAddress() + " ");
    return sb.toString();
}