path: root/src/Ivy.java

/**
 * a software bus package
 *
 * @author	Yannick Jestin
 * @author	<a href="http://www.tls.cena.fr/products/ivy/">http://www.tls.cena.fr/products/ivy/</a>
 *
 *<pre>
 *Ivy bus = new Ivy("Dummy agent","ready",null);
 *bus.bindMsg("(.*)",myMessageListener);
 *bus.start(getDomain(null));
 *</pre>
 *
 *  CHANGELOG:
 *  1.2.4:
 *    - added an accessor for doSendToSelf
 *    - waitForMsg() and waitForClient() to make the synchronization with
 *      other Ivy agents easier
 *    - with the bindAsyncMsg() to subscribe and perform each callback in a
 *      new Thread
 *    - bindMsg(regexp,messagelistener,boolean) allow to subscribe with a
 *      synchrone/asynch exectution
 *    - API change, IvyException raised when \n or \0x3 are present in bus.sendMsg()
 *    - bindListener are now handled
 *    - removeApplicationListener can throw IvyException
 *    - bus.start(null) now starts on getDomain(null), first the IVYBUS
 *    property, then the DEFAULT_DOMAIN, 127:2010
 *    - bindMsg() now throws an IvyException if the regexp is invalid !!!
 *    BEWARE, this can impact lots of programs ! (fixes J007)
 *    - no more includes the "broadcasting on " in the domain(String) method
 *    - new function sendToSelf(boolean) allow us to send messages to
 *    ourselves
 *  1.2.3:
 *    - adds a IVYBUS property to propagate the domain once set. This way,
 *    children forked through Ivy java can inherit from the current value.
 *    - adds synchronized flags to allow early disconnexion
 *  1.2.2:
 *    added the String domains(String d) function, in order to display the
 *  domain list
 *  1.2.1:
 *    bus.start(null) now starts on DEFAULT_DOMAIN. ( correction 1.2.4 This was not true.)
 *    added the getDomains in order to correctly display the domain list
 *    checks if the serverThread exists before interrupting it
 *    no has unBindMsg(String)
 *  1.2.0:
 *    setSoTimeout is back on the server socket
 *    added a regression test main()
 *    clients is now a Hashtable. the deletion now works better
 *    getIvyClientsByName allows the research of IvyClient by name
 *    getDomain doesnt throw IvyException anymore
 *    removed the close() disconnect(IvyClient c). Fixes a big badaboum bug
 *    getDomain becomes public
 *    adding the sendToSelf feature
 *    fixed the printStackTrace upon closing of the ServerSocket after a close()
 */
package fr.dgac.ivy ;
import java.net.*;
import java.io.*;
import java.util.*;

public class Ivy implements Runnable {

  /**
   * the name of the application on the bus
   */
  protected String appName;
  /**
   * the protocol version number
   */
  public static final int PROTOCOLVERSION = 3 ;
  public static final int PROTOCOLMINIMUM = 3 ;
  /**
   * the port for the UDP rendez vous, if none is supplied
   */
  public static final int DEFAULT_PORT = 2010 ;
  /**
   * the domain for the UDP rendez vous
   */
  public static final String DEFAULT_DOMAIN = "127.255.255.255:"+DEFAULT_PORT;
  /**
   * the library version, useful for development purposes only, when java is
   * invoked with -DIVY_DEBUG
   */
  public static final String libVersion ="1.2.4";

  private boolean debug;
  private static int clientSerial=0;	/* an unique ID for each IvyClient */
  private ServerSocket app;
  private Vector watchers;
  private volatile Thread serverThread; // to ensure quick communication of the end
  private Hashtable clients = new Hashtable();
  private Vector ivyApplicationListenerList = new Vector();
  private Vector ivyBindListenerList = new Vector();
  private Vector sendThreads = new Vector();
  private boolean stopped = false;
  protected int applicationPort;	/* Application port number */
  protected  String ready_message = null;
  protected boolean doProtectNewlines = false ;
  private boolean doSendToSelf = false ;
  protected SelfIvyClient selfIvyClient ;
  public final static int TIMEOUTLENGTH = 3000;
	
