diff options
| -rw-r--r-- | src/Ghost.java | 39 | ||||
| -rwxr-xr-x | src/Ivy.java | 92 | ||||
| -rwxr-xr-x | src/IvyClient.java | 4 | ||||
| -rw-r--r-- | src/IvyDaemon.java | 2 | ||||
| -rwxr-xr-x | src/IvyWatcher.java | 3 | ||||
| -rw-r--r-- | src/Makefile | 2 | ||||
| -rw-r--r-- | src/ProxyClient.java | 155 | ||||
| -rw-r--r-- | src/ProxyMaster.java | 140 |
8 files changed, 379 insertions, 58 deletions
diff --git a/src/Ghost.java b/src/Ghost.java new file mode 100644 index 0000000..ad6e78c --- /dev/null +++ b/src/Ghost.java @@ -0,0 +1,39 @@ +/** + * the Ghost peers on the bus ( Proxy scheme ) + * + * @author Yannick Jestin + * @author <a href="http://www.tls.cena.fr/products/ivy/">http://www.tls.cena.fr/products/ivy/</a> + * + * CHANGELOG: + * 1.2.12 + */ +package fr.dgac.ivy ; +import java.io.*; +import java.net.*; +import java.util.*; +import org.apache.regexp.*; + +class Ghost extends IvyClient { + + private String id; // given to the Proxy by the Master + private ProxyClient pc; + + Ghost(Ivy bus, Socket socket,int remotePort,boolean incoming,String id,ProxyClient pc) throws IOException { + super(bus,socket,remotePort,incoming); + this.id=id; + this.pc=pc; + } + + protected synchronized void sendBuffer( String buffer ) throws IvyException { + System.out.println("out buffer: "+buffer); + pc.forward(id,buffer); + super.sendBuffer(buffer); + } + + protected boolean newParseMsg(String s) throws IvyException { + System.out.println("in buffer: "+s); + pc.forward(id,buffer); + return super.newParseMsg(s); + } + +} diff --git a/src/Ivy.java b/src/Ivy.java index 7d26dff..2647434 100755 --- a/src/Ivy.java +++ b/src/Ivy.java @@ -13,6 +13,8 @@ *</pre> * * CHANGELOG: + * 1.2.12: + * - directMessage goes protected * 1.2.9: * - introducing setFilter() * - introducing IVYRANGE in to allow the bus service socket to start on a @@ -647,14 +649,14 @@ public class Ivy implements Runnable { } /* invokes the direct message callbacks */ - public void directMessage( IvyClient client, int id,String msgarg ){ + protected 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 IvyClient() at a given instant + * gives the (Vectored) list of IvyClient at a given instant */ public Vector getIvyClients() { Vector v=new Vector(); @@ -670,7 +672,7 @@ public class Ivy implements Runnable { } /** - * gives a list of IvyClient(s) with the name given in parameter + * gives a list of IvyClient with the name given in parameter * * @param name The name of the Ivy agent you're looking for */ @@ -689,32 +691,7 @@ public class Ivy implements Runnable { return v; } - /////////////////////////////////////////////////////////////////: - // - // Protected methods - // - /////////////////////////////////////////////////////////////////: - 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; - } - - /** + /** * returns the domain bus * * @param domainbus if non null, returns the argument @@ -722,13 +699,13 @@ public class Ivy implements Runnable { * otherwise it returns the IVYBUS property if non null, otherwise it * returns Ivy.DEFAULT_DOMAIN */ - public static String getDomain(String domainbus) { + public static String getDomain(String domainbus) { if ( domainbus == null ) domainbus = System.getProperty("IVYBUS"); if ( domainbus == null ) domainbus = DEFAULT_DOMAIN; return domainbus; } - /** + /** * returns the domain bus * * @since 1.2.8 @@ -744,7 +721,7 @@ public class Ivy implements Runnable { return getDomain(null); } - /** + /** * returns a "wana be unique" ID to make requests on the bus * * @since 1.2.8 @@ -755,7 +732,31 @@ public class Ivy implements Runnable { } private synchronized long nextId() { return current++; } - synchronized void addClient(IvyClient c) { + + /** + * prints a human readable representation of the list of domains + * + * @since 1.2.9 + */ + 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; + } + /////////////////////////////////////////////////////////////////: + // + // Protected methods + // + /////////////////////////////////////////////////////////////////: + + protected IvyClient createIvyClient(Socket s,int port, boolean domachin) throws IOException { + return new IvyClient(this,s,port,domachin); + } + + protected synchronized void addClient(IvyClient c) { if (clients==null||c==null) return; synchronized (clients) { clients.put(c.getClientKey(),c); @@ -763,25 +764,25 @@ public class Ivy implements Runnable { } } - synchronized void removeClient(IvyClient c) { + protected synchronized void removeClient(IvyClient c) { synchronized (clients) { clients.remove(c.getClientKey()); traceDebug("removed "+c+" from clients: "+getClientNames(clients)); } } - void addHalf(IvyClient c) { + protected void addHalf(IvyClient c) { synchronized(half){half.put(c.getClientKey(),c);} traceDebug("added "+c+" in half: "+getClientNames(half)); } - void removeHalf(IvyClient c) { + protected void removeHalf(IvyClient c) { if (half==null||c==null) return; synchronized(half){half.remove(c.getClientKey());} traceDebug("removed "+c+" from half: "+getClientNames(half)); } - boolean shouldIleave(IvyClient ic) { + private boolean shouldIleave(IvyClient ic) { traceDebug("looking for "+ic+" in "+getClientNames(half)+" and "+getClientNames(clients)); IvyClient peer=searchPeer(ic); if (peer==null) return false; @@ -816,7 +817,7 @@ public class Ivy implements Runnable { try { Socket socket = app.accept(); if ((thisThread!=serverThread)||stopped) break; // early disconnexion - new IvyClient(this,socket,0,true); // the peer called me + createIvyClient(socket,0,true); // the peer called me } catch (InterruptedIOException ie) { // traceDebug("server socket was interrupted. good"); if (thisThread!=serverThread) break; @@ -859,17 +860,7 @@ public class Ivy implements Runnable { 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 class Domain { String domainaddr; int port; public Domain(String domainaddr,int port) {this.domainaddr=domainaddr;this.port=port;} @@ -878,7 +869,6 @@ public class Ivy implements Runnable { public int getPort() { return port; } } - // 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); @@ -897,5 +887,3 @@ public class Ivy implements Runnable { } } // class Ivy - -/* EOF */ diff --git a/src/IvyClient.java b/src/IvyClient.java index af42342..b723433 100755 --- a/src/IvyClient.java +++ b/src/IvyClient.java @@ -363,7 +363,7 @@ public class IvyClient implements Runnable { traceDebug("Thread stopped"); } - private synchronized void sendBuffer( String buffer ) throws IvyException { + protected synchronized void sendBuffer( String buffer ) throws IvyException { buffer += "\n"; try { out.write(buffer.getBytes() ); @@ -428,7 +428,7 @@ public class IvyClient implements Runnable { return s; } - private boolean newParseMsg(String s) throws IvyException { + protected boolean newParseMsg(String s) throws IvyException { if (s==null) throw new IvyException("null string to parse in protocol"); byte[] b = s.getBytes(); int from=0,to=0,msgType; diff --git a/src/IvyDaemon.java b/src/IvyDaemon.java index 910a52d..71a2b93 100644 --- a/src/IvyDaemon.java +++ b/src/IvyDaemon.java @@ -42,7 +42,7 @@ public class IvyDaemon implements Runnable { public static int DEFAULT_SERVICE_PORT = 3456 ; public static final String DEFAULTNAME = "IvyDaemon"; - public static final String helpmsg = "usage: java fr.dgac.ivy.IvyDaemon [options]\n\t-b BUS\tspecifies the Ivy bus domain\n\t-p\tport number, default "+DEFAULT_SERVICE_PORT+"\n\t-n ivyname (default "+DEFAULTNAME+")\n\t-q\tquiet, no tty output\n\t-d\tdebug\n\t-h\thelp\nListens on the TCP port, and sends each line read on the Ivy bus. It is useful to launch one Ivy Daemon and let scripts send their message on the bus.\n"; + public static final String helpmsg = "usage: java fr.dgac.ivy.tools.IvyDaemon [options]\n\t-b BUS\tspecifies the Ivy bus domain\n\t-p\tport number, default "+DEFAULT_SERVICE_PORT+"\n\t-n ivyname (default "+DEFAULTNAME+")\n\t-q\tquiet, no tty output\n\t-d\tdebug\n\t-h\thelp\nListens on the TCP port, and sends each line read on the Ivy bus. It is useful to launch one Ivy Daemon and let scripts send their message on the bus.\n"; private static String name = DEFAULTNAME; public static void main(String[] args) throws IvyException, IOException { diff --git a/src/IvyWatcher.java b/src/IvyWatcher.java index 52edb4b..bf8dc3f 100755 --- a/src/IvyWatcher.java +++ b/src/IvyWatcher.java @@ -187,8 +187,7 @@ class IvyWatcher implements Runnable { if (!alreadyBroadcasted(remotehost.toString(),remotePort)) { traceDebug("no known agent originating from " + remotehost + ":" + remotePort); try { - Socket s = new Socket(remotehost,remotePort); - new IvyClient(bus,s,remotePort,false); + bus.createIvyClient(new Socket(remotehost,remotePort),remotePort,false); } catch ( java.net.ConnectException jnc ) { traceDebug("cannot connect to "+remotehostname+":"+remotePort+", he probably stopped his bus"); } diff --git a/src/Makefile b/src/Makefile index 2d33d08..e6a478f 100644 --- a/src/Makefile +++ b/src/Makefile @@ -3,7 +3,7 @@ include ../java.mk #.SUFFIXES: .java .class #SRCS = *.java SRCS = IvyApplicationAdapter.java IvyApplicationListener.java IvyBindListener.java IvyClient.java IvyException.java Ivy.java IvyMessageListener.java IvyWatcher.java SelfIvyClient.java WaiterClient.java Waiter.java PingCallback.java - TOOLS= IvyDaemon.java Probe.java After.java + TOOLS= IvyDaemon.java Probe.java After.java ProxyMaster.java ProxyClient.java Ghost.java OBJS = $(SRCS:.java=.class) DOCS = ../doc/html/api diff --git a/src/ProxyClient.java b/src/ProxyClient.java new file mode 100644 index 0000000..3f616c5 --- /dev/null +++ b/src/ProxyClient.java @@ -0,0 +1,155 @@ +/** + * ProxyClient: Ivy relay, first attempt + * + * @author Yannick Jestin + * @author <a href="http://www.tls.cena.fr/products/ivy/">http://www.tls.cena.fr/products/ivy/</a> + * + * (c) ENAC + * + * changelog: + * 1.2.12 + */ +package fr.dgac.ivy ; +import java.io.*; +import java.net.*; +import java.util.* ; +import gnu.getopt.Getopt; +import org.apache.regexp.*; + +public class ProxyClient extends Ivy { + + private Socket clientSocket; + private PrintWriter out; + private BufferedReader in; + private boolean isRunning=false; + private static boolean debug = (System.getProperty("IVY_DEBUG")!=null) ; + private volatile Thread clientThread;// volatile to ensure the quick communication + private Hashtable id=new Hashtable(); + private Vector ghosts = new Vector(); + + public static int DEFAULT_SERVICE_PORT = 3456 ; + public static final String DEFAULTNAME = "ProxyClient"; + public static final String helpmsg = "usage: java fr.dgac.ivy.ProxyClient [options] hostname\n\t-b BUS\tspecifies the Ivy bus domain\n\t-p\tport number, default "+DEFAULT_SERVICE_PORT+"\n\t-n ivyname (default "+DEFAULTNAME+")\n\t-q\tquiet, no tty output\n\t-d\tdebug\n\t-h\thelp\ncontacts the ProxyMaster on hostname\n"; + + private static String name = DEFAULTNAME; + public static void main(String[] args) { + Ivy bus; + Getopt opt = new Getopt("ProxyClient",args,"n:b:dqp:h"); + int c; + int servicePort = DEFAULT_SERVICE_PORT; + boolean quiet = false; + String domain=Ivy.getDomain(null); + while ((c = opt.getopt()) != -1) switch (c) { + case 'n': + name=opt.getOptarg(); + break; + case 'b': + domain=opt.getOptarg(); + break; + case 'q': + quiet=true; + break; + case 'd': + Properties sysProp = System.getProperties(); + sysProp.put("IVY_DEBUG","yes"); + break; + case 'p': + String s=""; + try { + servicePort = Integer.parseInt(s=opt.getOptarg()); + } catch (NumberFormatException nfe) { + System.out.println("Invalid port number: " + s ); + System.exit(0); + } + break; + case 'h': + default: + System.out.println(helpmsg); + System.exit(0); + } + String hostname="localhost"; + try { + ProxyClient pc = new ProxyClient(hostname,servicePort); + if (!quiet) System.out.println("broadcasting on "+pc.domains(domain)); + if (!quiet) System.out.println("contacting tcp:"+hostname+":"+servicePort); + pc.start(domain); + System.out.println("client running..."); + } catch (IvyException ie) { + System.out.println("error, can't connect to Ivy"); + ie.printStackTrace(); + System.exit(-1); + } catch (IOException ioe) { + System.out.println("error, can't connect to the proxy master"); + ioe.printStackTrace(); + System.exit(-1); + } + } + + public ProxyClient(String hostname,int servicePort) throws IOException { + super(name,name+" ready",null); + clientSocket = new Socket(hostname,servicePort) ; + in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream())); + out = new PrintWriter(new OutputStreamWriter(clientSocket.getOutputStream())); + clientThread=new Thread(new Servicer()); + clientThread.start(); + } + + RE getId=new RE("^ID id=(.*) value=(.*)"); + RE fwd=new RE("^Forward id=(.*) buffer=(.*)"); + void parseMsg(String msg) { + System.out.println("parsing "+msg); + if (getId.match(msg)) { + id.put(getId.getParen(1),getId.getParen(2)); + } else if (fwd.match(msg)) { + // received a forward request + // TODO give it to the puppet + } else { + System.out.println("unknown message "+msg); + } + } + + class Servicer implements Runnable { + public void run() { + Thread thisThread = Thread.currentThread(); + traceDebug("Thread started"); + String msg = null; + try { + while (true) { + msg=in.readLine(); + if (msg==null) break; + parseMsg(msg); + } + } catch (IOException ioe) { + traceDebug("Subreader exception ..."); + ioe.printStackTrace(); + System.exit(0); + } + traceDebug("Subreader Thread stopped"); + System.out.println("ProxyClient disconnected"); + } + } + + protected IvyClient createIvyClient(Socket s,int port, boolean domachin) throws IOException { + String key = getWBUId(); + System.out.println("asking for a key"); + out.println("GetID id="+key); // asks a centralized ID from ProxyMaster + out.flush(); + try { // waits for the answer + while (id.get(key)==null) { Thread.sleep(200); } + } catch (InterruptedException ie) { + ie.printStackTrace(); + } + System.out.println("key received"); + return new Ghost(this,s,port,domachin,(String)id.get(key),this); + } + + private void forward(String id,String buffer) { + out.println("Forward id="+id+" buffer="+buffer); + out.flush(); + } + + private static void traceDebug(String s){ + if (debug) System.out.println("-->ProxyClient "+name+"<-- "+s); + } + +} diff --git a/src/ProxyMaster.java b/src/ProxyMaster.java new file mode 100644 index 0000000..7ed9460 --- /dev/null +++ b/src/ProxyMaster.java @@ -0,0 +1,140 @@ +/** + * ProxyMaster: Ivy relay, first attempt + * + * @author Yannick Jestin + * @author <a href="http://www.tls.cena.fr/products/ivy/">http://www.tls.cena.fr/products/ivy/</a> + * + * (c) ENAC + * + * changelog: + * 1.2.12 + */ +package fr.dgac.ivy ; +import java.io.*; +import java.net.*; +import java.util.* ; +import gnu.getopt.Getopt; +import org.apache.regexp.*; + +public class ProxyMaster { + + private ServerSocket serviceSocket; + private boolean isRunning=false; + private static boolean debug=false; + private Vector proxyClients = new Vector(); + private static int serial=0; + + public static int DEFAULT_SERVICE_PORT = 3456 ; + public static final String DEFAULTNAME = "ProxyMaster"; + public static final String helpmsg = "usage: java fr.dgac.ivy.ProxyMaster [options]\n\t-p\tport number, default "+DEFAULT_SERVICE_PORT+"\n\t-q\tquiet, no tty output\n\t-d\tdebug\n\t-h\thelp\nListens on the TCP port for ProxyClients to join.\n"; + + private static String name = DEFAULTNAME; + public static void main(String[] args) { + Ivy bus; + Getopt opt = new Getopt("ProxyMaster",args,"dqp:h"); + int c; + int servicePort = DEFAULT_SERVICE_PORT; + boolean quiet = false; + while ((c = opt.getopt()) != -1) switch (c) { + case 'q': + quiet=true; + break; + case 'd': + Properties sysProp = System.getProperties(); + sysProp.put("IVY_DEBUG","yes"); + break; + case 'p': + String s=""; + try { + servicePort = Integer.parseInt(s=opt.getOptarg()); + } catch (NumberFormatException nfe) { + System.out.println("Invalid port number: " + s ); + System.exit(0); + } + break; + case 'h': + default: + System.out.println(helpmsg); + System.exit(0); + } + try { + if (!quiet) System.out.println("listening on "+servicePort); + ProxyMaster pm = new ProxyMaster(servicePort); + } catch (IOException ioe) { + System.out.println("error, can't set up the proxy master"); + ioe.printStackTrace(); + System.exit(-1); + } + } + + public ProxyMaster(int servicePort) throws IOException { + serviceSocket = new ServerSocket(servicePort) ; + while ( true ) { + try { + new SubReader(serviceSocket.accept()); + } catch( IOException e ) { + traceDebug("TCP socket reader caught an exception " + e.getMessage()); + } + } + } + + class SubReader extends Thread { + BufferedReader in; + PrintWriter out; + SubReader(Socket socket) throws IOException { + proxyClients.addElement(this); + System.out.println("ProxyClient connected"); + in=new BufferedReader(new InputStreamReader(socket.getInputStream())); + out=new PrintWriter(new OutputStreamWriter(socket.getOutputStream())); + start(); + } + public void send(String s) { + out.println(s); + out.flush(); + } + public void run() { + traceDebug("Subreader Thread started"); + String msg = null; + try { + while (true) { + msg=in.readLine(); + if (msg==null) break; + parseMsg(msg); + } + } catch (IOException ioe) { + traceDebug("Subreader exception ..."); + ioe.printStackTrace(); + System.exit(0); + } + traceDebug("Subreader Thread stopped"); + System.out.println("ProxyClient disconnected"); + proxyClients.removeElement(this); + } + + RE getId=new RE("^GetID id=(.*)"); + RE fwd=new RE("^Forward id=(.*) buffer=(.*)"); + void parseMsg(String msg) { + System.out.println("parsing "+msg); + if (getId.match(msg)) { + // a new Ghost has appeared + System.out.println("a new Ghost has appeared"); + int newGhostId = serial++; + out.println("ID id="+getId.getParen(1)+" value="+newGhostId); + out.flush(); + // TODO create Puppets in each other ProxyClient + for (Enumeration e=proxyClients.elements();e.hasMoreElements();) + ((SubReader)e.nextElement()).send("CreatePuppet id="+newGhostId); + } else if (fwd.match(msg)) { + System.out.println("forwarding "+msg); + // TODO forward the message to all relevant puppets + } else { + System.out.println("error unknown message "+msg); + } + } + } // class SubReader + + private static void traceDebug(String s){ + if (debug) System.out.println("-->ProxyMaster "+name+"<-- "+s); + } + +} |