downloadclient.java

peeranha42是jxta的 p2p程序核心

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

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;

import org.apache.log4j.Logger;

import net.jxta.endpoint.Message;
import net.jxta.endpoint.StringMessageElement;
import net.jxta.peergroup.PeerGroup;
import net.jxta.protocol.PipeAdvertisement;
import de.uni_bremen.informatik.p2p.peeranha42.core.network.NetworkException;
import de.uni_bremen.informatik.p2p.peeranha42.core.network.connection.BiDiConnection;
import de.uni_bremen.informatik.p2p.peeranha42.core.network.connection.MessageListener;
import de.uni_bremen.informatik.p2p.plugins.filesharing.control.FileUtilities;
import de.uni_bremen.informatik.p2p.plugins.filesharing.data.DownloadJob;
import de.uni_bremen.informatik.p2p.plugins.filesharing.network.download.DownloadProtocol;

/**
 * @author Lars Kordes
 */
//public class DownloadClient extends BiDiConnection implements MessageListener {
public class DownloadClient implements MessageListener {
	public static int MAX_DOWNLOADS = 5;
	
	/** Logger for warnings, debugs and fatals */
    protected static Logger log = Logger.getLogger(DownloadClient.class);
	
	private DownloadJob job;
	
	private BiDiConnection con = null;
	
	public DownloadClient(PeerGroup grp, PipeAdvertisement pipeadv, DownloadJob job) throws NetworkException {
		this.job = job;
		
		con = new BiDiConnection(grp, pipeadv, 5000, this);
		
		if(con != null) {
			//log.debug("\n___Sending ready");
			
			Message msg = new Message();
			StringMessageElement field = new StringMessageElement(DownloadProtocol.CLIENT_READY, "1", null);
	        msg.addMessageElement(field);
	        try {
	        	//log.debug("\n___Sending client ready");
	        	con.sendMessage(msg);
	        }
	        catch (Exception e) {
	        	//log.debug("\n___Could not send ready. " + e);
	        }
		}
	}
	
	public void messageEvent(Message msg) {
		if(job.getStatus() == DownloadJob.CANCELT) {
			sendMsg(DownloadProtocol.CLIENT_END, "-1");
		}
		
		//log.debug("\n___received message: " + msg.toString());
		
		if(msg == null) {
			//log.debug("\n___Received null-message.");
			return;
		}
		if(msg.getMessageElement(DownloadProtocol.SERVER_CHK) != null ) {
			//log.debug("\n___Received server chk");
			long nr = (new Long(msg.getMessageElement(DownloadProtocol.SERVER_CHK).toString())).longValue();
			byte[] data = msg.getMessageElement(DownloadProtocol.SERVER_DATA).getBytes(false);
			if(processDownload(nr, data)) nextChk();
			else {
				//log.debug("\n___Sending client end");
				sendMsg(DownloadProtocol.CLIENT_END, "-1");
			}
		} else if(msg.getMessageElement(DownloadProtocol.SERVER_END) != null ) {
			//log.debug("\n___Received server end");
			log.info("\n___Finishing file transfer.");
			nextChk();
		} else if(msg.getMessageElement(DownloadProtocol.SERVER_READY) != null ) {
			//log.debug("\n___Received server ready");
			log.info("\n___Could connect to server... and server is ready.");
			demandFile();
		} else if(msg.getMessageElement(DownloadProtocol.SERVER_WRONG_CHK) != null ) {
			log.debug("\n___Received server wrong chk");
			nextChk();
		} else if(msg.getMessageElement(DownloadProtocol.SERVER_ERROR) != null ) {
			log.error("\n___Received error message from remote peer. Closing connection.");
		} else if(msg.getMessageElement(DownloadProtocol.SERVER_OK) != null ) {
			//log.debug("\n___Received server ok");
			nextChk();
		}
	}
	
	/**
	 * 
	 */
	private void demandFile() {
		Message msg = new Message();
		StringMessageElement field = new StringMessageElement(DownloadProtocol.CLIENT_DEMAND, job.hash, null);
        msg.addMessageElement(field);
        
        try{
        	//log.debug("\n___Sending file which gets downloaded");
        	con.sendMessage(msg);
        }
        catch(NetworkException e) {
        	log.error("\n___Cannot send demand msg to remote peer: " + e);
        }
	}

	/**
	 * 
	 */
	private void nextChk() {
		long nr = job.getNextChunkNumber();
		if(nr == DownloadJob.D_NOT_FINISHED_YET) {
			//log.debug("\n___Download not finished yet.");
		}
		else if (nr == DownloadJob.D_FINISHED) {
			//log.debug("\n___Download finished.");
		} else {
			String chknr = String.valueOf(nr);
			//log.debug("\n___asking for next chk");
			sendMsg(DownloadProtocol.CLIENT_CHK, chknr);
		}
	}

	/**
	 * 
	 */
	private synchronized boolean processDownload(long nr, byte[] data) {
		if (data != null) {
            try {
            	RandomAccessFile raf = new RandomAccessFile(FileUtilities.TEMP_DIR + File.separator + job.filename + FileUtilities.TEMPSUFFIX, "rws");
            	raf.seek(nr * DownloadProtocol.CHK_SIZE);
            	raf.write(data);
            	raf.close();
            	job.addReceivedChunkNumber(nr);
            } catch (FileNotFoundException fnfe) {
                log.error(fnfe.toString());
                return false;
            } catch (IOException io) {
                log.error(io.toString());
                return false;
            }
        }
		return true;
	}

	private void sendMsg(String type, String content) {
		Message msg = new Message();
		StringMessageElement field = new StringMessageElement(type, content, null);
        msg.addMessageElement(field);
        
        try{
        	con.sendMessage(msg);
        }
		catch(Exception e) {
			log.error("Cannot send message to remote peer: " + e);
		}
	}
}