path: root/src/IvyClient.java

/**
 * A private Class for the the peers on the bus.
 *
 * @author	Yannick Jestin
 * @author	<a href="http://www.tls.cena.fr/products/ivy/">http://www.tls.cena.fr/products/ivy/</a>
 *
 * each time a connexion is made with a remote peer, the regexp are exchanged
 * once ready, a ready message is sent, and then we can send messages, 
 * die messages, direct messages, add or remove regexps, or quit. A thread is
 * created for each remote client.
 *
 *  CHANGELOG:
 *  1.2.3:
 *  - silently stops on InterruptedIOException.
 *  - direct Messages
 *  - deals with early stops during readline
 *  1.2.2:
 *  - cleared a bug causing the CPU to be eating when a remote client left the
 *    bus. closes Damien Figarol bug reported on december, 2002. It is handled
 *    in the readline() thread
 *  1.2.1:
 *  - removes a NullPointerException when stops pinging on a pinger that
 *    wasn't even started
 *  1.0.12:
 *  - introducing a Ping and Pong in the protocol, in order to detect the loss of
 *    connection faster. Enabled through the use of -DIVY_PING variable only
 *    the PINGTIMEOUT value in milliseconds allows me to have a status of the
 *    socket guaranteed after this timeout
 *  - right handling of IOExceptions in sendBuffer, the Client is removed from
 *    the bus
 *  - sendDie goes public, so does sendDie(String)
 *  - appName visibility changed from private to protected
 *  1.0.10:
 *  - removed the timeout bug eating all the CPU resources
 */
package fr.dgac.ivy ;
import java.lang.Thread;
import java.net.*;
import java.io.*;
import java.util.*;
import gnu.regexp.*;

public class IvyClient implements Runnable {
    
  /* the protocol magic numbers */
  final static int Bye = 0;	/* end of the peer */
  final static int AddRegexp = 1;/* the peer adds a regexp */
  final static int Msg = 2 ;	/* the peer sends a message */
  final static int Error = 3; 	/* error message */
  final static int DelRegexp = 4;/* the peer removes one of his regex */
  final static int EndRegexp = 5;/* no more regexp in the handshake */
  final static int SchizoToken = 6; /* avoid race condition in concurrent connexions */
  final static int DirectMsg = 7;/* the peer sends a direct message */
  final static int Die = 8;  /* the peer wants us to quit */
  final static int Ping = 9;  /* checks the presence of the other */
  final static int Pong = 10;  /* checks the presence of the other */

  final static String MESSAGE_TERMINATOR = "\n"; /* the next protocol will use \r */
  final static String StartArg = "\u0002";/* begin of arguments */
  final static String EndArg = "\u0003"; /* end of arguments */


  private static boolean debug = (System.getProperty("IVY_DEBUG")!=null) ;
  private Ivy bus;
  private Socket socket;
  private BufferedReader in;
  private OutputStream out;
  private Hashtable regexp_in = new Hashtable();
  private Hashtable regexp_text = new Hashtable();
  private int appPort;
  private boolean peerCalling;
  private volatile Thread clientThread;// volatile to ensure the quick communication
  private Integer clientKey ;
  private static boolean doping = (System.getProperty("IVY_PING")!=null) ;
  final static int PINGTIMEOUT = 5000; 
  private PINGER pinger;
  private volatile Thread pingerThread;

  // protected variables
  String appName;

  IvyClient(Ivy bus, Socket socket,boolean peerCalling,Integer clientKey) throws IOException {
    appName = "Unknown";
    appPort = 0;
    this.bus = bus;
    this.socket = socket;
    this.peerCalling=peerCalling;
    this.clientKey=clientKey;
    in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
    out = socket.getOutputStream();
    Hashtable regexps=bus.regexp_out;
    // sends our ID, whether we initiated the connexion or not
    // the ID is the couple "host name,application Port", the host name
    // information is in the socket itself, the port is not known if we
    // initiate the connexion
    send(SchizoToken,bus.applicationPort,bus.appName);
    // sends our regexps to the peer
    for (Enumeration e = regexps.keys(); e.hasMoreElements(); ) {
      Integer ikey = (Integer)e.nextElement();
      sendRegexp( ikey.intValue(),(String)regexps.get(ikey));
    }
    send( EndRegexp,0,"");
    // spawns a thread to manage the incoming traffic on this
    // socket. We should be ready to receive messages now.
    clientThread = new Thread(this);
    clientThread .start();
    if (doping) {
      pinger = new PINGER();
      pingerThread=new Thread(pinger);
      pingerThread.start();
    }
  }

  /**
   *  returns the name of the remote agent.
   */
  public String getApplicationName() { return appName ; }

  Integer getClientKey() { return clientKey ; }

  /**
   *  allow an Ivy package class to access the list of regexps at a
   *  given time.
   *  perhaps we should implement a new IvyApplicationListener method to
   *  allow the notification of regexp addition and deletion
   */
  Enumeration getRegexps() { return regexp_text.elements(); }

  int getAppPort() { return appPort ; }

