Counter counts an approximative number of messages going on the bus,

think xload, but for ivy messages. 1 second, 10 seconds, 1 minute

1 files changed, 152 insertions, 119 deletions

diff --git a/src/IvyClient.java b/src/IvyClient.java
index 4faeb00..cfbc320 100755
--- a/src/IvyClient.java
+++ b/src/IvyClient.java
@@ -31,7 +31,7 @@ public class IvyClient extends Thread {
   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 */
-  /* test pour jdk1.3 */
+  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 */
 
@@ -48,7 +48,7 @@ public class IvyClient extends Thread {
   private boolean peerCalling;
 
   IvyClient(Ivy bus, Socket socket,boolean peerCalling) throws IOException {
-    appName = "Unkown";
+    appName = "Unknown";
     appPort = 0;
     this.bus = bus;
     this.socket = socket;
@@ -72,8 +72,10 @@ public class IvyClient extends Thread {
     start();
   }
 
-
-  String getApplicationName() { return appName ; }
+  /**
+   *  returns the name of the remote agent.
+   */
+  public String getApplicationName() { return appName ; }
 
   /**
    *  allow an Ivy package class to access the list of regexps at a
@@ -100,10 +102,6 @@ public class IvyClient extends Thread {
       RE regexp = (RE)regexp_in.get(key);
       REMatch result = regexp.getMatch(message);
       if ( result != null ) {
-	// it's just to check if matching went right.... It is right.
-	//String tmp="";
-	//for (int i=1;i<=regexp.getNumSubs();i++) { tmp+="'"+result.toString(i)+"' "; }
-        //System.out.println(">> matching "+regexp.getNumSubs()+" blocks "+tmp);
         send(Msg,key,regexp.getNumSubs(),result);
 	count++;
       }
@@ -118,8 +116,6 @@ public class IvyClient extends Thread {
    */
   void close(String msg) throws IOException {
     traceDebug( msg );
-    socket.close(); // TODO it seems it doesnt stop the thread
-    out.close();
     in.close();
     gardefou=false;
   }
@@ -143,108 +139,23 @@ public class IvyClient extends Thread {
   public void run() {
     String msg = null;
     try {
-      traceDebug("Connected from "+ socket.getInetAddress().getHostName()+
-                 ":"+socket.getPort());
+      traceDebug("Connected from "+ socket.getInetAddress().getHostName()+ ":"+socket.getPort());
+      /*
+       * TODO bug ?!
+       * sometime we're stuck in readline while gardefou is false
+       * we need non blocking IO, found in the next 1.4 jdk ...
+       */
       while ( gardefou && ((msg=in.readLine()) != null )) {
-	int msgtype;	// type of the last message received
-	Integer msgid;	// ID of the last message received
-	RE regexp = null; // compiled regexp
-	String token = null;
-	/*
-	 * First stage: extract the message
-	 */
-        StringTokenizer st = new StringTokenizer(msg);
-        if(!st.hasMoreTokens()){close("Bad format no type '"+msg+"'");break;}
-        token=st.nextToken().trim();
-        if (token.length()==0){close("Bad format no type '"+msg+"'");break;}
-        try {
-	  msgtype  = Integer.parseInt(token);
-	} catch ( NumberFormatException e ) {
-          close("Bad format error parsing type'"+msg+"'");
-          break;
-        }
-        if(!st.hasMoreTokens()){close("Bad format no id '"+msg+"'");break;}
-	 
-        token=st.nextToken(StartArg).trim();
-        try {
-	  msgid=Integer.valueOf(token);
-	} catch ( NumberFormatException e ) {
-          close("Bad format error parsing id '"+token+"'");
-          break;
-        }
-	String msgarg="";
-	// if (st.hasMoreTokens()) msgarg=st.nextToken("\n").trim();
-	if (st.hasMoreTokens()) msgarg=st.nextToken("\n");
-	// TODO: here is a bug !
-	// quick and dirty fix: I do the trimming by hand later on.
-	// (cf ref: This method may be used to trim whitespace from the
-	// beginning and end of a string; in fact, it trims all ASCII control characters as well.           
-
-	/*
-	 * second stage: process the message
-	 */
-	switch (msgtype) {
-	case Bye: break;
-	case AddRegexp:
-	  msgarg.trim();
-          if ( bus.CheckRegexp(msgarg) ) {
-            try {
-	      regexp_in.put(msgid,new RE(msgarg));
-	      regexp_text.put(msgid,msgarg);
-	    } catch (REException e) {
-              System.err.println("Bad pattern: "+e.getMessage());
-            }
-	  } else {
-	    System.err.println(
-	   "Warning exp='"+msgarg+"' can't match removing from "+appName);
-	  }
-	  break;
-	case DelRegexp:
-	  regexp_in.remove(msgid);
-	  regexp_text.remove(msgid);
-	  break;
-	case EndRegexp:
-	  bus.connect(this);
-	  /* TODO
-	   * BUG ?  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:
-	  try {
-	    bus.callCallback(this,msgid,msgarg.substring(1,msgarg.length()-1));
-	  } catch (IvyException ie) {
-	    // calling an inexistant callback
-	    System.err.println("calling an inexistant callback, the caller must be wrong !");
-	  }
-	  break;
-	case Error:
-	  traceDebug("Error msg "+msgid+" "+msgarg);
-	  break;
-	case SchizoToken:
-	  appName=msgarg;
-	  appPort=msgid.intValue();
-	  if ( bus.checkConnected(this) ) {
-	    close("Quitting Application already connected");
-	    System.err.println("Rare ! A concurrent connect occured");
-          }
-	  break;
-	case DirectMsg:
-	  msgarg.trim();
-	  bus.directMessage( this, msgid.intValue(), msgarg );
-	  break;
-	case Die:
-	  bus.die( this,msgid.intValue());
-	  break;
-	default:
-	  System.err.println("*** IvyClient *** unhandled msg type "+
-	      	    msgtype+" "+msgid+msgarg);
-	  break;
-	} // switch 
-      } // while gardefou
-      traceDebug("normally Disconnected from "+
-       socket.getInetAddress().getHostName()+":"+socket.getPort());
+	try {
+	  newParseMsg(msg);
+	} catch (IvyException ie) {
+	  ie.printStackTrace();
+	}
+      }
+      traceDebug("normally Disconnected from "+ socket.getInetAddress().getHostName()+":"+socket.getPort());
+      socket.close();
+      out.close();
+      in.close();
     } catch (IOException e) {
       traceDebug("abnormally Disconnected from "+
         socket.getInetAddress().getHostName()+":"+socket.getPort());
@@ -253,22 +164,24 @@ public class IvyClient extends Thread {
     bus.removeClient( this );
   }
 
-  private void sendBuffer( String buffer ) {
+  private void sendBuffer( String buffer ) throws IvyException {
     buffer += "\n";
     try {
       out.write(buffer.getBytes() );
       out.flush();
     } catch ( IOException e ) {
-      /*
-       * TODO
-       * we should throw an exception here
-       */
-      System.err.println("IvyClient.sendBuffer.write failed. FIX ME");
+      throw new IvyException("IvyClient.sendBuffer.write failed: "+e.getMessage());
     }
   }
 
   private void send(int type, int id, String arg) {
-    sendBuffer(type+" "+id+StartArg+arg);
+    try {
+      sendBuffer(type+" "+id+StartArg+arg);
+    } catch (IvyException ie ) {
+      // TODO shoud fix the exception Handling here ...
+      System.err.println("received an exception: " + ie.getMessage());
+      ie.printStackTrace();
+    }
   }
 
   private void send(int type, Integer id, int nbsub, REMatch result) {
@@ -279,7 +192,127 @@ public class IvyClient extends Thread {
 	buffer += result.toString(sub)+EndArg;
       }
     }
-    sendBuffer(buffer);
+    try {
+      sendBuffer(buffer);
+    } catch (IvyException ie ) {
+      // TODO shoud fix the exception Handling here ...
+      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 Bye:
+	bus.die(this,msgId.intValue());
+	gardefou=false;
+	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);
+	/* TODO
+	 * BUG ?  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.add(s.substring(from,to));
+	    from=to;
+	    to++;
+	  }
+	}
+	String[] tab = new String[v.size()];
+	for (int i=0;i<v.size();i++) tab[i]=(String)v.elementAt(i);
+	bus.callCallback(this,msgId,tab);
+	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("Quitting Application already connected");
+	  } 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;
+      case Die:
+	gardefou=false;
+	bus.die(this,msgId.intValue());
+  	break;
+      default:
+        throw new IvyException("protocol error, unknown message type "+msgType);
+    }
   }
 
   private void sendDie() {send(Die,0,"");}