downloadserver.java

peeranha42是jxta的 p2p程序核心

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

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;

import net.jxta.document.MimeMediaType;
import net.jxta.endpoint.ByteArrayMessageElement;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.StringMessageElement;

import org.apache.log4j.Logger;

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.data.Share;
import de.uni_bremen.informatik.p2p.plugins.filesharing.network.download.DownloadProtocol;

/**
 * @author Lars Kordes
 */
public class DownloadServer implements MessageListener {

	/** Logger for warnings, debugs and fatals */
    protected static Logger log = Logger.getLogger(DownloadServer.class);
	
    private Share share = null;
    private BiDiConnection con;
    private ArrayList shares;
    
	/**
	 * @param connection
	 * @param shares
	 */
	public DownloadServer(BiDiConnection connection, ArrayList list) throws NetworkException {
		if(connection == null || list == null) {
			//log.debug("\n___something is null. con == null: " + (connection == null) + ", list == null: " + (list == null));
			throw new NetworkException("Cannot init server thread for download.");
		}
		con = connection;
		shares = list;
		
		con.setListener(this);
	}

	/**
	 * @param incomingMsg
	 */
	public void messageEvent(Message msg) {
		//log.debug("\n___Received message: " + msg.toString());
		if(msg == null) {
			//log.debug("\n___Received null-message.");
			return;
		}
		if(msg.getMessageElement(DownloadProtocol.CLIENT_READY) != null) {
			//log.debug("\n___Received client ready");
			//log.debug("\n___Sending server ready");
			sendMsg(DownloadProtocol.SERVER_READY, "1");
		} else if(msg.getMessageElement(DownloadProtocol.CLIENT_CHK) != null ) {
			long chk = Long.parseLong(msg.getMessageElement(DownloadProtocol.CLIENT_CHK).toString());
			//log.debug("\n___Received client chk. chk nr is " + chk);
			if(share == null) {
				//log.debug("\n___Sending server error. Client maybe did not ask for file before chk-demand.");
				sendMsg(DownloadProtocol.SERVER_ERROR, "-1");
			} else if(chk < 0) {
				//log.debug("\n___Sending wrong chk");
				sendMsg(DownloadProtocol.SERVER_WRONG_CHK, String.valueOf(chk));
			} else {
				try {
					byte[] dataArr;
					BufferedInputStream bis;
					try {
						bis = new BufferedInputStream(new FileInputStream(share.file));
					}
					catch(FileNotFoundException e) {
						//log.debug("\n___Sending server error. Could not find file, the client asked for.");
						sendMsg(DownloadProtocol.SERVER_ERROR, "-1");
						return;
					}
					bis.skip(chk * DownloadProtocol.CHK_SIZE);
					if(bis.available() < 0) {
						//log.debug("\n___Sending server end. Transfer finished");
						sendMsg(DownloadProtocol.SERVER_END, "1");
					} else {
						if(bis.available() < DownloadProtocol.CHK_SIZE) {
							dataArr = new byte[bis.available()];
							bis.read(dataArr, 0, bis.available());
						} else {
							dataArr = new byte[DownloadProtocol.CHK_SIZE];
							bis.read(dataArr, 0, DownloadProtocol.CHK_SIZE);
						}
						sendMsg(dataArr, chk);
					}
				}
				catch(IOException e) {
					log.error("\n___Cannot read from file: " + e);
					//log.debug("\n___Sending server error");
					sendMsg(DownloadProtocol.SERVER_ERROR, "-1");
				}
			}
			
		} else if(msg.getMessageElement(DownloadProtocol.CLIENT_DEMAND) != null) {
			log.debug("\n___Received client demand");
			// search for file and give response
			String hash = msg.getMessageElement(DownloadProtocol.CLIENT_DEMAND).toString();
			for(int i = 0; i < shares.size(); i++) {
				share = (Share) shares.get(i);
				if(share.hash.equals(hash)) {
					//log.debug("\n___Sending server ok. File found.");
					sendMsg(DownloadProtocol.SERVER_OK, "-1");
					return;
				}
			}
			// share not found
			if(share == null) {
				//log.debug("\n___Sending server error. Could not find file the client asked for.");
				sendMsg(DownloadProtocol.SERVER_ERROR, "-1");
			}
			
		} else if(msg.getMessageElement(DownloadProtocol.CLIENT_END) != null) {
			//log.debug("\n___Received client end");
		}
	}
	
	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("\n___Cannot send message to remote peer: " + e);
		}
	}
	
	private void sendMsg(byte[] bytes, long chk) {
		Message msg = new Message();
		ByteArrayMessageElement bame = new ByteArrayMessageElement(DownloadProtocol.SERVER_DATA, MimeMediaType.AOS, bytes, null);
        msg.addMessageElement(bame);
        StringMessageElement field = new StringMessageElement(DownloadProtocol.SERVER_CHK, String.valueOf(chk), null);
        msg.addMessageElement(field);
        
		try { con.sendMessage(msg); }
		catch(Exception e) {
			log.error("\n___Cannot send message to remote peer: " + e);
		}
	}
}