networkhandler.java

peeranha42是jxta的 p2p程序核心

package de.uni_bremen.informatik.p2p.plugins.filesharing.network;

import java.io.IOException;
import java.io.InputStream;
import java.util.Timer;
import java.util.TimerTask;

import net.jxta.document.AdvertisementFactory;
import net.jxta.document.MimeMediaType;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.peergroup.PeerGroup;
import net.jxta.protocol.PeerGroupAdvertisement;
import net.jxta.protocol.PipeAdvertisement;

import org.apache.log4j.Logger;

import de.uni_bremen.informatik.p2p.peeranha42.core.network.Network;
import de.uni_bremen.informatik.p2p.peeranha42.core.network.receiver.Receiver;
import de.uni_bremen.informatik.p2p.peeranha42.core.plugin.Client;
import de.uni_bremen.informatik.p2p.peeranha42.core.plugin.Plugin;
import de.uni_bremen.informatik.p2p.plugins.filesharing.control.ServerManager;
import de.uni_bremen.informatik.p2p.plugins.filesharing.control.PeerGroupSearch.DocumentClassifier;
import de.uni_bremen.informatik.p2p.plugins.filesharing.data.FilesharingState;
import de.uni_bremen.informatik.p2p.plugins.filesharing.data.Protocol;

/**
 * 
 * Handles incoming and outgoing Messages.
 * Incoming messages will be unpacked and processed, 
 * outgoing messages will be packed and sent.
 * 
 * @author Daniel Gehrke, Lars Kordes, Philipp Hoheisel
 *
 */
public class NetworkHandler implements Receiver {
    
    /** Logger for warnings, debugs and fatals */
    private Logger log = Logger.getLogger(NetworkHandler.class);
   
    /** Recerence to plugin */
    private Plugin fileSharing;

    /** PeerGroupAdvertisements for Document Classes */
    private PeerGroup audio;
    private PeerGroup pic;
    private PeerGroup misc;
    private PeerGroup vid;
    
    private boolean audioJoined = false;
    private boolean picJoined = false;
    private boolean miscJoined = false;
    private boolean vidJoined = false;
    
    /** Time to leave group after it was joined for searching */
    private int LEAVETIME = 60;
    
    private NetworkHandler() {
    }
    
    public NetworkHandler(Plugin _fileSharing) {
        fileSharing = _fileSharing;
    }
    
    /**
     * This method has to be implemented, because we use the plugin interface.
     * This metod should support incoming messages.
     *
     * @param pg actual peergroup where the method was sent
     * @param pa remote PipeAdvertisement
     * @param msg incoming message
     */
    public void receive(PeerGroup pg,
                        PipeAdvertisement pa,
                        Message msg) {

       if (msg == null) {
            return;
        }
        
        // Process received message
        this.processReceivedMessage(pg,msg,pa);

    }

    
    /**
     * Processes the received message and starts action regarding
     * @param msg Received JXTAMessage
     */
    public void processReceivedMessage(PeerGroup pg, Message msg, PipeAdvertisement pa) {
        
        log.debug("\n\n\nReceiving Message in Group: " + pg.getPeerGroupName() + "\n\n\n");
        
        // Searchrequest
        if (msg.getMessageElement(Protocol.SEARCHREQUEST) != null) {
            String req = msg.getMessageElement(Protocol.SEARCHREQUEST)
                            .toString();
            
            MessageElement searchTypeElement = msg.getMessageElement(Protocol.SEARCHTYPE);
            int searchType = -1;
            if (searchTypeElement != null) {
                searchType = Integer.parseInt(searchTypeElement.toString());
            }
            
            ServerManager.searchFor(pg,req, pa, searchType);
            
            return;
        }

        // Searchresult
        if (msg.getMessageElement(Protocol.SEARCHRESULT) != null) {
            String fname = msg.getMessageElement("fname").toString();
            String uname = msg.getMessageElement("uname").toString();
            String flen = msg.getMessageElement("flen").toString();
            String nrOfChunks = msg.getMessageElement("nrofchunks").toString();
            String hash = msg.getMessageElement("hash").toString();
            
            InputStream advStream = null;
            PipeAdvertisement padv = null;
            try {
            	MessageElement element = msg.getMessageElement("PipeAdvertisment");
            	if(element == null) return;
            	advStream = element.getStream();
            	padv = (PipeAdvertisement) AdvertisementFactory.newAdvertisement(MimeMediaType.XMLUTF8, advStream);
            }
            catch(IOException e) {
            	log.error("Cannot convert PipeAdvertisement from message:\n" + e.getStackTrace());
            	return;
            }
            
            if(padv != null) ServerManager.processResponse(fname, hash, uname, flen, nrOfChunks, padv);

            return;
        }

        /*
        // Donwload Request
        if (msg.getMessageElement(Protocol.DOWNLOADREQUEST) != null) {
            String filename = msg.getMessageElement(Protocol.DOWNLOADREQUEST)
                                .toString();
            String packetNrString = msg.getMessageElement(Protocol.PACKETNUMBER)
            					.toString();
            long packetNr = Long.parseLong(packetNrString);
            ServerManager.sendData(filename, packetNr, pa);

            return;
        }
        */

        /*
        // Receive part
        if (msg.getMessageElement(Protocol.DOWNLOADANSWER) != null) {
            String downloadFile =
                msg.getMessageElement(Protocol.DOWNLOADANSWER)
                   .toString();
            long packetnr = (new Long(msg.getMessageElement(Protocol.PACKETNUMBER).toString())).longValue();
            byte data[];
 	            data = msg.getMessageElement(Protocol.PACKETDATA)
	                             .getBytes(false);
             
 	        log.debug("Incoming part of a file: " + downloadFile + " (" + packetnr + ")");
 	        
            ServerManager.processDownload(downloadFile, data, packetnr);

            return;
        }
        */

    }

