p2psocket.java

jxta p2p 网络编程技术原代码 清华大学出版社出版

/**《JXTA网络编程》例程
*"第五章  JXTA深入编程"  
*
@作者 慈黎利
@单位 清华大学计算机系软件所知识工程组
@版本 1.0
@联系方式 [e-mail]
@描述 P2PSockst 负责在Peer 之间建立通讯连接
Copyright 2002
*/

package demo.p2psocket;
import java.net.*;
import java.io.*;

import net.jxta.pipe.*;
import net.jxta.protocol.*;
import net.jxta.peergroup.*;
import net.jxta.document.*;
import net.jxta.discovery.*;
import net.jxta.exception.*;
import net.jxta.endpoint.*;
import net.jxta.impl.endpoint.*;

// 源程序 P2PSocket.java

public class P2PSocket extends Thread implements OutputListener,PipeMsgListener{
	private String inputPipeName=null;   			//输入管道的名称
	private String outputPipeName=null;  			//输出管道的名称
	private PipeAdvertisement inputPipeAdv=null;    //输入管道广告
	private PipeAdvertisement outputPipeAdv=null;   //输出管道广告
	private InputPipe ip=null;                      //输入管道
	private OutputPipe op=null;                     //输出管道 
	private PipeMsgListener pml=null;				//输入管道的监听器
	private OutputListener opl=null;                //输出管道的监听器
	private PeerGroup pg=null;						//所属的组
	private PipeService ps=null;					//管道服务
	private DiscoveryService disc=null;  			//发现服务
	/**
	*  构造方法，创建并加入默认的组 同时获得相应的组服务 
	*/
	public P2PSocket() {
			if (pg==null) this.newPeerGroup();
			ps = pg.getPipeService();
			disc=pg.getDiscoveryService();
	}
	
	/**
	*  构造方法，创建并加入默认的组 同时获得相应的组服务 
	* @param inputPipeName 输入管道的名称
	* @param outputPipeName 输出管道的名称
	*/	
	public P2PSocket(String inputPipeName, String outputPipeName) {
		if (pg==null) this.newPeerGroup();
		ps = pg.getPipeService();
		disc=pg.getDiscoveryService();
		this.setInputPipeName(inputPipeName);
		this.setOutputPipeName(outputPipeName);
	}
	
	/**
	*  Peer与输入管道进行绑定并监听
	* @return 如果绑定成功返回true 
	*/	 
	public boolean bind() {
		for (int i=0;i<5;i++) {
			try {
				this.start();
				if (pml!=null) ip = ps.createInputPipe(inputPipeAdv, pml);
				else ip=ps.createInputPipe(inputPipeAdv,this);
			}catch (Exception e) {
				System.out.println("Error creating input pipe.");			
			}
			if (ip != null) {
				return true;
			}
		}
		return false;
	}
	
	/**
	*  每隔10分钟发布一次绑定的广告
	*/
	public void run() {
		try {
			disc.remotePublish(inputPipeAdv,DiscoveryService.ADV,10*60*1000);
			disc.publish(inputPipeAdv,DiscoveryService.ADV,10*60*1000,10*60*1000);	
			this.sleep(10*60*1000);
		}catch (InterruptedException ie) {
			System.out.println("出错 "+ie);
			System.exit(-1);
		}catch(IOException ioe) {
			System.out.println("出错! "+ioe);
			System.exit(-1);
		}
	}
	
	/**
	*  Peer与输出管道建立通讯联系
	* @return 如果绑定成功返回true 
	*/	 
	public boolean connect() { 
		for (int i=0;i<10;i++) {
			try {
				System.out.println("Create Outpipe");					
				if (opl!=null)	ps.createOutputPipe(outputPipeAdv, opl);
				else ps.createOutputPipe(outputPipeAdv,this);
			}catch (IOException e) {
				System.out.println("Error: OutputPipe creation failure");
				e.printStackTrace();
			}
			if (opl!=null && opl.getOutputPipe()!=null) break;
			try {
				Thread.sleep(5*1000);
			}catch (InterruptedException ie) {
				System.out.println("出错 "+ie);
				System.exit(-1);
			}
		}
		if (opl!=null && opl.getOutputPipe()!=null) {
			this.op=opl.getOutputPipe();
			return true;
		}else {
			System.out.println("通讯连接失败!");
			System.exit(-1);
		}
		return false;
	}
	