  void sendRegexp(int id,String regexp) {
    send(AddRegexp,id,regexp); /* perhaps we should perform some checking here */
  }

  public void delRegexp(int id) {send( DelRegexp,id,"");}

  /**
   * sends a direct message to the peer
   * @param id the numeric value provided to the remote client
   * @param message the string that will be match-tested
   */
  public void sendDirectMsg(int id,String message) { send(DirectMsg,id,message); }
  
  /**
   * closes the connexion to the peer.
   * @param notify should I send Bye message ?
   * the thread managing the socket is stopped
   */
  void close(boolean notify) throws IOException {
    traceDebug("closing connexion to "+appName);
    if (doping&&(pinger!=null)) { pinger.stopPinging(); }
    if (notify) sendBye("hasta la vista");
    stopListening();
    // bus.clientDisconnect(this);
    socket.close();  // should I also close in and out ?
  }

  /**
   * sends the substrings of a message to the peer for each matching regexp.
   * @param message the string that will be match-tested
   * @return the number of messages sent to the peer
   */
  int sendMsg( String message ) {
    int count = 0;
    for (Enumeration e = regexp_in.keys();e.hasMoreElements();) {
      Integer key = (Integer)e.nextElement();
      RE regexp = (RE)regexp_in.get(key);
      REMatch result = regexp.getMatch(message);
      if ( result != null ) {
        send(Msg,key,regexp.getNumSubs(),result);
	count++;
      }
    }
    return count;
  }
  
  void stopListening() {
    Thread t = clientThread;
    if (t==null) return; // we can be summoned to quit from two path at a time
    clientThread=null;
    t.interrupt();
  }

  /**
   * compares two peers the id is the couple (host,service port).
   * @param clnt the other peer
   * @return true if the peers are similir. This should not happen, it is bad
   * © ® (tm)
   */
  boolean sameClient( IvyClient clnt ) {
    return ( appPort != 0 && appPort == clnt.appPort )
      && ( getRemoteAddress() == clnt.getRemoteAddress() ) ;
  }

  /**
   * the code of the thread handling the incoming messages.
   */
  public void run() {
    Thread thisThread = Thread.currentThread();
    String msg = null;
    try {
      traceDebug("Connected from "+ socket.getInetAddress().getHostName()+ ":"+socket.getPort());
    } catch (Exception ie) {
      traceDebug("Interrupted while resolving remote hostname");
    }
    traceDebug("Thread started");
    while (clientThread==thisThread) {
      try {
	if ((msg=in.readLine()) != null ) {
	  if (clientThread!=thisThread) break; // early stop during readLine()
	  if (doping && (pingerThread!=null)) pingerThread.interrupt();
	  newParseMsg(msg);
	} else {
	  traceDebug("readline null ! leaving the thead");
	  break;
	}
      } catch (IvyException ie) {
	ie.printStackTrace();
      } catch (InterruptedIOException ioe) {
	traceDebug("I have been interrupted. I'm about to leave my thread loop");
	if (thisThread!=clientThread) break;
      } catch (IOException e) {
	traceDebug("abnormally Disconnected from "+ socket.getInetAddress().getHostName()+":"+socket.getPort());
	break;
      }
    }
    traceDebug("normally Disconnected from "+ appName);
    traceDebug("Thread stopped");
  }

  private void sendBuffer( String buffer ) throws IvyException {
    buffer += "\n";
    try {
      out.write(buffer.getBytes() );
      out.flush();
    } catch ( IOException e ) {
      traceDebug("I can't send my message to this client. He probably left");
      // first, I'm not a first class IvyClient any more
      bus.removeClient(this);
      // invokes the die applicationListeners
      bus.disconnectReceived(this);
      try {
        close(false);
      } catch (IOException ioe) {
	throw new IvyException("close failed"+ioe.getMessage());
      }
    }
  }

  private void send(int type, int id, String arg) {
    try {
      sendBuffer(type+" "+id+StartArg+arg);
    } catch (IvyException ie ) {
      System.err.println("received an exception: " + ie.getMessage());
      ie.printStackTrace();
    }
  }

  private void send(int type, Integer id, int nbsub, REMatch result) {
    String buffer = type+" "+id+StartArg;
    // Start at 1 because group 0 represent entire matching
    for(int sub = 1; sub <= nbsub; sub++) {
      if (result.getStartIndex(sub) > -1) {
	buffer += result.toString(sub)+EndArg;
      }
    }
    try {
      sendBuffer(buffer);
    } catch (IvyException ie ) {
      System.err.println("received an exception: " + ie.getMessage());
      ie.printStackTrace();
    }
  }

  private String dumpHex(String s) {
    byte[] b = s.getBytes();
    String out = "";
    String zu = "\t";
    for (int i=0;i<b.length;i++) {
      char c = s.charAt(i);
      out+=((int)c) + " ";
      zu+= ((c>15) ? c : 'X')+" ";
    }
    out += zu;
    return out;
  }

  private String dumpMsg(String s) {
    String deb = " \""+s+"\" "+s.length()+" cars, ";
    for (int i=0;i<s.length();i++) {
      deb+= "["+s.charAt(i) + "]:" + (int)s.charAt(i) +", ";
    }
    return s;
  }

