JxtaBiDiPipe questions
"giorgos d." <[email protected]>
| Newsgroups | gmane.comp.java.jxta.user |
|---|---|
| Message-ID | <004a01c59842$24adf190$07b35882@GWAIHIR> |
Hello people!
Im currently working on a distributed computing system using JXTA. A main
aspect of the application is distributing Tasks among peers. My first
attempt was using Messages and unicast pipes but the results were not very
good: sometimes Messages were lost on the way. Im trying now to use
BiDiPipes and i have some questions:
1. In the tutorial a server listens for a connection on his ServerPipe and
when that happens sends its msgs. The Client implements a listener on his
BiDiPipe (the other end of the ServerPipe?) and responds accordingly. What
about the client sending a request to the server? How the server handles
these events? The JXTAServerPipe implements an EventListener but how is this
used inside the application?
2. My first try with BiDiPipe was this:
- Peer A and B on one machine. Multicast TCP enabled on 127.0.0.1. No
HTTP. (also i never check the rendevouz peer on either of them, but i guess
thats not necessary)
- A sends a Request msg to B using B's unicast pipe. on the msg there is
a newly created _pipeAdvertisement (unpublished).
- A creates a new BiDiPipe and tris to connect() (with a very long
timeout)
using the new _pipeAdvertisment
- B reads the pipeAdvertisment from the Request msg and creates a
JXTAServerPipe
with it. Sets the timout to 0 and tries to accept(). Then send the Task
message.
I know that this is somehow the other way around but i thought that with a
big timeout the above procedure would be succesful. Well it isnt since Peer
A never manages to connect to the serverPipe.
Any suggestions/help would be very wellcome. Thank you!
Giorgos Doudalis.
No virus found in this outgoing message.
Checked by AVG Anti-Virus.
Version: 7.0.338 / Virus Database: 267.9.9/62 - Release Date: 02.08.2005
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
peerB.txt
(text/plain, 1.8 KB)
public void openInputPipe(PipeAdvertisement adv) {
PipeMsgListener myPipeListener = new PipeMsgListener () {
if (myMessage.getMessageElement("type").toString().equals("request")) {
System.out.println("Received REQUEST msg...");
stolenTask = getTask();
if (stolenTask!=null) {
stolenTask.setTaskOwner(writeAdvertisement(pipeAdv));
System.out.println("SEND:Task = f(" + stolenTask.getAnswer() + ")");
}
else
System.out.println("No TASK currently available");
PipeAdvertisement outputBiDiPipeAdv = readAdvertsement(myMessage.getMessageElement("pipe").toString());
sendRequestedTask(stolenTask, outputBiDiPipeAdv);
}
}
}
private void sendRequestedTask(FJTask requestedTask, PipeAdvertisement adv) {
System.out.println("Sending TASK message through pipe" + adv.getName()+ "...");
Message msg = new Message();
StringMessageElement sme1 = new StringMessageElement("type", "task" , null);
msg.addMessageElement(null, sme1);
try {
ByteArrayOutputStream myStream =
(ByteArrayOutputStream) new ByteArrayOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(myStream);
oos.reset();
oos.writeObject(requestedTask);
oos.close();
ByteArrayMessageElement sme2 = new ByteArrayMessageElement("payload", MimeMediaType.AOS, myStream.toByteArray(), null);
msg.addMessageElement(null, sme2);
serverBiDiPipe = new JxtaServerPipe(currentGroup, adv);
// we want to block until a connection is established
serverBiDiPipe.setPipeTimeout(0);
JxtaBiDiPipe bipipe = serverBiDiPipe.accept();
System.out.print("Trying to send...");
bipipe.sendMessage(msg);
System.out.println("Succesful!");
} catch (IOException e) {
System.out.println("Error sending TASK message");
e.printStackTrace();
}
}
peerA.txt
(text/plain, 1.4 KB)
public synchronized FJTask requestForTask(Object te) {
Message msg = new Message();
StringMessageElement sme1 = new StringMessageElement("type", "request" , null);
msg.addMessageElement(null, sme1);
PipeAdvertisement bdPipeAdv = createPipe(true);
StringMessageElement sme2 = new StringMessageElement("pipe", writeAdvertisement(bdPipeAdv) , null);
msg.addMessageElement(null, sme2);
System.out.println("Sending REQUEST message through pipe" + ((PipeAdvertisement) te).getName());
try {
myOutputPipe = pipes.createOutputPipe((PipeAdvertisement)te,10000);
myOutputPipe.send(msg);
} catch (IOException e) {
System.out.println("Error sending REQUEST message");
waitForTask = false;
e.printStackTrace();
}
try {
openBiDiPipe(bdPipeAdv);
} catch (IOException e1) {
System.out.println("Failed!");
e1.printStackTrace();
return null;
}
System.out.println("Successful!");
.
.
.
.
return stolenTask;
}
private void openBiDiPipe(PipeAdvertisement bdPipeAdv) throws IOException {
System.out.println("Creating the BiDi pipe to receive TASK");
bdPipe = new JxtaBiDiPipe();
bdPipe.setReliable(true);
System.out.print("Attempting to establish a BiDi connection...");
bdPipe.connect(currentGroup,
null,
bdPipeAdv,
180*1000,
// register as a message listener
new BiDiPipeListener());
}