  /**
   * Readies the structures for the software bus connexion.
   *
   * All the dirty work is done un the start() method
   * @see #start
   * @param name The name of your Ivy agent on the software bus
   * @param message The hellow message you will send once ready
   * @param appcb A callback handling the notification of connexions and
   * disconnections, may be null
   */
  public Ivy( String name, String message, IvyApplicationListener appcb) {
    appName = name;
    ready_message =  message;
    debug = (System.getProperty("IVY_DEBUG")!=null);
    if ( appcb != null ) ivyApplicationListenerList.addElement( appcb );
    selfIvyClient=new SelfIvyClient(this,name);
  }

  /**
   * Waits for a message to be received
   *
   * @since 1.2.4
   * @param regexp the message we're waiting for to continue the main thread.
   * @param timeout in millisecond, 0 if infinite
   * @return the IvyClient who sent the message, or null if the timeout is
   * reached
   */
  public IvyClient waitForMsg(String regexp,int timeout) throws IvyException {
    Waiter w  = new Waiter(timeout);
    int re = bindMsg(regexp,w);
    IvyClient ic=w.waitFor();
    unBindMsg(re);
    return ic;
  }

  /**
   * Waits for an other IvyClient to join the bus
   *
   * @since 1.2.4
   * @param name the name of the client we're waiting for to continue the main thread.
   * @param timeout in millisecond, 0 if infinite
   * @return the first IvyClient with the name or null if the timeout is
   * reached
   */
  public IvyClient waitForClient(String name,int timeout) throws IvyException {
    IvyClient ic;
    if (name==null) throw new IvyException("null name given to waitForClient");
    // first check if client with the same name is on the bus
    for (Enumeration e=clients.elements();e.hasMoreElements();) {
       ic = (IvyClient)e.nextElement();
      if (name.compareTo(ic.getApplicationName())==0) return ic;
    }
    // if not enter the waiting loop
    WaiterClient w  = new WaiterClient(name,timeout);
    int i = addApplicationListener(w);
    ic=w.waitForClient();
    removeApplicationListener(i);
    return ic;
  }

  /**
   * connects the Ivy bus to a domain or list of domains.
   *
   * <li>One thread (IvyWatcher) for each traffic rendezvous (either UDP broadcast or TCPMulticast)
   * <li>One thread (serverThread/Ivy) to accept incoming connexions on server socket
   * <li>a thread for each IvyClient when the connexion has been done
   *
   * @param domainbus a domain of the form 10.0.0:1234, it is 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.
   *
   */
  public void start(String domainbus) throws IvyException {
    if (domainbus==null) domainbus=getDomain(null);
    Properties sysProp = System.getProperties();
    sysProp.put("IVYBUS",domainbus);
    try {
      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);
    watchers = new Vector();

    Domain[] d = parseDomains(domainbus);
    // readies the rendezvous : an IvyWatcher (thread) per domain bus
    for (int index=0;index<d.length;index++){
      IvyWatcher watcher =new IvyWatcher(this,d[index].domainaddr,d[index].port);
      watchers.addElement(watcher);
    }
    serverThread = new Thread(this);
    serverThread.start();
    // sends the broadcasts and listen to incoming connexions
    for (int i=0;i<watchers.size();i++){ ((IvyWatcher)watchers.elementAt(i)).start(); }
  }

