helloworld.java

jboss的jms一个小程序

import java.io.FileInputStream;
import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;

import net.jxta.document.AdvertisementFactory;
import net.jxta.document.MimeMediaType;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.StringMessageElement;
import net.jxta.peergroup.PeerGroup;
import net.jxta.protocol.PipeAdvertisement;
import util.jxta.JxtaHelper;
import btpjxta.BTPPipeService;
import btpjxta.BTPService;
import btpjxta.BTPServiceFactory;
import btpjxta.Context;
import btpjxta.InputBTPPipe;
import btpjxta.MessageUtils;
import btpjxta.OutputBTPPipe;

public class HelloWorld {
	private InputBTPPipe inputPipe = null;
	private JxtaHelper jxtaHelper = null;
	private PeerGroup netPG = null;
	protected BTPService btpSvc = null;
	private String inputPipeAdvFile = "c:\\peer2-pipe.xml"; // 输入管道的通告文件路径
	private PipeAdvertisement inputPipeAdv = null; // 本对等点的输入管道的通告

	public void startJxta() {
		jxtaHelper = new JxtaHelper(null, null);
		jxtaHelper.setJxtaEnv(JxtaHelper.DEFAULT_JXTA_HOME + "-" + "peer2"); // 设置系统的环境变量
																				// .jxta-[peername]

		if (!jxtaHelper.createCustomNetPeerGroup("helleworld", "helloword")) { // 创建自定义的点组
			System.out.println("创建点组失败");
		} else {
			System.out.println("创建点组成功");
		}

		netPG = jxtaHelper.startJxta(); // 启动JXTA平台
		if (netPG == null) {
			System.out.println("点组为空");
		} else {
			System.out.println("点组id ：" + netPG.getPeerGroupID());
			System.out.println("点组名称 ：" + netPG.getPeerGroupName());
			System.out.println("点组通告 ：" + netPG.getPeerGroupAdvertisement());
			System.out.println("对等点ID ：" + netPG.getPeerID());
			System.out.println("对等点名称 ：" + netPG.getPeerName());
		}

		if(connectToRdvPeer()) // 成为一个集合点
		{
			System.out.println("连接成功");
		}

		try {
			BTPServiceFactory bsf = BTPServiceFactory.newInstance();
			this.btpSvc = bsf.getBTPService(jxtaHelper.getCurrentPeerGroup(),
					"peer2");
		} catch (Exception ex) {
			System.out.println("Cannot start BTPService" + ex);
		}
	}
	private boolean connectToRdvPeer()
	{	
		boolean connectedToRdv = false;
		for (int i = 0; i < 5; i++)
		{
			if (jxtaHelper.connectToRdv("tcp://" + "localhost" + ":" + "9702")) 
			{
				connectedToRdv = true;
				break;
			}

			try
			{
				// wait for 5 seconds before another attempt
				Thread.sleep(5000);
			}
			catch (InterruptedException e) {}
		}
		
		if (!connectedToRdv)
		{
			return false;
		}

		return true;
	}
	public InputBTPPipe ctreatInputPipe() {
		BTPPipeService btpPipeSvc = btpSvc.getBTPPipeService();

		// load pipe advertisement from the file
		inputPipeAdv = (PipeAdvertisement) jxtaHelper.loadAdv(inputPipeAdvFile);
		if (inputPipeAdv == null) {
			System.out
					.println("Failed to load pipe advertisement from the file '"
							+ inputPipeAdvFile + "'");
		}

		System.out.println("Going to create an input pipe");

		// create new InputBTPPipe instance for given pipe advertisement
		inputPipe = btpPipeSvc.createInputBTPPipe(inputPipeAdv);
		if (inputPipe == null) {
			System.out.println("Failed to create the input pipe");
		}

		if (!jxtaHelper.publishPipeAdv(inputPipeAdv, 600000)) {
			System.out.println("Failed to publish the pipe advertisement");
		}

		System.out.println("Input pipe is created");
		return inputPipe;
	}

	/**
	 * 一直在监听输入管道的消息，直到收到消息为止 Wait for a message with context indefinitely.
	 */
	private Message waitForMessage() {
		inputPipe = ctreatInputPipe();
		if (inputPipe == null)
			return null;

		Message msg = null;

		while (msg == null) {
			try {
				msg = inputPipe.waitForMessage();
			} catch (InterruptedException ex) {
			}

			if (msg == null)
				continue;
		}

		if (msg != null)
			System.out.println("Message received");

		return msg;
	}

	public static void main(String srgs[]) {
		HelloWorld hellow = new HelloWorld();
		hellow.startJxta();

		System.out.println("Msg : "+hellow.waitForMessage());
	}

}