pipelistener.java

利用Eclipse开发的基于sun公司jxta协议的Jxta管道编程的通信程序

package jxta3;

import java.io.FileInputStream;
import java.util.Date;

import net.jxta.document.AdvertisementFactory;
import net.jxta.document.MimeMediaType;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.endpoint.WireFormatMessage;
import net.jxta.endpoint.WireFormatMessageFactory;
import net.jxta.endpoint.Message.ElementIterator;
import net.jxta.exception.PeerGroupException;
import net.jxta.peergroup.NetPeerGroupFactory;
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.protocol.PipeAdvertisement;
import net.jxta.util.CountingOutputStream;
import net.jxta.util.DevNullOutputStream;


public class PipeListener implements PipeMsgListener{
	static PeerGroup netPeerGroup = null;
	private final static String SenderMessage = "PipeListenerMsg";
	private PipeService pipe;
	private PipeAdvertisement pipeAdv;
	private InputPipe pipeIn = null;
	
	public static void main(String[] args){
		PipeListener myapp = new PipeListener();
		myapp.startJxta();
		myapp.run();
	}

	private void startJxta() {
		try{
			//启动默认点组
			netPeerGroup = new NetPeerGroupFactory().getInterface();
		} catch(PeerGroupException e){
			System.out.println("fatal error: group creation failure");
			e.printStackTrace();
			System.exit(1);
		}
		pipe = netPeerGroup.getPipeService();
		System.out.println("Reading in pipe.adv");
		try{
			//从文件中初始化管道通告
			FileInputStream is = new FileInputStream("d:\\pipe.adv");
			pipeAdv = (PipeAdvertisement)(PipeAdvertisement) AdvertisementFactory.newAdvertisement(
                    MimeMediaType.XMLUTF8,is);
			is.close();
		} catch(Exception e){
			System.out.println("failed to read/parse pipe advertisement");
			e.printStackTrace();
			System.exit(-1);
		}
	}

	private void run() {
		try{
			//创建一个异步输入管道，如果有消息到达就会调用pipeMsgEvent
			System.out.println("Creating input pipe");
			pipeIn = pipe.createInputPipe(pipeAdv,this);
		} catch(Exception e){
			return;
		}
		if(pipeIn == null){
			System.out.println("cannot open InputPipe");
			System.exit(-1);
		}
		System.out.println("Waiting for msgs on input pipe");
	}

	public void pipeMsgEvent(PipeMsgEvent event) {
		Message msg = null;
		try{
			//从事件中获取信息
			msg = event.getMessage();
			if(msg == null)
				return;
			printMessageStats(msg,true);
		} catch(Exception e){
			e.printStackTrace();
			return;
		}
		//获取所有的消息元素
		Message.ElementIterator en = msg.getMessageElements();
		if(!en.hasNext()){
			return;
		}
		//获取键值为SenderMessage的value
		MessageElement msgElement = msg.getMessageElement(null,SenderMessage);
		if(msgElement.toString() == null){
			System.out.println("null msg received");
		} else
		{
			Date date = new Date(System.currentTimeMillis());
			System.out.println("Message received at: "+ date.toString());
			System.out.println("Message created at: "+ msgElement.toString());
		}
	}

	private void printMessageStats(Message msg, boolean verbose) {
		try{
			CountingOutputStream cnt;
			ElementIterator it = msg.getMessageElements();
			System.out.println("----------BeginMessage-------------");
			WireFormatMessage serialed = WireFormatMessageFactory.toWire(msg,
					new MimeMediaType("application/x-jxta-msg"),(MimeMediaType[])null);
			System.out.println("Message Size: "+serialed.getByteLength());
			while(it.hasNext()){
				MessageElement el = (MessageElement)it.next();
				String eName = el.getElementName();
				cnt = new CountingOutputStream(new DevNullOutputStream());
				el.sendToStream(cnt);
				long size = cnt.getBytesWritten();
				System.out.println("Element" + eName + " : " + size);
				if(verbose){
					System.out.println("["+el+"]");
				}
			}
			System.out.println("------------EndMessage--------------");
			
		} catch(Exception e){
			e.printStackTrace();
		}
	}
}