jxta通信程序，不知道为什么不好使。
服务端：package primecruncher;
import net.jxta.peergroup.PeerGroup;
import net.jxta.peergroup.PeerGroupFactory;
import net.jxta.endpoint.MessageElement;
import net.jxta.discovery.DiscoveryService;
import net.jxta.pipe.PipeID;
import net.jxta.exception.PeerGroupException;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.document.*;
import net.jxta.id.IDFactory;
import net.jxta.pipe.PipeService;
import net.jxta.endpoint.Message;
import java.io.FileInputStream;
import net.jxta.util.JxtaServerPipe;
import net.jxta.util.JxtaBiDiPipe;
import net.jxta.endpoint.StringMessageElement;
import java.io.IOException;
import java.io.FileOutputStream;
import net.jxta.pipe.PipeMsgListener;
import net.jxta.pipe.PipeMsgEvent;
public class PrimePeer implements PipeMsgListener {
private PipeAdvertisement pipeAdv;
private JxtaServerPipe serverPipe;
private final static String SenderMessage = "pipe_tutorial";
private static PeerGroup group;
private static DiscoveryService discoSvc;
private static final String PIPE_ADV_FILE = "primeserver_pipe.adv";
private final static String ReturnCred = "Authentication credential ";
private final static String peerGroupID = "peer group ID ";
private final static String peerID = "peer ID ";
privateJxtaBiDiPipe bipipe;
private String xiaoxi;
privatePipeService pipeSvc;
public static void main(String[] args) {
PrimePeer pp = new PrimePeer();
pp.startJxta();
pp.doAdvertise();
}
public PrimePeer() {
}
//初始化
private void startJxta()
{
try {
group = PeerGroupFactory.newNetPeerGroup();//获取GROUP引用
discoSvc = group.getDiscoveryService();
pipeSvc = group.getPipeService();
} catch (PeerGroupException e) {
System.out.println("Cannot create Net Peer Group: " + e.getMessage(
));
System.exit(-1);
}
}
//创建广告并发布
private void doAdvertise() {
try {
FileInputStream is = new FileInputStream(PIPE_ADV_FILE);
pipeAdv = (PipeAdvertisement) AdvertisementFactory.newAdvertisement(MimeMediaType.XMLUTF8, is);
is.close();
System.out.println("ok here");
//discoSvc.publish( pipeAdv, DiscoveryService.ADV);
System.out.println("ok here too");
// we want to block until a connection is established
} catch (Exception e) {
pipeAdv = (PipeAdvertisement)AdvertisementFactory. newAdvertisement(
PipeAdvertisement.getAdvertisementType());
PipeID pid = IDFactory.newPipeID(group.getPeerGroupID());
pipeAdv.setName("serve");
pipeAdv.setPipeID(pid);
//save pipeAd in file
Document pipeAdDoc = pipeAdv.getDocument(new MimeMediaType ("text/xml"));
try {
FileOutputStream os = new FileOutputStream(PIPE_ADV_FILE);
pipeAdDoc.sendToStream(os);
os.flush();
os.close();
System.out.println("Wrote pipe advertisement to disk.");
} catch (IOException ex) {
System.out.println("Can't save pipe advertisement to file " +
PIPE_ADV_FILE);
System.exit(-1);
System.out.println("failed to read/parse pipe advertisement");
e.printStackTrace();
System.exit(-1);
}
}
try {
discoSvc.remotePublish(pipeAdv, DiscoveryService.ADV);
System.out.println("Published module spec adv");
} catch (Exception e) {
System.out.println("Trouble publishing module spec adv: " +
e.getMessage());
}
//等待连接
while(true){
System.out.println("starting ServerPipe");
try {
serverPipe = new JxtaServerPipe(group, pipeAdv);
serverPipe.setPipeTimeout(0);
System.out.println("starting");
bipipe = serverPipe.accept();
System.out.println("starting1 ");
if (bipipe != null ) {
System.out.println("BiDi Pipe created");
receiveAndSendTestMessage(bipipe);
}
} catch (Exception e) {
e.printStackTrace();
System.out.println("not created");
return;
}
}
}
//发消息
private void receiveAndSendTestMessage (JxtaBiDiPipe pipe) {
try {
Messagemsg = new Message();
StringMessageElement sme = new StringMessageElement(SenderMessage, ReturnCred, null);
msg.addMessageElement(null, sme);
StringMessageElement sme1 = new StringMessageElement(peerGroupID, group.getPeerGroupID().toString(), null);
msg.addMessageElement(null, sme1);
StringMessageElement sme2 = new StringMessageElement(peerID, group.getPeerID().toString(), null);
msg.addMessageElement(null, sme2);
System.out.println("Sending back 'Hello World'");
pipe.sendMessage(msg);
}
catch (Exception ie) {
ie.printStackTrace();
}
}
//接收消息
public void pipeMsgEvent(PipeMsgEvent ev) {
Message msg = ev.getMessage();
MessageElement msgElement = msg.getMessageElement(null, SenderMessage);
xiaoxi=msgElement.toString();
System.out.println(xiaoxi);
// Get message
if (msgElement.toString() == null) {
System.out.println("null msg received");
}
}
}
客户端代码：
package primecruncher;
import net.jxta.peergroup.PeerGroup;
import net.jxta.peergroup.PeerGroupFactory;
import net.jxta.endpoint.StringMessageElement;
import net.jxta.util.JxtaBiDiPipe;
import net.jxta.discovery.DiscoveryService;
import net.jxta.discovery.DiscoveryListener;
import net.jxta.discovery.DiscoveryEvent;
import net.jxta.pipe.PipeService;
import net.jxta.pipe.PipeMsgListener;
import net.jxta.exception.PeerGroupException;
import net.jxta.protocol.DiscoveryResponseMsg;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.pipe.PipeMsgEvent;
import net.jxta.rendezvous.RendezVousService;
import net.jxta.endpoint.Message;
import java.util.*;
import net.jxta.endpoint.MessageElement;
public class PrimeClient implements DiscoveryListener,PipeMsgListener{
private static PeerGroup group;
private Message msg,msg1;
private static DiscoveryService discoSvc;
private static PipeService pipeSvc;
String b;
String xiaoxi;
PipeAdvertisement pipeAdvert;
private RendezVousService rdv;
public PrimeClient(){}
public static void main(String[] args)
{
PrimeClient cl = new PrimeClient();
try{
cl.startJxta();
cl.doDiscovery();
System.out.println("wuyichang");
}catch(NullPointerException ee)
{ System.out.println("yichang");}
}
//初始化
private void startJxta() {
try {
group = PeerGroupFactory.newNetPeerGroup();
discoSvc = group.getDiscoveryService();
System.out.println("startJxta");
pipeSvc = group.getPipeService();
} catch (PeerGroupException e) {
System.out.println("Can't create net peer group: " +
e.getMessage());
System.exit(-1);
}
}
//远程搜索
private void doDiscovery()
{ discoSvc.addDiscoveryListener(this);
int peerQueryID = discoSvc.getRemoteAdvertisements(null, DiscoveryService.ADV,
null, null, 5);
System.out.println("sousuo");
}
//监听搜索结果并作出处理
public void discoveryEvent(DiscoveryEvent ev)
{ DiscoveryResponseMsg res = ev.getResponse();
Enumeration er = res.getAdvertisements();
if (er != null ) {
while (er.hasMoreElements())
{
pipeAdvert = (PipeAdvertisement)er.nextElement();
System.out.println("faxianguanggao");
}
}
//建立连接发送消息
try{
JxtaBiDiPipe bidipipe =newJxtaBiDiPipe(group,pipeAdvert,20000,this ,true );
msg = new Message();
StringMessageElement sme = new StringMessageElement("SenderMessage", "ReturnCred", null);
msg.addMessageElement(null, sme);
StringMessageElement sme1 = new StringMessageElement("a", group.getPeerGroupID().toString(), null);
msg.addMessageElement(null, sme1);
StringMessageElement sme2 = new StringMessageElement("b", group.getPeerID().toString(), null);
msg.addMessageElement(null, sme2);
bidipipe.sendMessage(msg);
System.out.println("fasong");
}catch(Exception e)
{
e.printStackTrace();
System.out.println("not created");
}
}
//接收消息
public void pipeMsgEvent(PipeMsgEvent ev) {
msg1 = ev.getMessage();
MessageElement msgElement = msg1.getMessageElement(null, "SenderMessage");
xiaoxi=msgElement.toString();
System.out.println(xiaoxi);
if (msgElement.toString() == null) {
System.out.println("null msg received");
}
}
}
页:
[1]