/** * the peers on the bus. * * @author Yannick Jestin * @author http://www.tls.cena.fr/products/ivy/ * * 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.6 * - major cleanup to handle simultaneous connections, e.g., between two * busses within the same process ( AsyncAPI test is very stressful ) * I made an assymetric processing to elect the client that should * disconnect based on the socket ports ... might work... * - jakarta regexp are not meant to be threadsafe, so for match() and * compile() must be enclaused in a synchronized block * - now sends back an error message when an incorrect regexp is sent * the message is supposed to be readable * - sendMsg has no more async parameter * 1.2.5: * - no more java ping pong * 1.2.5: * - use org.apache.regexp instead of gnu-regexp * http://jakarta.apache.org/regexp/apidocs/ * 1.2.4: * - sendBuffer goes synchronized * - sendMsg now has a async parameter, allowing the use of threads to * delegate the sending of messages * - API change, IvyException raised when \n or \0x3 are present in bus.sendMsg() * - breaks the connexion with faulty Ivy clients (either protocol or invalid * regexps, closes bug J007 (CM)) * - sendDie now always requires a reason * - invokes the disconnect applicationListeners at the end of the run() * loop. * closes Bug J006 (YJ) * - changed the access of some functions ( sendRegexp, etc ) to protected * 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 org.apache.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, aka BeginRegexp in other implementations */ final static int DirectMsg = 7;/* the peer sends a direct message */ final static int Die = 8; /* the peer wants us to quit */ 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 */ final static String escape ="\u001A"; final static char escapeChar = escape.charAt(0); final static char endArgChar = EndArg.charAt(0); final static char newLineChar = '\n'; // private variables private static Integer csMutex=new Integer(0); private static int clientSerial=0; /* an unique ID for each IvyClient */ private String messages_classes[] = null; private Ivy bus; private Socket socket; private BufferedReader in; private OutputStream out; private int remotePort=0; private volatile Thread clientThread;// volatile to ensure the quick communication private Integer clientKey; private boolean discCallbackPerformed = false; private String remoteHostname="unresolved"; // protected variables String appName="none"; Hashtable regexps = new Hashtable(); Hashtable regexpsText = new Hashtable(); static boolean debug = (System.getProperty("IVY_DEBUG")!=null) ; // int protocol; private boolean incoming; IvyClient() { } IvyClient(Ivy bus, Socket socket,int remotePort,boolean incoming) throws IOException { synchronized(csMutex) { clientKey=new Integer(clientSerial++); } this.bus = bus; this.remotePort = remotePort; this.incoming = incoming; in = new BufferedReader(new InputStreamReader(socket.getInputStream())); out = socket.getOutputStream(); incoming=(remotePort==0); traceDebug(((incoming)?"incoming":"outgoing")+" connection on "+socket); this.socket = socket; if (!incoming) { synchronized(bus) { bus.addHalf(this); // register a half connexion sendSchizo(); // the registering (handShake) will take place at the reception of the regexps... } } remoteHostname = socket.getInetAddress().getHostName(); clientThread = new Thread(this); // clientThread handles the incoming traffic clientThread.start(); } // 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 private void sendSchizo() throws IOException { traceDebug("sending our service port "+bus.applicationPort); Hashtable tosend=bus.selfIvyClient.regexpsText; sendString(SchizoToken,bus.applicationPort,bus.appName); for (Enumeration e = tosend.keys(); e.hasMoreElements(); ) { Integer ikey = (Integer)e.nextElement(); sendRegexp(ikey.intValue(),(String)tosend.get(ikey)); } sendString( EndRegexp,0,""); } synchronized private void handShake() throws IvyException { synchronized(bus) { bus.removeHalf(this); bus.addClient(this); } } public String toString() { return "IC["+clientKey+","+bus.getSerial()+"] "+bus.appName+":"+appName+":"+remotePort; } public String toStringExt() { return "client socket:"+socket+", remoteport:" + remotePort; } /** * returns the name of the remote agent. */ public String getApplicationName() { return appName ; } /** * returns the host name of the remote agent. * @since 1.2.7 */ public String getHostName() { return remoteHostname ; } /** * 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 * The content is not modifyable because String are not mutable, and cannot * be modified once they are create. * @see IvyClient#getRegexpsArray() getRegexpsArray to get a String[] result */ public Enumeration getRegexps() { return regexpsText.elements(); } /** * allow an Ivy package class to access the list of regexps at a * given time. * @since 1.2.4 */ public String[] getRegexpsArray() { String[] s = new String[regexpsText.size()]; int i=0; for (Enumeration e=getRegexps();e.hasMoreElements();) s[i++]=(String)e.nextElement(); return s; } /** * 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) throws IvyException { if ( (message.indexOf(IvyClient.newLineChar)!=-1)||(message.indexOf(IvyClient.endArgChar)!=-1)) throw new IvyException("newline character not allowed in Ivy messages"); sendString(DirectMsg,id,message); } /* closes the connexion to the peer */ protected void close(boolean notify) throws IOException { traceDebug("closing connexion to "+appName); if (notify) sendBye("hasta la vista"); stopListening(); socket.close(); // TODO is it necessary ? trying to fix a deadlock } /** * asks the remote client to leave the bus. * @param message the message that will be carried */ public void sendDie(String message) { sendString(Die,0,message); } /** * checks the "validity" of a regular expression. * @param exp the string being a regular expression * @return true if the regexp is valid * @since 1.2.4 */ public boolean CheckRegexp( String exp ) { boolean ok = true; if ( exp.startsWith( "^" )&&messages_classes!=null) { ok=false; for (int i=0 ; i < messages_classes.length;i++) { if (messages_classes[i].equals(exp.substring(1))) return true; } } return ok; } /////////////////////////////////////////////////// // // PROTECTED METHODS // /////////////////////////////////////////////////// static String decode(String s) { return s.replace(escapeChar,'\n'); } static String encode(String s) { return s.replace('\n',escapeChar); } Integer getClientKey() { return clientKey ; } protected void sendRegexp(int id,String regexp) {sendString(AddRegexp,id,regexp);} protected void delRegexp(int id) {sendString(DelRegexp,id,"");} protected int sendMsg(String message) { int count = 0; for (Enumeration e = regexps.keys();e.hasMoreElements();) { Integer key = (Integer)e.nextElement(); RE regexp = (RE)regexps.get(key); // re.match fails sometimes when it is called concurrently .. // see 28412 on jakarta regexp bugzilla synchronized (regexp) { if (!regexp.match(message)) continue; // no match } count++; // match sendResult(Msg,key,regexp); } return count; } /////////////////////////////////////////////////// // // PRIVATE METHODS // /////////////////////////////////////////////////// /* interrupt the listener thread */ private 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). * true if the peers are similar. This should not happen, it is bad */ protected int compareTo(IvyClient clnt) { // return clnt.clientKey.compareTo(clientKey); // Wrong. it's random... return (clnt.socket.getPort()-socket.getLocalPort()); } protected boolean equals(IvyClient clnt) { if (clnt==this) return false; // TODO go beyond the port number ! add some host processing, cf: // IvyWatcher ... if (remotePort==clnt.remotePort) return true; /* e.g. if (socket.getInetAddress()==null) return false; if (clnt.socket.getInetAddress()==null) return false; if (!socket.getInetAddress().equals(clnt.socket.getInetAddress())) return false; */ return false; } /* * the code of the thread handling the incoming messages. */ public void run() { traceDebug("Thread started"); Thread thisThread = Thread.currentThread(); String msg = null; try { traceDebug("connection established with "+ socket.getInetAddress().getHostName()+ ":"+socket.getPort()); } catch (Exception ie) { traceDebug("Interrupted while resolving remote hostname"); } while (clientThread==thisThread) { try { if ((msg=in.readLine()) != null ) { if (clientThread!=thisThread) break; // early stop during readLine() if (!newParseMsg(msg)) { close(true); break; } } else { traceDebug("readline null ! leaving the thead"); break; } } catch (IvyException ie) { traceDebug("caught an IvyException"); 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) { if (clientThread!=null) { traceDebug("abnormally Disconnected from "+ socket.getInetAddress().getHostName()+":"+socket.getPort()); } break; } } traceDebug("normally Disconnected from "+ appName); bus.removeClient(this); // invokes the disconnect applicationListeners if (!discCallbackPerformed) bus.clientDisconnects(this); discCallbackPerformed=true; traceDebug("Thread stopped"); } private synchronized 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 disconnect applicationListeners if (!discCallbackPerformed) bus.clientDisconnects(this); discCallbackPerformed=true; try { close(false); } catch (IOException ioe) { throw new IvyException("close failed"+ioe.getMessage()); } } } private void sendString(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 sendResult(int type,Integer id, RE regexp) { try { String buffer = type+" "+id+StartArg; for(int i=1;i15) ? c : 'X')+" "; } outDump += zu; return outDump; } private String dumpMsg(String s) { String deb = " \""+s+"\" "+s.length()+" cars, "; for (int i=0;i=b.length) { System.out.println("Ivy protocol error from "+appName); return false; } try { msgType = Integer.parseInt(s.substring(from,to)); } catch (NumberFormatException nfe) { System.out.println("Ivy protocol error on msgType from "+appName); return false; } from=to+1; while ((to=b.length) { System.out.println("Ivy protocol error from "+appName); return false; } try { msgId = new Integer(s.substring(from,to)); } catch (NumberFormatException nfe) { System.out.println("Ivy protocol error from "+appName+" "+s.substring(from,to)+" is not a number"); return false; } 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 String message=s.substring(from,b.length); bus.dieReceived(this,msgId.intValue(),message); // 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 if (!discCallbackPerformed) bus.clientDisconnects(this); discCallbackPerformed=true; try { close(false); } catch (IOException ioe) { throw new IvyException(ioe.getMessage()); } break; case AddRegexp: String regexp=s.substring(from,b.length); if ( CheckRegexp(regexp) ) { try { regexps.put(msgId,new RE(regexp)); regexpsText.put(msgId,regexp); bus.regexpReceived(this,msgId.intValue(),regexp); } catch (RESyntaxException e) { // the remote client sent an invalid regexp ! traceDebug("invalid regexp sent by " +appName+" ("+regexp+"), I will ignore this regexp"); sendBuffer(Error+e.toString()); } } else { throw new IvyException("regexp Warning exp='"+regexp+"' can't match removing from "+appName); } break; case DelRegexp: regexps.remove(msgId); String text=(String)regexpsText.remove(msgId); bus.regexpDeleted(this,msgId.intValue(),text); break; case EndRegexp: bus.clientConnects(this); if (bus.ready_message!=null) sendMsg(bus.ready_message); break; case Msg: Vector v = new Vector(); while (toIvyClient["+clientKey+","+bus.getSerial()+"] "+bus.appName+" (remote "+appName+")<-- "+s); } private void traceDebug(String[] tab){ String s = " string array " + tab.length + " elements: "; for (int i=0;i