    /** initializes groups and joins main group */
    public void init() {
        FilesharingState.peergroup = Network.createServiceGroup(FilesharingState.pluginid);
        Network.register(this, FilesharingState.peergroup, FilesharingState.username);

        PeerGroupAdvertisement groups[] = Client.getGroups(FilesharingState.pluginid);
        for (int i = 0; i < groups.length; i++) {
            if (groups[i].getName().equals("FilesharingMisc")) {
                misc = Network.createGroup(groups[i],FilesharingState.peergroup);
            }
            if (groups[i].getName().equals("FilesharingAudio")) {
                audio = Network.createGroup(groups[i],FilesharingState.peergroup);
            }
            if (groups[i].getName().equals("FilesharingVideo")) {
                vid = Network.createGroup(groups[i],FilesharingState.peergroup);
            }
            if (groups[i].getName().equals("FilesharingPic")) {
                pic = Network.createGroup(groups[i],FilesharingState.peergroup);
            }
        }
        
        if (misc == null || audio == null || vid == null || pic==null) {
            log.fatal("FilesharingPlugin: Initialization of Subgroups failed");
//            throw new Exception("FilesharingPlugin: Initialization of Subgroups failed");
        }

    }
    
    
    /** 
     * 
     * Joins or leaves peergroups depending on DocumentClassifier results
     * 
     * */
    public void remanageCategories() {
        
        // Joining Groups if necessary
        if (!audioJoined &&
                (FilesharingState.categories & DocumentClassifier.AUDIO) == DocumentClassifier.AUDIO) {
            log.debug("Joining Subgroup: " + audio.getPeerGroupName());
            Network.register(this,audio,FilesharingState.username);
            audioJoined = true;
        }
        if (!vidJoined &&
                (FilesharingState.categories & DocumentClassifier.VIDEO) == DocumentClassifier.VIDEO) {
            log.debug("Joining Subgroup: " + vid.getPeerGroupName());
            Network.register(this,vid,FilesharingState.username);
            vidJoined = true;
        }
        if (!picJoined &&
                (FilesharingState.categories & DocumentClassifier.PIC) == DocumentClassifier.PIC) {
            log.debug("Joining Subgroup: " + pic.getPeerGroupName());
            Network.register(this,pic,FilesharingState.username);
            picJoined = true;
        }
        if (!miscJoined &&
                (FilesharingState.categories & DocumentClassifier.MISC) == DocumentClassifier.MISC) {
            log.debug("Joining Subgroup: " + misc.getPeerGroupName());
            Network.register(this,misc,FilesharingState.username);
            miscJoined = true;
        }

        // Leaving Groups if necessary
        if (audioJoined &&
                (FilesharingState.categories & DocumentClassifier.AUDIO) != DocumentClassifier.AUDIO) {
            log.debug("Leaving Subgroup: " + audio.getPeerGroupName());
            Network.deregister(this,audio);
            audioJoined = false;
        }
        if (vidJoined &&
                (FilesharingState.categories & DocumentClassifier.VIDEO) != DocumentClassifier.VIDEO) {
            log.debug("Leaving Subgroup: " + vid.getPeerGroupName());
            Network.deregister(this,vid);
            vidJoined = false;
        }
        if (picJoined &&
                (FilesharingState.categories & DocumentClassifier.PIC) != DocumentClassifier.PIC) {
            log.debug("Leaving Subgroup: " + pic.getPeerGroupName());
            Network.deregister(this,pic);
            picJoined = false;
        }
        if (miscJoined &&
                (FilesharingState.categories & DocumentClassifier.MISC) != DocumentClassifier.MISC) {
            log.debug("Leaving Subgroup: " + misc.getPeerGroupName());
            Network.deregister(this,misc);
            miscJoined = false;
        }
        
        
    }

