chat.java

JXTA的 即时通信系统 JXTACHAT

import net.jxta.discovery.DiscoveryService;
import net.jxta.document.MimeMediaType;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.Message.ElementIterator;
import net.jxta.endpoint.MessageElement;
import net.jxta.endpoint.WireFormatMessage;
import net.jxta.endpoint.WireFormatMessageFactory;
import net.jxta.peergroup.PeerGroup;
import net.jxta.pipe.InputPipe;
import net.jxta.pipe.PipeMsgEvent;
import net.jxta.pipe.PipeMsgListener;
import net.jxta.pipe.PipeService;
import net.jxta.platform.NetworkManager;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.util.CountingOutputStream;
import net.jxta.util.DevNullOutputStream;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

public class chat implements PipeMsgListener {

	static transient NetworkManager manager;
	static PeerGroup netPeerGroup = null;
	static PeerGroup myGroup = null;

	static PipeService pipeService;
	static PipeAdvertisement pipeAdv;
	private InputPipe inputPipe = null;

	// 初始化为NetPeerGroup的查找服务
	static DiscoveryService discoSvc;
	
	// 初始化为新建点组的查找服务
	static DiscoveryService discoMy;

	static List peers = new ArrayList();

	/**
	 * 初始化JXTA 创建默认的点组，再新建一个点组myGroup，加入到这个点组，获取管道服务，创建管道通告
	 */
	public chat() {
		// 创建默认点组
		manager = null;
		try {
			manager = new net.jxta.platform.NetworkManager(
					NetworkManager.ConfigMode.EDGE, "Peer1", new File(new File(
							".cache"), "Peer1").toURI());
			manager.setUseDefaultSeeds(true);
			manager.setConfigPersistent(true);
			manager.startNetwork();
		} catch (Exception e) {
			e.printStackTrace();
			System.exit(-1);
		}
		netPeerGroup = manager.getNetPeerGroup();

		// 获取默认点组的搜索服务
		discoSvc = netPeerGroup.getDiscoveryService();

		// 查找点组chatpg1，如果没有查找到，创建点组cgatpg1，如果找到就加入到这个点组
		JoinPeerGroup jpg = new JoinPeerGroup();
		try {
			myGroup = jpg.JoinPeerGroup(netPeerGroup, discoSvc);
		} catch (Exception e) {
			e.printStackTrace();
		}

		discoMy = myGroup.getDiscoveryService();
		// 获取新点组的管道服务
		pipeService = myGroup.getPipeService();
		// 创建管道通告
		pipeAdv = MulticastOutPipe.getPipeAdvertisement();
	}

	public static void main(String args[]) {
		chat server = new chat();

		// 搜索同点组的在线的对等点
		DiscoveryPeers dp = new DiscoveryPeers();
		dp.setHdisco(discoMy);
		dp.run();
		peers = dp.getPeers();
		if (peers.size() == 0) {
			System.out.println("现在没有在线用户");
		} else {
			for (int i = 0; i < peers.size(); i++)
				System.out.println("用户：" + peers.get(i).toString() + "在线");
		}

		// 创建输入管道，并监听
		server.start();

		// 创建输出管道，并监听
		String value = System.getProperty("RDVWAIT", "false");
		boolean waitForRendezvous = Boolean.valueOf(value);

		MulticastOutPipe mop = new MulticastOutPipe(waitForRendezvous,
				pipeService, pipeAdv, myGroup.getPeerName(), myGroup);
		mop.run();
	}

	/**
	 * 创建输入管道
	 */
	public void start() {
		try {
			inputPipe = pipeService.createInputPipe(pipeAdv, this);
		} catch (IOException io) {
			io.printStackTrace();
			return;
		}
		if (inputPipe == null) {
			System.out.println(" cannot open InputPipe");
			System.exit(-1);
		}
		System.out.println("开始聊天");
	}

	/**
	 * 当监听到有消息到达时候，就触发以下事件
	 */
	public void pipeMsgEvent(PipeMsgEvent event) {
		Message msg;
		try {
			// Obtain the message from the event
			msg = event.getMessage();
		} catch (Exception e) {
			e.printStackTrace();
			return;
		}

		Message.ElementIterator en = msg.getMessageElements();
		if (!en.hasNext()) {
			return;
		}

		MessageElement msgElement = msg.getMessageElement(null,
				MulticastOutPipe.MESSAGE_NAME_SPACE);
		MessageElement MsgPeerN = msg.getMessageElement(null, "peerName");

		if (msgElement.toString() == null) {
			System.out.println("null msg received");
		} else {
			if (MsgPeerN.toString().equals("Peer1")) {
				System.out.println("您说:" + msgElement.toString());
			} else {
				System.out.println(MsgPeerN.toString() + "说:"
						+ msgElement.toString());
			}
		}
	}
}