package fr.dgac.ivy ; import java.lang.Thread; import java.net.*; import java.io.*; import java.util.*; import gnu.regexp.*; /** * A private Class for the the peers on the bus. * * @author François-Régis Colin * @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.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 */ 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 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; } /** * 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.stopPinging(); } if (notify) sendBye("hasta la vista"); stopListening(); // bus.clientDisconnect(this); socket.close(); // should I also close in and out ? } 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(); traceDebug("Connected from "+ socket.getInetAddress().getHostName()+ ":"+socket.getPort()); String msg = null; while ( clientThread==thisThread ) { try { if ((msg=in.readLine()) != null ) { if (doping && (pingerThread!=null)) pingerThread.interrupt(); newParseMsg(msg); } } catch (IvyException ie) { ie.printStackTrace(); } catch (InterruptedIOException ioe) { System.out.println("I have been interrupted. I'm about to leave my thread loop"); if (thisThread!=clientThread) break; } catch (IOException e) { if (clientThread!=thisThread) break; traceDebug("abnormally Disconnected from "+ socket.getInetAddress().getHostName()+":"+socket.getPort()); } } // while traceDebug("normally Disconnected from "+ appName); } 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;i15) ? c : 'X')+" "; } out += zu; return out; } private String dumpMsg(String s) { String deb = " \""+s+"\" "+s.length()+" cars, "; for (int i=0;i=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) 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 (toIvyClient "+bus.appName+":"+appName+"<-- "+s); } class PINGER implements Runnable { boolean isPinging = false; public void run() { isPinging=true; while (isPinging) { try { Thread.sleep(PINGTIMEOUT); sendPing("are you here ?"); } catch (InterruptedException ie) { } } } public void stopPinging() { isPinging=false; pingerThread.interrupt();} } } // class IvyClient /* EOF */