In order to install the testbed software on a 32-bit Windows 2000 / XP / Vista / 7 computer, please download the JXTA Testbed 2.1 installer. Beside your testbed login and a public IP address you need these TCP ports: 2022, 4000, 9700-9709 (default values can be changed during setup). They must never be occupied by another application and have to be open in the firewall. If the installation went well, your computer should appear at the bottom of the detailed status webpage in the dynamic nodes section. Note that the initial status is failure until it has been automatically configured.
In case you're interested in the core of the JXTA testbed, you'll find the sources in the SVN repository.
In order to write your code using the JXTA testbed, you have to include (at least) jxta.jar and bcprov-jdk14.jar of the JXTA library in your classpath as well as the testbed library aeolus.jar. Without these libraries, it won't compile.
If you plan to use any advanced features of JXTA, you may need additional jar files from the JXTA library.
In case you're interested in the internals of the JXTA testbed API or need them for debugging purposes, you'll find the source of the testbed API in the SVN repository.
If you'd like to write a JXTA service that just receives input from services users (unicast communication), you may use the class aeolus.edge.BasicService and attach a aeolus.edge.BasicServiceListener to it in order to receive incoming messages:
package aeolus.edge;
public interface BasicServiceListener {
public void messageReceived(net.jxta.endpoint.Message msg);
}To provide a unified interface together with user-defined services, you'll furthermore have to extend the abstract class aeolus.edge.ServiceImpl:
package aeolus.edge;
public abstract class ServiceImpl {
public ServiceImpl(net.jxta.peergroup.PeerGroup netPeerGroup, Settings settings) {}
public abstract Service getService();
}Thus, a minimal implementation looks like this:
package my.package;
import net.jxta.endpoint.*;
import aeolus.edge.*;
public class MyService extends ServiceImpl implements BasicServiceListener {
private BasicService service;
public MyService(PeerGroup netPeerGroup, Settings settings) {
super(netPeerGroup, settings);
service = new BasicService("JXTA-MY-SRVC", "My JXTA Service Description", "v1.0");
service.addBasicServiceListener(this);
// TODO: perform additional initialization tasks
// Attention: this code must not block! Do lengthy operations in seperate threads!
// Note: usually you'll start your service's main thread (containing your actual program) here
//MyMainThread mt = new MyMainThread(netPeerGroup);
//mt.start();
}
protected void finalize() {
service.removeBasicServiceListener(this);
}
public Service getService() {
return service;
}
public void messageReceived(Message msg) {
// process message, e.g. look for content in 'DataTag'
MessageElement msgElement = msg.getMessageElement(null, "DataTag");
if(msgElement != null)
System.out.println("message received; content: " + msgElement);
else
System.out.println("message received, but no DataTag found; message: " + msg);
}
}To discover the service just implemented from another peer, you may use the class aeolus.edge.ServiceDiscovery and implemenent the aeolus.edge.ServiceDiscoveryEventHandler interface in order to be notified of when a service is found:
package aeolus.edge;
public interface ServiceDiscoveryEventHandler {
public void handleServiceDiscoveryEvent(net.jxta.protocol.ModuleSpecAdvertisement mdsadv);
}Here is a minimal implementation, that looks for the JXTA-MY-SRVC service as long as it has found 3 (distinct or redundant) corresponding advertisements. Whenever a service is discovered, a hello world message is sent:
package my.package;
import java.net.*;
import net.jxta.endpoint.*;
import net.jxta.peergroup.*;
import net.jxta.pipe.*;
import net.jxta.protocol.*;
import aeolus.edge.*;
public class MyDiscovery implements ServiceDiscoveryEventHandler {
private volatile int found = 0;
private final Object responseMonitor = new Object();
private PeerGroup netPeerGroup;
private ServiceDiscovery discovery;
public MyDiscovery(PeerGroup netPeerGroup) {
this.netPeerGroup = netPeerGroup;
discovery = new ServiceDiscovery(netPeerGroup, this, "JXTA-MY-SRVC", 15000, 10);
}
public void waitForThreeHits() {
discovery.start(true);
synchronized(responseMonitor) {
while(found < 3) {
try { responseMonitor.wait(); } catch(InterruptedException ie) {}
}
}
discovery.stop();
}
public void handleServiceDiscoveryEvent(ModuleSpecAdvertisement mdsadv) {
try {
System.out.println("version of discovered service: " + mdsadv.getVersion());
OutputPipe outputPipe = netPeerGroup.getPipeService().createOutputPipe(mdsadv.getPipeAdvertisement(), 10000);
Message msg = new Message();
msg.addMessageElement(null, new StringMessageElement("DataTag", "hello world from " + InetAddress.getLocalHost().getCanonicalHostName(), null));
outputPipe.send(msg);
synchronized(responseMonitor) {
found += 1;
responseMonitor.notify();
}
} catch(Exception e) {
System.err.println("error sending message:");
e.printStackTrace();
}
}
}In order to let your service actually do something, you need to create a main thread that contains the actual code of the service. For the example above it could look like that:
package my.package;
import net.jxta.peergroup.PeerGroup;
public class MyMainThread extends Thread {
private PeerGroup netPeerGroup;
public MyThread(PeerGroup netPeerGroup) {
this.netPeerGroup = netPeerGroup;
}
public void run() {
// TODO: do something, e.g. discovery other services
//AnotherDiscovery ad = new AnotherDiscovery(netPeerGroup);
//ad.waitForThreeHits();
}
}If you'd like to establish bidirectional communication using sockets between users and your JXTA service, you may use the class aeolus.edge.AdvancedService and implement the aeolus.edge.AdvancedServiceConnectionHandler interface:
package aeolus.edge;
public interface AdvancedServiceConnectionHandler {
public void newConnection(java.net.Socket socket);
}Thus, a minimal service implementation looks like this:
package my.package;
import java.io.*;
import java.net.*;
import aeolus.edge.*;
public class MyService extends ServiceImpl implements AdvancedServiceConnectionHandler {
private AdvancedService service;
public MyService(PeerGroup netPeerGroup, Settings settings) {
super(netPeerGroup, settings);
service = new AdvancedService("JXTA-MY-SRVC", "My JXTA Service Description", "v1.0", 10, this);
// TODO: perform additional initialization tasks
// Attention: this code must not block! Do lengthy operations in seperate threads!
// Note: usually you'll start your service's main thread (containing your actual program) here
//MyMainThread mt = new MyMainThread(netPeerGroup);
//mt.start();
}
public Service getService() {
return service;
}
public void newConnection(Socket socket) {
try {
OutputStream out = socket.getOutputStream();
DataOutput dos = new DataOutputStream(out);
InputStream in = socket.getInputStream();
DataInput dis = new DataInputStream(in);
System.out.println("message received; content: " + dis.readUTF());
dos.writeUTF("welcome");
out.flush();
out.close();
in.close();
} catch(Exception e) {
System.err.println("error handling connection from client:");
e.printStackTrace();
} finally {
try { socket.close(); } catch(IOException io) {}
}
}
}The discovery is very similar to the unicast case, except that a socket is used. Here is a minimal implementation, that looks for the JXTA-MY-SRVC service as long as it has found 3 (distinct or redundant) corresponding advertisements. Whenever a service is discovered, a hello world message is sent:
package my.package;
import java.io.*;
import java.net.*;
import net.jxta.peergroup.*;
import net.jxta.protocol.*;
import net.jxta.socket.*;
import aeolus.edge.*;
public class MyDiscovery implements ServiceDiscoveryEventHandler {
private volatile int found = 0;
private final Object responseMonitor = new Object();
private PeerGroup netPeerGroup;
private ServiceDiscovery discovery;
public MyDiscovery(PeerGroup netPeerGroup) {
this.netPeerGroup = netPeerGroup;
discovery = new ServiceDiscovery(netPeerGroup, this, "JXTA-MY-SRVC", 15000, 10);
}
public void waitForThreeHits() {
discovery.start(true);
synchronized(responseMonitor) {
while(found < 3) {
try { responseMonitor.wait(); } catch(InterruptedException ie) {}
}
}
discovery.stop();
}
public void handleServiceDiscoveryEvent(ModuleSpecAdvertisement mdsadv) {
try {
System.out.println("version of discovered service: " + mdsadv.getVersion());
JxtaSocket socket = new JxtaSocket(netPeerGroup, null, mdsadv.getPipeAdvertisement(), 10000, true);
OutputStream out = socket.getOutputStream();
DataOutput dos = new DataOutputStream(out);
InputStream in = socket.getInputStream();
DataInput dis = new DataInputStream(in);
dos.writeUTF("hello world from " + InetAddress.getLocalHost().getCanonicalHostName());
out.flush();
System.out.println("response received; content: " + dis.readUTF());
out.close();
in.close();
socket.close();
synchronized(responseMonitor) {
found += 1;
responseMonitor.notify();
}
} catch(Exception e) {
System.err.println("error communicating with service:");
e.printStackTrace();
}
}
}If you'd like to write your own edge peer, e.g. to enable your stand-alone application to connect to the JXTA network, you should extend the class aeolus.edge.EdgePeer. Here is a simple example:
package my.package;
import java.io.*;
import net.jxta.exception.*;
import aeolus.edge.*;
public class MyApp extends EdgePeer {
public MyApp() throws IOException {
super();
}
public static void main(String[] args) throws IOException, PeerGroupException {
MyApp app = new MyApp();
if(!app.init("JXTA-MY-APP", "My JXTA Application Description")) {
System.err.println("could not initialize peer");
System.exit(1);
}
// BEGIN: testing code
MyDiscovery discovery = new MyDiscovery(app.netPeerGroup);
discovery.waitForThreeHits();
System.exit(0);
// END: testing code
}
}To run your edge peer, you need to properly setup the configuration files rdvs.txt, relays.txt, services.txt, and settings.txt, where rdvs.txt and relays.txt should contain a list of the rendezvous peers resp. relay peers in the form tcp://1.2.3.4:9701 (with 1.2.3.4 being the IP addess and 9701 the JXTA port). File services.txt can be empty; in case you just want to run an extra edge peer on a specific node (e.g. a computer with MySQL database), you can enter the services which you like to run here (ensure that these services can be found in the classpath). A template with default values for settings.txt is available here.
For some strange undocumented reasons (probably related to buffering and blocking I/O), JXTA doesn't support just any kind of stream. For example, if you would replace DataInputStream / DataOutputStream by ObjectInputStream / ObjectOutputStream in the code above, it won't work any more. But since sending Java objects it's quite a common task, let's look for a workaround:
OutputStream out = socket.getOutputStream(); DataOutput dos = new DataOutputStream(out); ... Object myObjectToSend; ... ByteArrayOutputStream baos = new ByteArrayOutputStream(); ObjectOutputStream oos = new ObjectOutputStream(baos); oos.writeObject(myObjectToSend); // send some more data here if applicable oos.close(); baos.close(); byte[] buffer = baos.toByteArray(); dos.writeInt(buffer.length); dos.write(buffer, 0, buffer.length); out.flush(); ... out.close();
To read the Java objects on the other peer, you could use code like this:
InputStream in = socket.getInputStream(); DataInput dis = new DataInputStream(in); ... Object myObjectToReceive; ... int len = dis.readInt(); byte[] buffer = new byte[len]; dis.readFully(buffer); ByteArrayInputStream bais = new ByteArrayInputStream(buffer); ObjectInputStream ois = new ObjectInputStream(bais); myObjectToReceive = ois.readObject(); // receive some more data here if applicable ois.close(); bais.close(); ... in.close();
In case you need Readers / Writers instead of I/O Streams, things get a bit simpler:
OutputStream out = socket.getOutputStream(); DataOutput dos = new DataOutputStream(out); ... CharArrayWriter caw = new CharArrayWriter(); // write your data caw.close(); dos.writeUTF(new String(caw.toCharArray())); out.flush(); ... out.close();
And the sample code for the receiver side:
InputStream in = socket.getInputStream(); DataInput dis = new DataInputStream(in); ... CharArrayReader car = new CharArrayReader(dis.readUTF().toCharArray()); // read your data back car.close(); ... in.close();