  public Domain[] parseDomains(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
      }
    }
    return d;
  }

  /**
   * disconnects from the Ivy bus
   */
  public synchronized void stop() {
    if (stopped) return;
    stopped=true;
    traceDebug("beginning stopping the bus");
    try {
      // stopping the serverThread
      Thread t=serverThread;
      serverThread=null;
      if (t!=null) t.interrupt(); // The serverThread might be stopped even before having been created
      app.close();
      // stopping the IvyWatchers
      for (int i=0;i<watchers.size();i++){ ((IvyWatcher)watchers.elementAt(i)).stop(); }
      // stopping the remaining IvyClients
      for (Enumeration e=clients.elements();e.hasMoreElements();) {
        IvyClient c = (IvyClient)e.nextElement();
	c.close(true);
	removeClient(c);
      }
    } catch (IOException e) {
     traceDebug("IOexception Stop ");
    }
    traceDebug("the bus should have stopped so far");
  }

  /**
   * Toggles the sending of messages to oneself, the remote client's
   * IvyMessageListeners are processed first, and ourself afterwards.
   * @param boolean true if you want to send the message to yourself. Default
   * is false
   * @since 1.2.4
   */
  public void sendToSelf(boolean b) {doSendToSelf=b;}

  /**
   * @param boolean do I send message to myself ?
   * @since 1.2.4
   */
  public boolean isSendToSelf() {return doSendToSelf;}

  /**
   * returns our self IvyClient.
   * @since 1.2.4
   */
  public IvyClient getSelfIvyClient() {return selfIvyClient;}

  /**
   * Toggles the encoding/decoding of messages to prevent bugs related to the
   * presence of a "\n"
   * @param boolean true if you want to send the message to yourself. Default
   * is false.
   * @since 1.2.5?
   * The default escape character is a ESC 0x1A
   */
  private void protectNewlines(boolean b) {doProtectNewlines=b;}

  /**
   * Performs a pattern matching according to everyone's regexps, and sends
   * the results to the relevant ivy agents sequentially
   *
   * @param message A String which will be compared to the regular
   * expressions of the different clients
   * @return the number of messages actually sent
   */
  public int sendMsg(String message) throws IvyException {
    return sendMsg(message,false);
  }

  /**
   * Performs a pattern matching according to everyone's regexps, and sends
   * the results to the relevant ivy agents, using as many threads as needed.
   *
   * @since 1.2.4
   * @param message A String which will be compared to the regular
   * expressions of the different clients
   * @return always returns -1
   */
  public int sendAsyncMsg(String message,boolean async) throws IvyException {
    return sendMsg(message,true);
  }

  /**
   * Performs a pattern matching according to everyone's regexps, and sends
   * the results to the relevant ivy agents.
   *
   * @since 1.2.4
   * @param message A String which will be compared to the regular
   * expressions of the different clients
   * @param async if true, the sending will be performed in a separate thread,
   * default is false
   * @return if async is true, always returns -1, else returns the number of messages actually sent
   */
  public int sendMsg(String message,boolean async) throws IvyException {
    int count = 0;
    if (doProtectNewlines)	
      message=IvyClient.encode(message);
    else if ( (message.indexOf(IvyClient.newLineChar)!=-1)||(message.indexOf(IvyClient.endArgChar)!=-1))
      throw new IvyException("newline character not allowed in Ivy messages");
    for ( Enumeration e=clients.elements();e.hasMoreElements();) {
      IvyClient client = (IvyClient)e.nextElement();
      count += client.sendMsg(message, async);
    }
    if (doSendToSelf) count+=selfIvyClient.sendSelfMsg(message);
    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 <em>doesn't</em> receive its own messages.
   * <p>Example:
   * <br>the Ivy agent A performs <pre>b.bindMsg("^Hello (*)",cb);</pre>
   * <br>the Ivy agent B performs <pre>b2.sendMsg("Hello world");</pre>
   * <br>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 regexp a perl regular expression, groups are done with parenthesis
   * @param callback any objects implementing the IvyMessageListener
   * interface, on the AWT/Swing framework
   * @return the id of the regular expression
   */
  public int bindMsg(String sregexp, IvyMessageListener callback ) throws IvyException  {
    return bindMsg(sregexp,callback,false);
  }
    
  /**
   * 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 regexp a perl regular expression, groups are done with parenthesis
   * @param callback any objects implementing the IvyMessageListener
   * interface, on the AWT/Swing framework
   * @return the int ID of the regular expression.
   */
  public int bindAsyncMsg(String sregexp, IvyMessageListener callback ) throws IvyException  {
    return bindMsg(sregexp,callback,true);
  }

  /**
   * 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 <em>doesn't</em> receive its own messages.
   * <p>Example:
   * <br>the Ivy agent A performs <pre>b.bindMsg("^Hello (*)",cb);</pre>
   * <br>the Ivy agent B performs <pre>b2.sendMsg("Hello world");</pre>
   * <br>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 regexp a perl regular expression, groups are done with parenthesis
   * @param callback any objects implementing the IvyMessageListener
   * interface, on the AWT/Swing framework
   * @param async  if true, each callback will be run in a separate thread,
   * default is false
   * @return the id of the regular expression
   */
  public int bindMsg(String sregexp, IvyMessageListener callback,boolean async ) throws IvyException {
    // adds the regexp to our collection in selfIvyClient
    int key=selfIvyClient.bindMsg(sregexp,callback,async);
    // notifies the other clients this new regexp
    for (Enumeration e=clients.elements();e.hasMoreElements();) {
      IvyClient c = (IvyClient)e.nextElement();
      c.sendRegexp(key,sregexp);
    }
    return key;
  }

  /**
   * unsubscribes a regular expression
   *
   * @param id the id of the regular expression, returned when it was bound
   */
  public void unBindMsg(int id) throws IvyException {
    selfIvyClient.unBindMsg(id);
    for (Enumeration e=clients.elements();e.hasMoreElements();)
      ((IvyClient)e.nextElement()).delRegexp(id );
  }

  /**
   * unsubscribes a regular expression
   *
   * @return a boolean, true if the regexp existed, false otherwise or
   * whenever an exception occured during unbinding
   * @param String the string for the regular expression
   */
  public boolean unBindMsg(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 int addBindListener(IvyBindListener callback){
    ivyBindListenerList.addElement(callback);
    return ivyBindListenerList.indexOf(callback);
  }

  /**
   * removes a bind listener
   * @param id the id of the bind listener to remove
   * @since 1.2.4
   */
  public void removeBindListener(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 int addApplicationListener(IvyApplicationListener callback){
    ivyApplicationListenerList.addElement(callback);
    return ivyApplicationListenerList.indexOf( callback );
  }

  /**
   * removes an application listener
   * @param id the id of the application listener to remove
   */
  public void removeApplicationListener(int id) throws IvyException {
    try {
      ivyApplicationListenerList.removeElementAt(id);
    } catch (ArrayIndexOutOfBoundsException aie) {
      throw new IvyException(id+" is not a valid Id");
    }
  }

  /* invokes the application listeners upon arrival of a new Ivy client */
  protected void clientConnects(IvyClient client){
    for ( int i = 0 ; i < ivyApplicationListenerList.size(); i++ ) {
      ((IvyApplicationListener)ivyApplicationListenerList.elementAt(i)).connect(client);
    }
  }

  /* invokes the application listeners upon the departure of an Ivy client */
  protected void clientDisconnects(IvyClient client){
    for ( int i = 0 ; i < ivyApplicationListenerList.size(); i++ ) {
      ((IvyApplicationListener)ivyApplicationListenerList.elementAt(i)).disconnect(client);
    }
  }

  /* invokes the bind listeners */
  protected void regexpReceived(IvyClient client,int id,String sregexp){
    for ( int i = 0 ; i < ivyBindListenerList.size(); i++ ) {
      ((IvyBindListener)ivyBindListenerList.elementAt(i)).bindPerformed(client,id,sregexp);
    }
  }

  /* invokes the bind listeners */
  protected void regexpDeleted(IvyClient client,int id,String sregexp){
    for ( int i = 0 ; i < ivyBindListenerList.size(); i++ ) {
      ((IvyBindListener)ivyBindListenerList.elementAt(i)).unbindPerformed(client,id,sregexp);
    }
  }

  /*
   * removes a client from the list
   */
  void removeClient(IvyClient c) { clients.remove(c.getClientKey()); }

  /**
   * invokes the application listeners when we are summoned to die
   * then stops
   */
  public void dieReceived(IvyClient client, int id,String message){
    for ( int i=0 ;i<ivyApplicationListenerList.size();i++ ) {
      ((IvyApplicationListener)ivyApplicationListenerList.elementAt(i)).die(client,id,message);
    }
  }

  /* invokes the direct message callbacks */
  public void directMessage( IvyClient client, int id,String msgarg ){
    for ( int i = 0 ; i < ivyApplicationListenerList.size(); i++ ) {
      ((IvyApplicationListener)ivyApplicationListenerList.elementAt(i)).directMessage(client,id, msgarg);
    }
  }

  /**
   * gives the names of IvyClient(s) 
   */
  public Vector getIvyClients() {
    Vector v=new Vector();
    for (Enumeration e=clients.elements();e.hasMoreElements();) {
      v.addElement(e.nextElement());
    }
    return v;
  }

  /**
   * gives a list of IvyClient(s) with the name given in parameter
   *
   * @param name The name of the Ivy agent you're looking for
   */
  public Vector getIvyClientsByName(String name) {
    Vector v=new Vector();
    for (Enumeration e=clients.elements();e.hasMoreElements();) {
      IvyClient ic = (IvyClient)e.nextElement();
      if ( ((ic.getApplicationName()).compareTo(name))==0 ) v.addElement(ic);
    }
    return v;
  }

  /////////////////////////////////////////////////////////////////:
  //
  // Protected methods
  //
  /////////////////////////////////////////////////////////////////:

  synchronized void addClient(Socket socket,boolean peerCalling,int protocolVersion) throws IOException {
    if (stopped) return;
    IvyClient client = new IvyClient( this, socket,peerCalling,new Integer(clientSerial++),protocolVersion);
    clients.put(client.getClientKey(),client);
    traceDebug(getClientNames());
  }

  private static String[] myTokenize(String s,String separator) {
    int index=0, last=0, length=s.length();
    Vector v = new Vector();
    if (length!=0) while (true) {
      index=s.indexOf(separator,last);
      if (index==-1) {
	v.addElement(s.substring(last,length));
	break;
      } else if (index<s.length()) {
	v.addElement(s.substring(last,index));
	last=index+1;
      } else {
	break;
      }
    }
    String[] tab = new String[v.size()];
    v.copyInto(tab);
    return tab;
  }

  public static String getDomain(String domainbus) {
    if ( domainbus == null ) domainbus = System.getProperty("IVYBUS");
    if ( domainbus == null ) domainbus = DEFAULT_DOMAIN;
    return domainbus;
  }


  /*
   * prevents two clients from connecting to each other at the same time
   * there might still be a lingering bug here, that we could avoid with the
   * SchizoToken.
   */
  boolean checkConnected( IvyClient clnt ) {
    if ( clnt.getAppPort() == 0 ) return false;
    for (Enumeration e=clients.elements();e.hasMoreElements();) {
      IvyClient client = (IvyClient)e.nextElement();
      if ( clnt != client && client.sameClient( clnt ) ) return true;
    }
    return false;
  }

  /*
   * the service socket thread reader main loop
   */
  public void run() {
    // System.out.println("Ivy service Thread started"); // THREADDEBUG
    Thread thisThread=Thread.currentThread();
    while(thisThread==serverThread){
      try {
        Socket socket = app.accept();
	if ((thisThread!=serverThread)||stopped) break; // early disconnexion
	addClient(socket,true,0); // the peer called me TODO I don't know his protocol version
      } catch (InterruptedIOException ie) {
	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.printStackTrace();
	} else { traceDebug("my server socket has been closed"); }
      }
    }
    traceDebug("stopping the server Thread");
    // System.out.println("Ivy service Thread stopped"); // THREADDEBUG
  }


  private void traceDebug(String s){ if (debug) System.out.println("-->ivy<-- "+s); }

  synchronized void registerThread(Thread t) { sendThreads.add(t); }
  synchronized void unRegisterThread(Thread t) { sendThreads.remove(t); }
  synchronized Thread getOneThread() {
    if (sendThreads.size()==0) return null;
    return (Thread) sendThreads.firstElement();
  }

  void waitForAll() {
    Thread t;
    System.out.println("DEVELOPMENT WAITFORALL sendThreads size : " + sendThreads.size());
    try { while ((t=getOneThread())!=null) { t.join(); } }
    catch (InterruptedException ie) { System.out.println("waitForAll Interrupted"); }
    System.out.println("DEVELOPMENT END WAITFORALL");
  }

  /* a small private method for debbugging purposes */
  private String getClientNames() {
    String s = appName+" clients are: ";
    for (Enumeration e=clients.elements();e.hasMoreElements();){
      s+=((IvyClient)e.nextElement()).getApplicationName()+" ";
    }
    return s;
  }

  public String domains(String toparse) {
    String s="";
    Ivy.Domain[] d = parseDomains(toparse);
    for (int index=0;index<d.length;index++) {
      s+=d[index].getDomainaddr()+":"+d[index].getPort()+" ";
    }
    return s;
  }


  class Domain {
    private String domainaddr;
    private int port;
    public Domain(String domainaddr,int port) {this.domainaddr=domainaddr;this.port=port;}
    public String toString() {return domainaddr+":"+port;}
    public String getDomainaddr() { return domainaddr; }
    public int getPort() { return port; }
  }


  /*
   * unitary test. Normally, running java fr.dgac.ivy.Ivy should stop in 2.3 seconds :)
   */
  public static void main(String[] args) {
    Ivy bus = new Ivy("Test Unitaire","TU ready",null);
    try {
      bus.start(null);
      System.out.println("waiting 5 seconds for a coucou");
      System.out.println(((bus.waitForMsg("^coucou",5000))!=null)?"coucou received":"coucou not received");
      System.out.println("waiting 5 seconds for IvyProbe");
      System.out.println(((bus.waitForClient("IVYPROBE",5000))!=null)?"Ivyprobe joined the bus":"nobody came");
      bus.stop();
    } catch (IvyException ie) {
      ie.printStackTrace();
    }
  }
 
} // class Ivy

/* EOF */