    /* (non-Javadoc)
     * @see de.uni_bremen.informatik.p2p.peeranha42.core.network.receiver.Receiver#getPlugin()
     */
    public Plugin getPlugin() {
        return fileSharing;
    }
    
    
    /**
     * Disconnect from network
     *
     */
    public void disconnect() {
        
        if (audioJoined) {
            Network.deregister(this,audio);
        }
        
        if (vidJoined) {
            Network.deregister(this,vid);
        }
        
        if (picJoined) {
            Network.deregister(this,pic);
        }
        
        if (miscJoined) {
            Network.deregister(this,misc);
        }
        
        // deregister plugin at client
        Network.deregister(this, FilesharingState.peergroup);
    }

    /**
     * 
     * If necessary joins the specified group, if not joined yet
     * 
     * @param searchType
     */
    private PeerGroup joinForSearch(int searchType) {
        
        // Determine searchgroup
        PeerGroup searchGroup = null;
        if (searchType == DocumentClassifier.AUDIO) {
            searchGroup = audio;
        }
        if (searchType == DocumentClassifier.VIDEO) {
            searchGroup = vid;
        }
        if (searchType == DocumentClassifier.PIC) {
            searchGroup = pic;
        }
        if (searchType == DocumentClassifier.MISC) {
            searchGroup = misc;
        }
        

        if (searchGroup != null) {
        	// Join group if necessary
	        if (searchType == DocumentClassifier.AUDIO && !audioJoined) {
	            log.debug("Joining Subgroup for Search: " + searchGroup.getPeerGroupName());
	            Network.register(this,searchGroup,FilesharingState.username);
	        }
	        if (searchType == DocumentClassifier.VIDEO && !vidJoined) {
	            log.debug("Joining Subgroup for Search: " + searchGroup.getPeerGroupName());
	            Network.register(this,searchGroup,FilesharingState.username);
	        }
	        if (searchType == DocumentClassifier.PIC && !picJoined) {
	            log.debug("Joining Subgroup for Search: " + searchGroup.getPeerGroupName());
	            Network.register(this,searchGroup,FilesharingState.username);
	        }
	        if (searchType == DocumentClassifier.MISC && !miscJoined) {
	            log.debug("Joining Subgroup for Search: " + searchGroup.getPeerGroupName());
	            Network.register(this,searchGroup,FilesharingState.username);
	        }
	    }
        
        // If necessary, leave group after LEAVETIME seconds
        PeerGroupLeaverTask peerGroupLeaver = new PeerGroupLeaverTask(this,searchGroup,searchType);
        new Timer().schedule(peerGroupLeaver,LEAVETIME*1000);
        
        return searchGroup;
    }

    /**
     * @param searchType
     * @param msg
     */
    public void sendSearchMsg(int searchType, Message msg) {
       
        // Let networkHandler join searchgroup if not joined yet
        PeerGroup searchGroup = this.joinForSearch(searchType);
        
        // Send search request
        log.debug("\n\n\n\n\n\n\nSending SearchRequest in Peergroup: " + searchGroup.getPeerGroupName());
        try {
            Network.sendMsg(FilesharingState.pluginid,searchGroup,msg);
        } catch (IOException e) {
            log.fatal(e);
        }

    }
}

class PeerGroupLeaverTask extends TimerTask {

    /** Logger for warnings, debugs and fatals */
    private Logger log = Logger.getLogger(PeerGroupLeaverTask.class);

    private Receiver receiver;
    private int searchType;
    private PeerGroup searchGroup;
    
    public PeerGroupLeaverTask(Receiver _receiver, PeerGroup _searchGroup, int _searchType) {
        receiver = _receiver;
        searchType = _searchType;
        searchGroup = _searchGroup;
    }
    
    /* (non-Javadoc)
     * @see java.util.TimerTask#run()
     */
    public void run() {
        if ((FilesharingState.categories & searchType) != searchType) {
            log.debug("Leaving Group only joined for searching: " + searchGroup.getPeerGroupName());
            Network.deregister(receiver,searchGroup);
        } else {
            log.debug("Group was not only joined for searching. Staying in: "  + searchGroup.getPeerGroupName());
        }
        
    }
    
}