	/**
	*  Peer与输出管道建立通讯联系
	* @ param outputPipeName 输出管道的名字
	* @ return 如果绑定成功返回true 
	*/	 
	public boolean connect(String outputPipeName) {
		this.setOutputPipeName(outputPipeName);
		if (this.connect()) return true;
		else return false;
	}
	
	/**
	* 实现输出管道监听器接口
	* @ param event 输出管道绑定发生的事件
	*/
	public void outputPipeEvent(OutputPipeEvent event) {
			op = event.getOutputPipe();	
	}
	
	/**
	* 利用输出管道发送消息
	* @ param mess 需要发送的消息
	* @ return 如果发送成功返回true 
	*/	 
	public boolean send(Message mess) {
	if (opl!=null) op=opl.getOutputPipe();
	while (op==null) 
	try {
		Thread.sleep(5*1000);
	}catch (InterruptedException ie) {
		System.out.println("出错 "+ie);
		System.exit(-1);
	} 
		try {
			op.send(mess);
			return true;		
		}catch (IOException ioe) {
			System.out.println("发送消息失败!");
			return false;
		}
	}
	
	/** 
	* 实现输入管道监听器接口
	* @ param event 输入管道绑定发生的事件
	*/
	public void pipeMsgEvent ( PipeMsgEvent event ){
		
	}
	/** 
	* 设定输出管道的监听器
	* @ param opl 输出管道的监听器
	*/
	public void setOutListener(OutputListener opl) {
		this.opl=opl;
	}
	
	/** 
	* 设定输入管道的监听器
	* @ param pml 输入管道的监听器
	*/
	public void setInListener(PipeMsgListener pml) {
		this.pml=pml;
	}
	
	/** 
	* 设定Peer 隶属的组 并取得相应的组服务
	* @ param pg 特定的组
	*/
	public void setPeerGroup(PeerGroup pg) {
		this.pg=pg;
		this.ps=pg.getPipeService();
		this.disc=pg.getDiscoveryService();
	}
	/**
	* 设定监听管道的名字，生成输入管道广告
	* @param inputPipeName 输入管道的名字
	*/
	public void setInputPipeName(String inputPipeName) {
		this.inputPipeName=inputPipeName;
		inputPipeAdv=createPipeAdvFromFile("adv/"+inputPipeName+".xml");
	}
	
	/**
	* 设定输出管道的名字,生成输出管道广告
	* @param inputPipeName 输出管道的名字
	*/
	public void setOutputPipeName(String outputPipeName) {
		this.outputPipeName=outputPipeName;
		outputPipeAdv=createPipeAdvFromFile("adv/"+outputPipeName+".xml");
	}
	/**
	* 获得输入管道
	* @return 输入管道
	*/
	public InputPipe getInputPipe() {
		return this.ip;
	}
	/**
	* 获得输入管道的名字
	* @return 输入管道的名字
	*/
	public String getInputPipeName() {
		return this.inputPipeName;
	}
	/**
	* 获得输出管道
	* @return 输出管道
	*/
    public OutputPipe getOutputPipe() {
    	return this.op;
    }
	/**
	* 获得输出管道的名字
	* @return 输出管道的名字
	*/
	public String getOutputPipeName() {
		return this.outputPipeName;
	}
	
	/**
	* 创建并加入默认的组NetPeerGroup,同时获得相应的组服务
	*/
	private void newPeerGroup(){
		try {
			pg = PeerGroupFactory.newNetPeerGroup();            
		} 
		catch (PeerGroupException e) {
			System.out.println("fatal error : group creation failure");
			e.printStackTrace();
			System.exit(-1);
		}	
	}
	
	/**
	* 从XML文件中成生管道广告
	* 为简便起见，例子不提供Pipe Advertisement 的搜索方法
	* 所有用到的广告都是从文件生成，文件存放在ADV 目录下
	* @ param filename  生成广告的文件名
	* @ return 返回管道广告
	*/
	private PipeAdvertisement createPipeAdvFromFile(String filename) {
    	PipeAdvertisement pipeAdv=null;
    	try {
    		FileInputStream is=new FileInputStream(filename);
    		pipeAdv=(PipeAdvertisement) 
    				AdvertisementFactory.newAdvertisement(
    					new MimeMediaType("text/xml"),is);
    		is.close();
    	}catch(Exception e) {
    		System.err.println("Error to create pipeAdvertisement from file");
    		e.printStackTrace();
    		System.exit(1);
    	}
    	return pipeAdv;    		
    }
    
    
}