  private void newParseMsg(String s) throws IvyException {
    byte[] b = s.getBytes();
    int from=0,to=0,msgType;
    Integer msgId;
    while ((to<b.length)&&(b[to]!=' ')) to++;
    if (to>=b.length) throw new IvyException("protocol error");
    try {
      msgType = Integer.parseInt(s.substring(from,to));
    } catch (NumberFormatException nfe) {
      throw new IvyException("protocol error on msgType");
    }
    from=to+1;
    while ((to<b.length)&&(b[to]!=2)) to++;
    if (to>=b.length) throw new IvyException("protocol error");
    try {
      msgId = new Integer(s.substring(from,to));
    } catch (NumberFormatException nfe) {
      throw new IvyException("protocol error on identifier");
    }
    from=to+1;
    switch (msgType) {
      case Die:
	traceDebug("received die Message from " + appName);
	// first, I'm not a first class IvyClient any more
	bus.removeClient(this);
	// invokes the die applicationListeners
	bus.dieReceived(this,msgId.intValue());
	// makes the bus die
	bus.stop();
	try {
	  close(false);
	} catch (IOException ioe) {
	  throw new IvyException(ioe.getMessage());
	}
	break;
      case Bye:
	// the peer quits
	traceDebug("received bye Message from "+appName);
	// first, I'm not a first class IvyClient any more
	bus.removeClient(this);
	// invokes the die applicationListeners
	bus.disconnectReceived(this);
	try {
	  close(false);
	} catch (IOException ioe) {
	  throw new IvyException(ioe.getMessage());
	}
	break;
      case AddRegexp:
	String regexp=s.substring(from,b.length);
	if ( bus.CheckRegexp(regexp) ) {
	  try {
	    regexp_in.put(msgId,new RE(regexp));
	    regexp_text.put(msgId,regexp);
	  } catch (REException e) {
	    throw new IvyException("regexp error " +e.getMessage());
	  }
	} else {
	  throw new IvyException("regexp Warning exp='"+regexp+"' can't match removing from "+appName);
	}
	break;
      case DelRegexp:
	regexp_in.remove(msgId);
	regexp_text.remove(msgId);
	break;
      case EndRegexp:
	bus.connect(this);
	/* 
	 * the peer is perhaps not ready to handle this message
	 * an assymetric processing should be written
	 */
	if (bus.ready_message!=null) sendMsg(bus.ready_message);
	break;
      case Msg:
	Vector v = new Vector();
	while (to<b.length) {
	  while ( (to<b.length) && (b[to]!=3) ) to++;
	  if (to<b.length) {
	    v.addElement(s.substring(from,to));
	    to++;
	    from=to;
	  }
	}
	String[] tab = new String[v.size()];
	for (int i=0;i<v.size();i++) {
	  tab[i]=(String)v.elementAt(i);
	  // for developpemnt purposes
	  // System.out.println(" *"+tab[i]+"* "+(tab[i]).length());
	}
	bus.callCallback(this,msgId,tab);
	break;
      case Pong:
	String paramPong=s.substring(from,b.length);
	traceDebug("Ping msg from "+appName+" : "+paramPong);
	break;
      case Ping:
	// I receive a ping. I can answer a pong.
	String param=s.substring(from,b.length);
	traceDebug("Ping msg from "+appName+" : "+param);
	sendPong(param);
	break;
      case Error:
	String error=s.substring(from,b.length);
	traceDebug("Error msg "+msgId+" "+error);
  	break;
      case SchizoToken:
	appName=s.substring(from,b.length);
	appPort=msgId.intValue();
	if ( bus.checkConnected(this) ) {
	  try  {
	    close(false);
	  } catch (IOException ioe) {
	  throw new IvyException("io " + ioe.getMessage());
	  }
	  throw new IvyException("Rare ! A concurrent connect occured");
	}
	break;
      case DirectMsg: 
	String direct=s.substring(from,b.length);
	bus.directMessage( this, msgId.intValue(), direct );
	break;
      default:
        throw new IvyException("protocol error, unknown message type "+msgType);
    }
  }

  void sendPong(String s) {send(Pong,0,s);}
  void sendPing(String s) {send(Ping,0,s);}

  private void sendBye() {send(Bye,0,"");}
  private void sendBye(String message) {send(Bye,0,message);}
  
  public void sendDie() { send(Die,0,""); }
  public void sendDie(String message) {send(Die,0,message);}

  private InetAddress getRemoteAddress() { return socket.getInetAddress(); }

  public String toString() {
    return "IvyClient "+bus.appName+":"+appName;
  }

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

  class PINGER implements Runnable {
    boolean isPinging = false;
    public void run() {
      isPinging=true;
      traceDebug("Pinger Thread started");
      while (isPinging) {
	try {
	  Thread.sleep(PINGTIMEOUT);
	  sendPing("are you here ?");
	} catch (InterruptedException ie) {
	}
      }
      traceDebug("Pinger Thread stopped");
    }
    public void stopPinging() { isPinging=false; pingerThread.interrupt();}
  }

}