a thread for each IvyClient when the connexion has been done
* @throws IvyException if there is a problem joining the bus
* @param domainbus a domain of the form 10.0.0:1234, A good practice is to
* sick to a null value, so that your agent will honor the IVY_BUS parameter
* given to the jvm (java -DIVYBUS= ... . Otherwise, you can provide some
* hard-coded value, similar to the
* netmask without the trailing .255. This will determine the meeting point
* of the different applications. Right now, this is done with an UDP
* broadcast. Beware of routing problems ! You can also use a comma
* separated list of domains.
*
* 1.2.8: goes synchronized. I don't know if it's really useful
*
*/
public final void start(final String domainbus) throws IvyException {
if (!stopped) throw new IvyException("cannot start a bus that's already started");
setStarting(true); // this will remain true entil one of the PacketSenders has finished
stopped=false;
String db = domainbus;
if (db == null) {
db = getDomain(null);
}
Properties sysProp = System.getProperties();
sysProp.put("IVYBUS" , db);
String range = (String)sysProp.get("IVYRANGE");
Matcher match;
if ((range != null)&&(match = rangeRE.matcher(range)).matches()) {
int rangeMin = Integer.parseInt(match.group(1));
int rangeMax = Integer.parseInt(match.group(2));
int index = rangeMin;
traceDebug("trying to allocate a TCP port between " + rangeMin + " and " + rangeMax);
boolean allocated = false;
while (!allocated) try {
if (index>rangeMax) throw new IvyException("no available port in IVYRANGE" + range );
synchronized (lockApp) {
app = new ServerSocket(index);
app.setSoTimeout(TIMEOUTLENGTH);
applicationPort = app.getLocalPort();
}
allocated = true;
} catch (BindException e) {
index++;
} catch (IOException e) {
throw new IvyException("can't open TCP service socket " + e );
}
}
else try {
synchronized (lockApp) {
app = new ServerSocket(0);
app.setSoTimeout(TIMEOUTLENGTH);
applicationPort = app.getLocalPort();
}
} catch (IOException e) {
throw new IvyException("can't open TCP service socket " + e );
}
traceDebug("lib: " + LIBVERSION + " protocol: " + PROTOCOLVERSION + " TCP service open on port " + applicationPort);
Domain[] d = parseDomains(db);
if (d.length == 0) throw new IvyException("no domain found in " + db);
watcherId = getWBUId().replace(' ' , '*'); // no space in the watcherId
// readies the rendezvous : an IvyWatcher (thread) per domain bus
for (Domain dom: d) watchers.addElement(new IvyWatcher(this , dom.domainaddr , dom.port));
serverThread = new Thread(this);
serverThread.start();
// sends the broadcasts and listen to incoming connexions
for (IvyWatcher iw: watchers) iw.doStart();
}
protected final Domain[] parseDomains(final String domainbus) {
StringTokenizer st = new StringTokenizer(domainbus , ",");
Domain[] d = new Domain[st.countTokens()];
int index = 0;
while ( st.hasMoreTokens()) {
String s = st.nextToken();
try {
d[index++] = new Domain(IvyWatcher.getDomain(s) , IvyWatcher.getPort(s));
} catch (IvyException ie) {
// do nothing
ie.printStackTrace();
}
}
// fixes the port values ...
int lastport = Ivy.DEFAULT_PORT;
for (index--; index >= 0; index--) {
Domain dom = d[index];
if (dom.port == 0) dom.port = lastport;
lastport = dom.port;
}
return d;
}
private void waitForRemote(String s) {
try {
while (starting==true) {
Thread.sleep(GRACEDELAY);
traceDebug("I'm waiting before "+s+", a starting tread is in progress");
}
} catch (InterruptedException ie) {
// should not happen, and it's not a problem anyway
}
}
/**
* disconnects from the Ivy bus.
*/
public final void stop() {
waitForRemote("stopping");
if (stopped) return;
stopped = true;
serverThread = null;
traceDebug("beginning stopping");
try {
// stopping the serverThread
Thread t = serverThread;
if (t != null) {
t.interrupt(); // The serverThread might be stopped even before having been created
// System.out.println("IZZZ joining " + t);
try { t.join(); } catch ( InterruptedException ie ) {
ie.printStackTrace();
}
}
synchronized (lockApp) { app.close(); }
// stopping the IvyWatchers
for (IvyWatcher iw: watchers) iw.doStop();
watchers.clear();
// stopping the remaining IvyClients
for (IvyClient c : clients.values()) {
if (c != null) {
c.close(true);
removeClient(c);
}
}
} catch (IOException e) {
traceDebug("IOexception Stop ");
}
traceDebug("end stopping");
}
/**
* Toggles the sending of messages to oneself, the remote client's
* IvyMessageListeners are processed first, and ourself afterwards.
* @param b true if you want to send the message to yourself. Default
* is false
* @since 1.2.4
*/
public final void sendToSelf(final boolean b) {
doSendToSelf = b;
}
/**
* do I send messsages to myself ?
* @return a boolean
* @since 1.2.4
*/
public final boolean isSendToSelf() {
return doSendToSelf;
}
/**
* selfIvyClient accesssor.
* @return our selfIvyClient
* @since 1.2.4
* @since 1.2.4
*/
public final SelfIvyClient getSelfIvyClient() {
return selfIvyClient;
}
/**
* Toggles the encoding/decoding of messages to prevent bugs related to the
* presence of a "\n".
* @param b true if you want to enforce encoding of newlines. Default
* is false. Every receiver will have to decode newlines
* @since 1.2.5
* The default escape character is a ESC 0x1A
*/
public final void protectNewlines(final boolean b) {
doProtectNewlines = b;
}
/**
* Performs a pattern matching according to everyone's regexps, and sends
* the results to the relevant ivy agents.
* @throws IvyException if there is a problem sending the message
* @param message A String which will be compared to the regular
* expressions of the different clients
* @return returns the number of messages actually sent
*/
public final int sendMsg(final String message) throws IvyException {
int count = 0;
waitForRemote("sending");
synchronized (lock) {
traceDebug("sending "+message);
String msg = message;
if (doProtectNewlines) msg = IvyClient.encode(message);
else if ( (msg.indexOf(IvyClient.newLineChar) != -1)||(msg.indexOf(IvyClient.endArgChar) != -1))
throw new IvyException("newline character not allowed in Ivy messages");
for ( IvyClient client : clients.values()) if (client != null) count += client.sendMsg(msg);
if (doSendToSelf) count+=selfIvyClient.sendSelfMsg(msg);
traceDebug("end sending "+message+" to "+count+" clients");
}
return count;
}
/**
* Subscribes to a regular expression.
*
* The callback will be executed with
* the saved parameters of the regexp as arguments when a message will sent
* by another agent. A program doesn't receive its own messages.
* Example:
*
the Ivy agent A performs
b.bindMsg("^Hello (*)",cb);
*
the Ivy agent B performs b2.sendMsg("Hello world");
*
a thread in A will uun the callback cb with its second argument set
* to a array of String, with one single element, "world"
* @param sregexp a perl regular expression, groups are done with parenthesis
* @param callback any objects implementing the IvyMessageListener
* interface, on the AWT/Swing framework
* @throws IvyException if there is a problem in the binding, be it regexp
* or network
* @return the id of the regular expression
*/
public final int bindMsg(final String sregexp , final IvyMessageListener callback ) throws IvyException {
return bindMsg(sregexp , callback , BindType.NORMAL);
}
/**
* Subscribes to a regular expression with asyncrhonous callback execution.
*
* Same as bindMsg, except that the callback will be executed in a separate
* thread each time.
* WARNING : there is no way to predict the order of execution
* of the * callbacks, i.e. a message received might trigger a callback before
* another one sent before
*
* @since 1.2.4
* @param sregexp a perl compatible regular expression, groups are done with parenthesis
* @param callback any objects implementing the IvyMessageListener
* interface, on the AWT/Swing framework
* @param type if set to NORMAL, it's a normal bind, if it's ASYNC, the
* callback will be created in a newly spawned Thread (Heavy ressources), if
* it's SWING, the callback will be deferred to the Swing Event Dispatch
* Tread
* @throws IvyException if there is a problem binding (network, regexp...)
* @return the int ID of the regular expression.
*/
public final int bindAsyncMsg(final String sregexp, final IvyMessageListener callback, BindType type ) throws IvyException {
return bindMsg(sregexp , callback , type);
}
/**
* Subscribes to a regular expression.
*
* The callback will be executed with
* the saved parameters of the regexp as arguments when a message will sent
* by another agent. A program doesn't receive its own messages,
* except if sendToSelf() is set to true.
* Example:
*
the Ivy agent A performs
b.bindMsg("^Hello (*)",cb);
*
the Ivy agent B performs b2.sendMsg("Hello world");
*
a thread in A will uun the callback cb with its second argument set
* to a array of String, with one single element, "world"
* @since 1.2.4
* @param sregexp a perl regular expression, groups are done with parenthesis
* @param callback any objects implementing the IvyMessageListener
* interface, on the AWT/Swing framework
* @param type if NORMAL (default) it's a normal bind, if ASYNC, each callback will be run in a separate thread, if SWING, the callback will be deferred to the Swing Event Dispatch Thread
* default is NORMAL
* @throws IvyException if there is a problem binding (regexp, network)
* @return the id of the regular expression
*/
public final int bindMsg(final String sregexp , final IvyMessageListener callback , final BindType type ) throws IvyException {
// adds the regexp to our collection in selfIvyClient
int key = selfIvyClient.bindMsg(sregexp , callback , type);
// notifies the other clients this new regexp
for (IvyClient c : clients.values() ) if (c != null) c.sendRegexp(key , sregexp);
return key;
}
/**
* Subscribes to a regular expression for one time only, useful for
* requests, in cunjunction with getWBUId().
*
* The callback will be executed once and only once, and the agent will
* unsubscribe
* @since 1.2.8
* @param sregexp a perl regular expression, groups are done with parenthesis
* @param callback any objects implementing the IvyMessageListener
* interface, on the AWT/Swing framework
* @throws IvyException if there is a problem during the binding
* @return the id of the regular expression
*/
public final int bindMsgOnce(final String sregexp, final IvyMessageListener callback ) throws IvyException {
Once once = new Once(callback);
int id = bindMsg(sregexp , once);
once.setRegexpId(id);
return id;
}
/**
* unsubscribes a regular expression using the id provided at bind time.
*
* @param id the id of the regular expression, returned when it was bound
* @throws IvyException if the id is not valid anymore
*/
public final void unBindMsg(final int id) throws IvyException {
selfIvyClient.unBindMsg(id);
for (IvyClient ic : clients.values() ) if (ic != null) ic.delRegexp(id );
}
/**
* unsubscribes a regular expression based on its string.
*
* @return a boolean, true if the regexp existed, false otherwise or
* whenever an exception occured during unbinding
* @param re the string for the regular expression
*/
public final boolean unBindMsg(final String re) { return selfIvyClient.unBindMsg(re); }
/**
* adds a bind listener to a bus.
* @param callback is an object implementing the IvyBindListener interface
* @return the id of the bind listener, useful if you wish to remove it later
* @since 1.2.4
*/
public final int addBindListener(final IvyBindListener callback){
ivyBindListenerList.addElement(callback);
return ivyBindListenerList.indexOf(callback);
}
/**
* removes a bind listener.
* @param id the id of the bind listener to remove
* @throws IvyException if id is not known
* @since 1.2.4
*/
public final void removeBindListener(final int id) throws IvyException {
try {
ivyBindListenerList.removeElementAt(id);
} catch (ArrayIndexOutOfBoundsException aie) {
throw new IvyException(id + " is not a valid Id");
}
}
/**
* adds an application listener to a bus.
* @param callback is an object implementing the IvyApplicationListener
* interface
* @return the id of the application listener, useful if you wish to remove
* it later
*/
public synchronized final int addApplicationListener(final IvyApplicationListener callback){
ivyApplicationListenerList.addElement(callback);
return ivyApplicationListenerList.indexOf( callback );
}
/**
* removes an application listener.
* @param id the id of the application listener to remove
* @throws IvyException if there is no such id
*/
public synchronized final void removeApplicationListener(final int id) throws IvyException {
try {
ivyApplicationListenerList.removeElementAt(id);
} catch (ArrayIndexOutOfBoundsException aie) {
throw new IvyException(id + " is not a valid Id");
}
}
/**
* sets the filter expression.
* @param f the extensive list of strings beginning the messages
* @since 1.2.9
*
* once this filter is set, when a client subscribes to a regexp of the
* form "^dummystring...", there is a check against the filter list. If no
* keyword is found to match, the binding is just ignored.
*/
public final synchronized void setFilter(final String[] f){
filter = java.util.Arrays.copyOf(f , f.length);
}
static {
// compiles the static regexps
try {
rangeRE = Pattern.compile("(\\d+)-(\\d+)"); // tcp range min and max
bounded = Pattern.compile("^\\^([a-zA-Z0-9_-]+).*");
} catch ( PatternSyntaxException res ) {
res.printStackTrace();
System.out.println("Regular Expression bug in Ivy source code ... bailing out");
}
}
/**
* checks the "validity" of a regular expression if a filter has been set.
* @since 1.2.9
* @param exp a string regular expression
* TODO must it be synchronized ( RE was not threadsafe, java regexp is )
*/
public final boolean checkRegexp(final String exp) {
if (filter == null) return true; // there's no message filter
Matcher m = bounded.matcher(exp);
if (!m.matches()) return true; // the regexp is not bounded
//System.out.println("the regexp is bounded, "+bounded.getParen(1));
// else the regexp is bounded. The matching string *must* be in the filter
String prems = m.group(1);
for (String f: filter) if (f.compareTo(prems) == 0) return true;
// traceDebug(" classFilter ["+filter[i]+"] vs regexp ["+prems+"]");
return false;
}
// a private class used by bindMsgOnce, to ensure that a callback will be
// executed once, and only once
private class Once implements IvyMessageListener {
private boolean received = false;
private int id = -1;
private IvyMessageListener ocallback = null;
Once(final IvyMessageListener callback){ ocallback = callback; }
synchronized void setRegexpId(final int fid){ id = fid; }
public void receive(final IvyClient ic , final String[] args){
synchronized(Once.this) {
// synchronized because it will most likely be called
// concurrently, and I *do* want to ensure that it won't
// execute twice
if (received||(ocallback == null)||(id == -1)) return;
received = true;
try { Ivy.this.unBindMsg(id); } catch (IvyException ie) { ie.printStackTrace(); }
ocallback.receive(ic , args);
}
}
}
/* invokes the application listeners upon arrival of a new Ivy client */
protected synchronized final void clientConnects(final IvyClient client){
for (IvyApplicationListener ial : ivyApplicationListenerList) ial.connect(client);
}
/* invokes the application listeners upon the departure of an Ivy client */
protected synchronized final void clientDisconnects(final IvyClient client){
for (IvyApplicationListener ial : ivyApplicationListenerList) ial.disconnect(client);
}
/* invokes the bind listeners */
protected final void regexpReceived(final IvyClient client , final int id , final String sregexp){
for (IvyBindListener ibl : ivyBindListenerList) ibl.bindPerformed(client , id , sregexp);
}
/* invokes the bind listeners */
protected final void regexpDeleted(final IvyClient client , final int id , final String sregexp){
for (IvyBindListener ibl : ivyBindListenerList) ibl.unbindPerformed(client , id , sregexp);
}
/*
* invokes the application listeners when we are summoned to die
* then stops
*/
protected synchronized final void dieReceived(final IvyClient client , final int id , final String message){
for (IvyApplicationListener ial : ivyApplicationListenerList) ial.die(client , id , message);
}
/* invokes the direct message callbacks */
protected synchronized final void directMessage(final IvyClient client , final int id , final String msgarg ){
for (IvyApplicationListener ial : ivyApplicationListenerList) ial.directMessage(client , id, msgarg);
}
/**
* gives the (Vectored) list of IvyClient at a given instant.
* @return a vector of IvyClients
*/
public final Vector getIvyClients() {
Vector v = new Vector();
for (IvyClient ic : clients.values() ) if (ic != null) v.addElement(ic);
return v;
}
/**
* gives a list of IvyClient with the name given in parameter.
*
* @param name The name of the Ivy agent you're looking for
* @return a vector of IvyClients
*/
public final Vector getIvyClientsByName(final String name) {
Vector v = new Vector();
String icname;
for (IvyClient ic : clients.values() ) {
if ( (ic == null)||((icname = ic.getApplicationName()) == null) ) break;
if (icname.compareTo(name) == 0) v.addElement(ic);
}
return v;
}
/**
* returns the domain bus.
*
* @param domainbus if non null, returns the argument
* @return It returns domainbus, if non null,
* otherwise it returns the IVYBUS property if non null, otherwise it
* returns Ivy.DEFAULT_DOMAIN
*/
public static final String getDomain(final String domainbus) {
String db = null;
db = domainbus;
if ( db == null ) db = System.getProperty("IVYBUS");
if ( db == null ) db = DEFAULT_DOMAIN;
return db;
}
/**
* returns the domain bus.
*
* @since 1.2.8
* @param progname The name of your program, for error message
* @param args the String[] of arguments passed to your main()
* @return returns the domain bus, ascending priority : ivy default bus, IVY_BUS
* property, -b domain on the command line
*/
public static final String getDomainArgs(final String progname, final String[] args) {
Getopt opt = new Getopt(progname , args , "b:");
int c;
if ( ((c = opt.getopt()) != -1) && c == 'b' ) return opt.getOptarg();
return getDomain(null);
}
/**
* returns a "wana be unique" ID to make requests on the bus.
*
* @since 1.2.8
* @return returns a string wich is meant to be noisy enough to be unique
*/
public final String getWBUId() {
return "ID<" + appName + myserial + ":" + nextId() + ":" + generator.nextInt() + ">";
}
private synchronized long nextId() { return current++; }
/**
* prints a human readable representation of the list of domains.
*
* @since 1.2.9
*/
public String domains(String toparse) {
StringBuffer s = new StringBuffer();
Ivy.Domain[] d = parseDomains(toparse);
for (Ivy.Domain dd : d) s.append(dd.getDomainaddr() + ":" + dd.getPort() + " ");
return s.toString();
}
/////////////////////////////////////////////////////////////////:
//
// Protected methods
//
/////////////////////////////////////////////////////////////////:
protected IvyClient createIvyClient(Socket s , int port, boolean domachin) throws IOException {
setStarting(true); // this one will stop when the client has finished starting
IvyClient i = new IvyClient(this , s , port , domachin);
i.doStart();
return i;
}
protected synchronized void removeClient(IvyClient c) {
synchronized(lock) {
clients.remove(c.getClientKey());
traceDebug("removed " + c + " from clients: " + getClientNames(clients));
}
}
protected synchronized void handShake(IvyClient c) {
synchronized(lock) {
removeHalf(c);
if (clients == null||c == null) return;
// TODO check if it's not already here !
IvyClient peer = searchPeer(c);
if ((peer == null) || peer.distanceTo(c)>0 ){
clients.put(c.getClientKey() , c);
setStarting(false);
traceDebug("added " + c + " in clients: " + getClientNames(clients));
} else {
traceDebug("not adding "+c+" in clients, double connexion detected, removing lowest one");
try {
c.close(false);
} catch (IOException ioe) {
// TODO
}
}
}
}
protected synchronized void addHalf(IvyClient c) {
synchronized(lock){ half.put(c.getClientKey() , c); }
traceDebug("added " + c + " in half: " + getClientNames(half));
}
protected synchronized void removeHalf(IvyClient c) {
synchronized(lock) {
if (half == null||c == null) return;
half.remove(c.getClientKey());
}
traceDebug("removed " + c + " from half: " + getClientNames(half));
}
/*
private synchronized boolean shouldIleave(IvyClient ic) {
traceDebug("looking for " + ic + " in " + getClientNames(half) + " and " + getClientNames(clients));
IvyClient peer = searchPeer(ic);
if (peer == null) return false;
boolean shoulda = peer.distanceTo(ic)>0;
traceDebug(ic + " " + ic.toStringExt() + ((shoulda) ? " must leave " : " must not leave"));
traceDebug(peer + " " + peer.toStringExt() + ((!shoulda) ? " must leave " : " must not leave"));
return shoulda;
}
*/
private synchronized IvyClient searchPeer(IvyClient ic) {
synchronized(lock) {
//for (Enumeration e = half.elements(); e.hasMoreElements(); ) {
// peer = e.nextElement();
// if ((peer != null)&&(peer.equals(ic))) return peer;
// }
for (IvyClient peer : clients.values()) if ((peer != null)&&(peer.equals(ic))) return peer;
}
return null;
}
/*
* the service socket thread reader main loop
*/
public void run() {
traceDebug("service thread started"); // THREADDEBUG
Thread thisThread = Thread.currentThread();
Socket socket = null;
while ( thisThread == serverThread ){
try {
synchronized (this) {
//System.out.println("DEBUG stopped: "+stopped);
if ((thisThread != serverThread)||stopped) break; // early disconnexion
}
synchronized (lockApp) {
socket = app.accept(); // TODO I can't synchronize on (this) in the run
}
synchronized (this) {
if ((thisThread != serverThread)||stopped) break; // early disconnexion
createIvyClient(socket , 0 , true); // the peer called me
}
} catch (InterruptedIOException ie) {
// traceDebug("server socket was interrupted. good");
if (thisThread != serverThread) break;
} catch( IOException e ) {
if (serverThread == thisThread) {
traceDebug("Error IvyServer exception: " + e.getMessage());
System.out.println("Ivy server socket reader caught an exception " + e.getMessage());
System.out.println("this is probably a bug in your JVM ! (e.g. blackdown jdk1.1.8 linux)");
throw new RuntimeException();
} else {
traceDebug("my server socket has been closed");
}
}
}
traceDebug("service thread stopped"); // THREADDEBUG
}
protected String getAppName() { return appName; }
protected int getAppPort() { return applicationPort; }
protected String getReadyMessage() { return ready_message; }
protected boolean getProtectNewlines() { return doProtectNewlines; }
protected void setStarting(boolean s) {
synchronized(readyToSend) {
traceDebug("setStarting "+s);
starting = s;
}
}
protected String getWatcherId() { return watcherId; }
protected int getBufferSize() { return bufferSize; }
protected int getSerial() { return myserial; }
private void traceDebug(String s){
if (debug) System.out.println("-->Ivy[" + myserial + "]<-- " + s);
}
// stuff to guarantee that all the treads have left
synchronized void registerThread(Thread t) { sendThreads.addElement(t); }
synchronized void unRegisterThread(Thread t) { sendThreads.removeElement(t); }
synchronized Thread getOneThread() {
if (sendThreads.size() == 0) return null;
return (Thread) sendThreads.firstElement();
}
// a small private method for debbugging purposes
private String getClientNames(Hashtable t) {
StringBuffer s = new StringBuffer();
s.append("(");
for (IvyClient ic : t.values() ) if (ic != null) s.append(ic.getApplicationName() + ",");
s.append(")");
return s.toString();
}
private static class Domain {
private String domainaddr;
private int port;
public Domain(String ddomainaddr , int dport) { this.domainaddr = ddomainaddr;this.port = dport; }
public String toString() { return domainaddr + ":" + port; }
public String getDomainaddr() { return domainaddr; }
public int getPort() { return port; }
}
} // class Ivy