peerdiscoverythread.java

Java p2p程序设计2002年版

/*
 * PeerDiscoveryThread.java
 * @Author Daniel Brookshier [e-mail]
 * Created on December 11, 2001, 6:54 PM
 */

package com.sams.jxta.explorer;


import java.util.Enumeration;
import java.util.Hashtable;
import org.apache.log4j.Category;

import net.jxta.document.Advertisement;
import net.jxta.document.MimeMediaType;
import net.jxta.document.AdvertisementFactory;
import net.jxta.peergroup.PeerGroup;
import net.jxta.exception.PeerGroupException;
import net.jxta.protocol.DiscoveryResponseMsg;
import net.jxta.protocol.PeerGroupAdvertisement;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.protocol.ModuleImplAdvertisement;
import net.jxta.discovery.DiscoveryEvent;
import net.jxta.discovery.DiscoveryService;
import net.jxta.discovery.DiscoveryListener;

import javax.swing.JTree.DynamicUtilTreeNode;
import javax.swing.JTree;
import javax.swing.tree.DefaultTreeModel;
import javax.swing.tree.DefaultMutableTreeNode;

import java.io.ByteArrayInputStream;

public class PeerDiscoveryThread implements Runnable, DiscoveryListener {
    private static final Category LOG =  Category.getInstance(PeerDiscoveryThread.class.getName());
    
    protected static final long   REMOTE_DISCOVERY_INTERVAL_QUICK = 1*60 * 1000;
    protected static final long   REMOTE_DISCOVERY_INTERVAL = 15*60 * 1000;
    protected static final long  PARANOID_COUNT = 10;
    protected final static MimeMediaType XMLMIMETYPE = new MimeMediaType("text/xml");
   
    
    protected boolean done=false;
    protected Thread thread;
    protected DiscoveryService disco;
    protected DynamicUtilTreeNode peers;
    protected DynamicUtilTreeNode groups;
    protected DynamicUtilTreeNode advertisements;
    
    protected Hashtable peerHash  ;
    protected Hashtable groupHash  ;
    protected Hashtable advertisementHash ;
    protected JTree tree;
    protected DynamicUtilTreeNode parent;
    protected PeerGroup parentGroup;
    protected PeerGroup peerGroup;
    protected Advertisement advertisement;
    public PeerDiscoveryThread(JTree tree,PeerGroup parentGroup,  DynamicUtilTreeNode parent){
        
        this.parentGroup = parentGroup;
        this.tree = tree;
        this.parent = parent;
        
        peerHash = new Hashtable();
        groupHash = new Hashtable();
        advertisementHash = new Hashtable();
        
        
        groups = new DynamicUtilTreeNode("groups",groupHash);
        peers = new DynamicUtilTreeNode("peers",peerHash);
        advertisements = new DynamicUtilTreeNode("adverts",advertisementHash);
        
        parent.add(groups);
        parent.add(peers);
        parent.add(advertisements);
        ((DefaultTreeModel)tree.getModel()).reload(parent) ;
        
        if (parent.getUserObject() instanceof Advertisement){
            this.advertisement = (Advertisement)parent.getUserObject();
        }else{
            peerGroup = parentGroup;
        }
        
    }
    public void run() {
        if (advertisement != null){
            try{
                peerGroup = parentGroup.newGroup(advertisement);
            }catch(PeerGroupException pge){
                parent.add(new JTree.DynamicUtilTreeNode("Cannot open group: "+pge.getMessage(),new Hashtable()));
                LOG.error("Unable to initialize peer group.",pge);
                return;
            }
        }
        
        
        int i =0;
        int counter = 0;
        long time = 0;
        disco = peerGroup.getDiscoveryService();
        while( !done) {
            disco.addDiscoveryListener(this);
            disco.getRemoteAdvertisements(null, DiscoveryService.GROUP, null,null, 100);
            disco.getRemoteAdvertisements(null, DiscoveryService.PEER, null,null,  100);
            disco.getRemoteAdvertisements(null, DiscoveryService.ADV, null,null, 100);
            try {
                
                if ( ++counter < PARANOID_COUNT){
                    time = REMOTE_DISCOVERY_INTERVAL_QUICK;
                }else{
                    time = REMOTE_DISCOVERY_INTERVAL;
                }
                Thread.currentThread().sleep(time);
            } catch (InterruptedException e) {
                LOG.debug("Discovery Sleeping thred Interrupted");
                done = true;
            }
            
        }
    }
    
    public void discoveryEvent(DiscoveryEvent event) {
        DiscoveryResponseMsg msg  = event.getResponse();
        Enumeration enum = msg.getResponses();
        String str = null;
        boolean changedGroup = false;
        boolean changedPeer = false;
        boolean changedAdvertisement = false;
        while (enum.hasMoreElements()) {
            try {
                str = (String) enum.nextElement();
                if (str == null) continue;
                // Create Peer(Group)Advertisement from response.
                ByteArrayInputStream stream = new ByteArrayInputStream(str.getBytes());
                Advertisement adv = (Advertisement) AdvertisementFactory.newAdvertisement(XMLMIMETYPE, stream);
                String type = "";
                if (adv instanceof PeerGroupAdvertisement ){
                    if (groupHash.get(adv.getID()) == null){
                        groupHash.put(adv.getID(),adv);
                        JTree.DynamicUtilTreeNode node = new JTree.DynamicUtilTreeNode(adv,new Hashtable());
                        groups.add(node);
                        System.out.println(">>>>> added group");
                        changedGroup = true;
                    }
                }else if (adv instanceof PeerAdvertisement ){
                    if (peerHash.get(adv.getID()) == null){
                        peerHash.put(adv.getID(),adv);
                        DefaultMutableTreeNode node = new DefaultMutableTreeNode(adv);
                        peers.add(node);
                        changedPeer = true;
                    }
                } else {
                    if (adv.getID() != null){
                        if ( advertisementHash.get(adv.getID()) == null){
                            advertisementHash.put(adv.getID(),adv);
                            DefaultMutableTreeNode node = new DefaultMutableTreeNode(adv);
                            advertisements.add(node);
                            changedAdvertisement = true;
                        }
                    }else{
                        if (adv instanceof ModuleImplAdvertisement){
                            if ( advertisementHash.get(((ModuleImplAdvertisement)adv).getModuleSpecID()) == null){
                                advertisementHash.put(((ModuleImplAdvertisement)adv).getModuleSpecID(),adv);
                                DefaultMutableTreeNode node = new DefaultMutableTreeNode(adv);
                                advertisements.add(node);
                                changedAdvertisement = true;
                            }
                        }else  if ( advertisementHash.get(new Long(adv.getLocalExpirationTime())) == null){
                            advertisementHash.put(new Long(adv.getLocalExpirationTime()),adv);
                            DefaultMutableTreeNode node = new DefaultMutableTreeNode(adv);
                            advertisements.add(node);
                            changedAdvertisement = true;
                        }
                    }
                }
            } catch (Exception ex) {
                LOG.error("Error processing a new advertisement:\""+str+"\"",ex);
            }
            if (changedGroup){
                ((DefaultTreeModel)tree.getModel()).reload(groups) ;
            }
            if (changedPeer){
                ((DefaultTreeModel)tree.getModel()).reload(peers) ;
            }
            if (changedAdvertisement){
                ((DefaultTreeModel)tree.getModel()).reload(advertisements) ;
            }
            
        }// end while
    }// end of discoveryEvent()
    
}//End of class PeerDiscoveryThread