multicastoutpipe.java

JXTA的 即时通信系统 JXTACHAT

import net.jxta.discovery.DiscoveryService;
import net.jxta.document.AdvertisementFactory;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.StringMessageElement;
import net.jxta.id.IDFactory;
import net.jxta.pipe.OutputPipe;
import net.jxta.pipe.OutputPipeEvent;
import net.jxta.pipe.OutputPipeListener;
import net.jxta.pipe.PipeID;
import net.jxta.pipe.PipeService;
import net.jxta.platform.NetworkManager;
import net.jxta.protocol.PipeAdvertisement;
import java.io.File;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Date;
import net.jxta.peergroup.*;
import net.jxta.rendezvous.RendezVousService;
import net.jxta.rendezvous.RendezvousEvent;
import net.jxta.rendezvous.RendezvousListener;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

public class MulticastOutPipe implements Runnable,OutputPipeListener, RendezvousListener {

	public final static String MESSAGE_NAME_SPACE = "chatmsg";
	private String peerName;
	private boolean waitForRendezvous = false;
	private PipeService pipeService;
	private PipeAdvertisement pipeAdv;
	private OutputPipe outputPipe;
	//private final Object lock = new String("lock");
	// 指向集合点服务对象
	private RendezVousService rendezvous;
	// 新建的点组
	private PeerGroup myGroup = null;
	// 指定的管道的ID
	public final static String PIPEIDSTR = "urn:jxta:uuid-59616261646162614E50472050325033C0C1DE89719B456691A596B983BA0E1004";

	/**
	 * 初始化参数
	 */
	public MulticastOutPipe(boolean waitForRendezvous,PipeService pipeService,
			PipeAdvertisement pipeAdv, String peerName, PeerGroup myGroup) {
		this.waitForRendezvous = waitForRendezvous;
		this.pipeService = pipeService;
		this.pipeAdv = pipeAdv;
		this.peerName = peerName;
		this.myGroup = myGroup;
	}
	/**
	 * 创建输出管道通告，管道ID是指定的，类型是多播型的
	 * @return
	 */
	public static PipeAdvertisement getPipeAdvertisement() {
		PipeID pipeID = null;
		try {
			pipeID = (PipeID) IDFactory.fromURI(new URI(PIPEIDSTR));
		} catch (URISyntaxException use) {
			use.printStackTrace();
		}
		PipeAdvertisement advertisement = (PipeAdvertisement) AdvertisementFactory
				.newAdvertisement(PipeAdvertisement.getAdvertisementType());
		advertisement.setPipeID(pipeID);
		advertisement.setType(PipeService.PropagateType);
		advertisement.setName("Pipe tutorial");
		return advertisement;
	}
	
	public synchronized void run() {
		try {
			// 获取默认点组的集合点服务
			rendezvous = myGroup.getRendezVousService();

			// 注册集合点服务的时间处理对象
			rendezvous.addListener(this);
			pipeService.createOutputPipe(pipeAdv, this);
			if (!rendezvous.isConnectedToRendezVous()) {
				try {
					System.out.println("等待连接集合点");
					wait();
					System.out.println("集合点连接成功");
					pipeService.createOutputPipe(pipeAdv, this);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}

			}

		} catch (IOException e) {
			System.out.println("创建输出管道失败");
			e.printStackTrace();
			System.exit(-1);
		}
	}

	/**
	 * 输出管道的监听事件
	 */
	public void outputPipeEvent(OutputPipeEvent event) {
		// 获得输出管道的对象
		outputPipe = event.getOutputPipe();
		Message msg;
		try {
			msg = new Message();
			while (true) {
				msg = new Message();
				// 将键盘输入的字符串加到消息中
				StringMessageElement sme = new StringMessageElement(
						MESSAGE_NAME_SPACE, getStrFromKey(), null);
				msg.addMessageElement(null, sme);

				// 将这个对等点的名字加到消息中
				StringMessageElement peerN = new StringMessageElement(
						"peerName", peerName, null);
				msg.addMessageElement(null, peerN);
				// 发送消息
				outputPipe.send(msg);
			}
		} catch (IOException e) {
			System.out.println("failed to send message");
			e.printStackTrace();
			System.exit(-1);
		}
		// stop();
	}

	/**
	 * 处理集合点事件
	 */
	public synchronized void rendezvousEvent(RendezvousEvent event) {
		if (event.getType() == event.RDVCONNECT
				|| event.getType() == event.RDVRECONNECT) {
			notify();
		}
	}

	/**
	 * 从键盘获得输入的字符串
	 * 
	 * @return 返回键盘输入的字符串
	 */
	public String getStrFromKey() {
		try {
			BufferedReader br = new BufferedReader(new InputStreamReader(
					System.in));
			return br.readLine();
		} catch (IOException ioe) {
			System.err.println("从键盘读取字符串时发生I/O错误!");
			System.exit(-1);
		}
		return null;
	}

}