ProxyAgent


1   /*
2    * Copyright (C) 2001 - 2005 ScalAgent Distributed Technologies
3    * Copyright (C) 1996 - 2000 BULL
4    * Copyright (C) 1996 - 2000 INRIA
5    *
6    * This library is free software; you can redistribute it and/or
7    * modify it under the terms of the GNU Lesser General Public
8    * License as published by the Free Software Foundation; either
9    * version 2.1 of the License, or any later version.
10   * 
11   * This library is distributed in the hope that it will be useful,
12   * but WITHOUT ANY WARRANTY; without even the implied warranty of
13   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14   * Lesser General Public License for more details.
15   * 
16   * You should have received a copy of the GNU Lesser General Public
17   * License along with this library; if not, write to the Free Software
18   * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
19   * USA.
20   *
21   * The present code contributor is ScalAgent Distributed Technologies.
22   */
23  package fr.dyade.aaa.agent;
24  
25  import java.io.*;
26  import java.util.Hashtable  ;
27  import java.util.Enumeration  ;
28  
29  import org.objectweb.util.monolog.api.BasicLevel;
30  
31  import fr.dyade.aaa.util.*;
32  
33  public abstract class ProxyAgent extends Agent {
34  
35    public static final int DRIVER_IN = 1;
36    public static final int DRIVER_OUT = 2;
37  
38    /** true if connect may block */
39    protected boolean blockingCnx = true;
40    /** true if proxy may handle multiple connection */
41    protected boolean multipleCnx = false;
42    /** flow control in driver in */
43    protected int inFlowControl = 20;
44    /** communication with drvOut */
45    protected transient Queue qout;
46  
47    /** manage input stream */
48    private transient DriverIn drvIn = null;
49    /** manage output stream */
50    private transient DriverOut drvOut = null;
51    /** manage connection step, optional */
52    private transient DriverConnect drvCnx;
53  
54    /** <code>true</code> if the proxy manages multiple connections. */
55    protected boolean multiConn = false;
56    /** 
57     * Table holding the <code>DriverMonitor</code> objects, each one holding a
58     * connection set (a pair of drivers, a qout, ois, oos, ...).
59     * For multi-connections management.
60     *
61     * @see  DriverMonitor
62     */
63    protected transient Hashtable   driversTable;
64    /**
65     * Used in multi-connections context for identifying each
66     * connection.
67     */
68    private int driversKey = 1;
69  
70    /** <code>true</code> if the proxy is being finalized. */
71    boolean finalizing = false;
72  
73    /**
74     * Returns default log topic for proxies. Its method  overriddes
75     * the default one in Agent, the resulting logging topic is
76     * <code>Debug.A3Proxy</code> dot the real classname.
77     */
78    protected String   getLogTopic() {
79      String   classname = getClass().getName();
80      return Debug.A3Proxy + '.' +
81        classname.substring(classname.lastIndexOf('.') +1);
82    }
83  
84    public ProxyAgent() {
85      this(null);
86    }
87  
88    public ProxyAgent(String   n) {
89      this(AgentServer.getServerId(), n);
90    }
91  
92    public ProxyAgent(short to, String   n) {
93      // ProxyAgent is "pined" in memory.
94      super(to, n, true);
95      drvIn = null;
96      drvOut = null;
97      drvCnx = null;
98    }
99  
100   /*
101    * Constructor used to build Well Known Services agents.
102    *
103    * @param name    symbolic name
104    * @param stamp   well known stamp
105    */
106   public ProxyAgent(String   name, int stamp) {
107     super(name, true, stamp);
108     drvIn = null;
109     drvOut = null;
110     drvCnx = null;
111   }
112 
113   /**
114    * Provides a string image for this object.
115    *
116    * @return    printable image of this object
117    */
118   public String   toString() {
119     return "(" + super.toString() +
120       ",blockingCnx=" + blockingCnx +
121       ",multipleCnx=" + multipleCnx +
122       ",inFlowControl=" + inFlowControl +
123       ",multiConn=" + multiConn +
124       ",qout=" + qout +
125       ",driversKey=" + driversKey + ")";
126   }
127   
128   /**
129    * Method setting the <code>ProxyAgent</code> in multiConn mode.
130    * To be called immediately after the <code>ProxyAgent</code> instanciation.
131    */
132   public void setMultiConn() {
133     multiConn = true;
134   }
135 
136   /**
137    * Initializes the transient members of this agent.
138    * This function is first called by the factory agent, then by the system
139    * each time the agent server is restarted.
140    * <p>
141    * This function is not declared final so that derived classes
142    * may change their reload policy.
143    *
144    * @param firstTime       true when first called by the factory
145    *
146    * @exception Exception
147    *    unspecialized exception
148    */
149   protected void agentInitialize(boolean firstTime) throws Exception   {
150     super.agentInitialize(firstTime);
151 
152     // In single connection mode, creating qout once:
153     if (!multiConn)
154       qout = new Queue();
155     // In multi connections mode, creating the driversTable:
156     else
157       driversTable = new Hashtable  ();
158 
159     reinitialize();
160   }
161 
162   /**
163    * Reinitializes the agent, that is reconnects its input and output.
164    * This function may be called only when all drivers are null
165    * if the <code>ProxyAgent</code> manages only one connection at a time.
166    * Otherwise, a multiConn <code>ProxyAgent</code> will reinitialize
167    * even if the current drivers are not null.
168    */
169   protected void reinitialize() throws IOException {
170     if (drvIn != null || drvOut != null) {
171       if (!multiConn) throw new IllegalStateException  ();
172     }
173 
174     drvCnx = new DriverConnect(this, blockingCnx, multipleCnx);
175     drvCnx.start();
176 
177     // If the ProxyAgent manages multi-connections, storing the connection set
178     // in a DriverMonitor and putting it in the driversTable.
179     if (multiConn) {
180       DriverMonitor dMonitor = new DriverMonitor(drvIn, drvOut, qout, ois, oos,
181                                                  drvCnx);
182 
183       driversTable.put(new Integer  (driversKey), dMonitor); 
184       driversKey++;
185     }
186   }
187 
188   /** input stream, created by subclass during connect */
189   protected transient NotificationInputStream ois = null;
190   /** output stream, created by subclass during connect */
191   protected transient NotificationOutputStream oos = null;
192 
193   /**
194    * Initializes the connection with the outside, up to creating
195    * the input and output streams <code>ois</code> and <code>oos</code>.
196    *
197    * @exception Exception
198    *    unspecialized exception
199    */
200   public abstract void connect() throws Exception  ;
201 
202   /**
203    * Closes the connection with the outside.
204    *
205    * @exception Exception
206    *    unspecialized exception
207    */
208   public abstract void disconnect() throws Exception  ;
209 
210   /**
211    * Connects the streams provided by the user to this proxy agent
212    * to two created drivers. The streams must be created by the
213    * <code>connect</code> function defined in derived classes.
214    * <p>
215    * If the connection step blocks, this function is executed
216    * in a separate thread controled by <code>drvCnx</code> (see
217    * <code>Initialize</code>).
218    * 
219    * @exception java.net.SocketException  If the server socket is being closed.
220    */
221   void createDrivers() throws Exception   {
222     drvCnx.canStop = true;
223     try {
224       connect();
225     } catch (InterruptedException   exc) {
226       logmon.log(BasicLevel.DEBUG, getName() + "InterruptedException");
227     } finally {
228       if (drvCnx != null)
229         drvCnx.canStop = false;
230     }
231 
232     if ((drvCnx == null) || (! drvCnx.isRunning)) return;
233 
234     if (! multiConn) {  
235       if (logmon.isLoggable(BasicLevel.DEBUG))
236         logmon.log(BasicLevel.DEBUG, getName() + ", connected");
237       if (oos != null) {
238         drvOut = new DriverOut(DRIVER_OUT, this, qout, oos);
239         drvOut.start();
240       }
241       if (ois != null) {
242         drvIn = new DriverIn(DRIVER_IN, this, ois, inFlowControl);
243         drvIn.start();
244       }
245     }
246     // If the ProxyAgent is multiConn, creating drvIn and drvOut with
247     // the additionnal driversKey parameter and also creating a new qout.
248     else {
249       if (logmon.isLoggable(BasicLevel.DEBUG))
250         logmon.log(BasicLevel.DEBUG, "connected - driversKey=" + driversKey);
251       if (ois != null) {
252         drvIn = new DriverIn(DRIVER_IN, this, ois, inFlowControl, driversKey);
253         drvIn.start();
254       }
255       if (oos != null) {
256         qout = new Queue(); 
257         drvOut = new DriverOut(DRIVER_OUT, this, qout, oos, driversKey);
258         drvOut.start();
259       }
260     }
261   }
262 
263   /**
264    * Stops all drivers (non multiConn mode).
265    * May be multiply called.
266    */
267   protected void stop() {
268     if (multiConn) return;
269 
270     if (drvCnx != null) {
271       drvCnx.stop();
272       drvCnx = null;
273     }
274     if (drvIn != null) {
275       drvIn.stop();
276       drvIn = null;
277     }
278     if (drvOut != null) {
279       drvOut.stop();
280       drvOut = null;
281     }
282     if (logmon.isLoggable(BasicLevel.DEBUG))
283       logmon.log(BasicLevel.DEBUG, "stopped");
284   }
285 
286   /**
287    * Method stopping the specified connection set (multi-connections mode). 
288    *
289    * @param drvKey  key identifying the connection set to stop.
290    */
291   protected void stop(int drvKey) {
292     DriverMonitor dMonitor =
293       (DriverMonitor) driversTable.get(new Integer  (drvKey));
294 
295     if (dMonitor != null) {
296       if (dMonitor.drvCnx != null) {
297         (dMonitor.drvCnx).stop();
298         dMonitor.drvCnx = null;
299       }
300       if (dMonitor.drvIn != null) {
301         (dMonitor.drvIn).stop();
302         dMonitor.drvIn = null;
303       }
304       if (dMonitor.drvOut != null) {
305         (dMonitor.drvOut).stop();
306         dMonitor.drvOut = null;
307       }
308     if (logmon.isLoggable(BasicLevel.DEBUG))
309       logmon.log(BasicLevel.DEBUG, "stopped - driversKey=" + driversKey);
310     }
311   }
312 
313   /** Method cleaning DriverOut. Single connection mode only. */
314   public void cleanDriverOut() {
315     if (! multiConn) {
316       if (drvOut != null)
317         drvOut.clean();
318     }
319   }
320 
321   /**
322    * Method cleaning the <code>DriverOut</code> specified
323    * by the key parameter (multi-connections mode).
324    *
325    * @param  drvKey key identifying the connection set.
326    */ 
327   public void cleanDriverOut(int drvKey) {
328     DriverMonitor dMonitor =
329       (DriverMonitor) driversTable.get(new Integer  (drvKey));
330     if (dMonitor != null) {
331       if (dMonitor.drvOut != null)
332         (dMonitor.drvOut).clean();
333     }
334   }
335 
336   /** Closes all the connections. */
337   protected void closeAllConnections() {
338     Enumeration   keys = driversTable.keys();
339     while (keys.hasMoreElements()) {
340       Integer   key = (Integer  ) keys.nextElement();
341 
342       DriverMonitor dMonitor = (DriverMonitor) driversTable.get(key);
343       if (dMonitor != null) {
344         if (dMonitor.ois != null) {
345           try {
346            (dMonitor.ois).close();
347           } catch (IOException exc) {}
348           dMonitor.ois = null;
349         }
350         if (dMonitor.oos != null) {
351           try {
352            (dMonitor.oos).close();
353           } catch (IOException exc) {}
354           dMonitor.oos = null;
355           dMonitor.qout = null;
356         }
357         stop(key.intValue());
358       }
359     }
360     driversTable.clear();
361   }
362 
363   /** 
364    * Method called by the ProxyAgent <code>DriverIn</code> instances to
365    * forward the notifications they got from their input streams. 
366    * <p>
367    * May be overridden for specific behaviour as long as the proxy state
368    * is not modified by the method, because it does not occur within a
369    * transaction.
370    *
371    * @param key  Driver identifier.
372    * @param not  Notification to forward.
373    */ 
374   protected void driverReact(int key, Notification not) {
375     if (logmon.isLoggable(BasicLevel.DEBUG))
376       logmon.log(BasicLevel.DEBUG, "Proxy " + this + " gets not " + not 
377                  + " from driver " + key);
378 
379     sendTo(this.getId(), not);
380   }
381 
382   /**
383    * Method called by subclasses to directly send their notifications
384    * to the right <code>DriverOut</code>.
385    *
386    * @param key  Driver identifier.
387    * @param not  Notification to send out.
388    * @exception Exception  If the driver to pass the notification to can't
389    *              be retrieved from the key parameter.
390    */
391   protected void sendOut(int key, Notification not) throws Exception   {
392     if (logmon.isLoggable(BasicLevel.DEBUG))
393       logmon.log(BasicLevel.DEBUG, "Proxy " + this + " gets not " + not 
394                  + " to pass to driver " + key);
395     try {
396       DriverMonitor dMon = (DriverMonitor) driversTable.get(new Integer  (key));
397       (dMon.getQout()).push(not);
398     }
399     catch (Exception   e ) {
400       throw new Exception  ("Can't forward notification " + not
401                            + " to driver out " + key + ": " + e);
402     }
403   }
404     
405   /**
406    * Method implementing the <code>ProxyAgent</code> reactions to
407    * notifications.
408    * Forwards notifications coming from an identified agent onto the outgoing
409    * connection.
410    *
411    * @param from    agent sending notification
412    * @param not     notification to react to
413    *
414    * @exception Exception
415    *    unspecialized exception
416    */
417   public void react(AgentId from, Notification not) throws Exception   {
418     int drvKey;
419     DriverMonitor dMon;
420     DriverIn dIn;
421     Queue qo;
422     try {
423       if (not instanceof DriverDone)
424         driverDone((DriverDone) not);
425       else if (not instanceof FlowControlNot) {
426         // Allowing drvIn to read more data
427         drvKey = ((FlowControlNot) not).driverKey;
428         // MultiConn proxy: getting the right driverIn to control:
429         if (drvKey != 0) {
430           dMon = (DriverMonitor) driversTable.get(new Integer  (drvKey));
431           dIn = dMon.drvIn;
432           dIn.recvFlowControl((FlowControlNot) not);
433         }
434         else
435           drvIn.recvFlowControl((FlowControlNot) not);
436       }
437       else if (not instanceof DeleteNot) {
438         closeAllConnections();
439         super.react(from, not);
440       } 
441       // If notification comes from an identified agent:
442       else if (! from.equals(this.getId()))
443         qout.push(not);
444       else
445         super.react(from, not);
446     } catch (Exception   exc) {
447       if (logmon.isLoggable(BasicLevel.ERROR))
448         logmon.log(BasicLevel.ERROR,
449                    "error in " + this + ".react(" + from + ", " + not + ")",
450                    exc);
451       stop();
452       // the proxy agent may restart
453     }
454   }
455 
456   /**
457    * Reacts to end of driver execution.
458    * <p>
459    * This is the end of the driver thread, however the thread resources
460    * may not have been released. This is why <code>close</code> is called
461    * on the notification streams, which requires from the stream classes
462    * to cope with a call to <code>close</code> when some resources may have
463    * been released.
464    */
465   protected void driverDone(DriverDone not) throws IOException { 
466     if (!multiConn) {
467       switch (not.getDriver()) {
468         case DRIVER_IN:
469           try {
470             ois.close();
471           } catch (Exception   e) {}
472           ois = null;
473           drvIn = null;
474           break;
475         case DRIVER_OUT:
476           try {
477             oos.close();
478           } catch (Exception   e) {}
479           oos = null;
480           drvOut = null;
481           break;
482       }
483     }
484     // In case of multiConn, the driver to close is identified in the
485     // DriverDone notification.
486     else {
487       int drvKey = not.getDriverKey();
488       DriverMonitor dMonitor = (DriverMonitor)
489         driversTable.get(new Integer  (drvKey));
490 
491       if (dMonitor != null) {
492         switch (not.getDriver()) {
493           case DRIVER_IN:
494         if (dMonitor.drvIn != null)
495           (dMonitor.drvIn).close();
496         dMonitor.ois = null;
497         dMonitor.drvIn = null;
498             break;
499           case DRIVER_OUT:
500         if (dMonitor.drvOut != null)
501           (dMonitor.drvOut).close();
502         dMonitor.oos = null;
503         dMonitor.drvOut = null;
504             break;
505         }
506         // When both drivers have been closed, removing the entry
507         // corresponding to their pair from the driversTable.
508         if (dMonitor.drvIn == null && dMonitor.drvOut == null)
509           driversTable.remove(new Integer  (drvKey));
510       }
511     } 
512   }
513 
514   /**
515    * Finalizes this proxy agent execution. Calls <code>disconnect</code> to
516    * close the open streams, and <code>stop</code> to stop the drivers.
517    *
518    * @param lastTime    true when last called by the factory on agent deletion.
519    */
520   public void agentFinalize(boolean lastime) {
521     if (logmon.isLoggable(BasicLevel.DEBUG))
522       logmon.log(BasicLevel.DEBUG,
523                  toString() + " agentFinalize -> " + drvCnx);
524 
525     finalizing = true;
526 
527     if (multiConn)
528       closeAllConnections();
529     else {
530       try {
531         ois.close();
532       } catch (Exception   exc) {}
533       try {
534         oos.close();
535       } catch (Exception   exc) {}
536     }
537 
538     try {
539       disconnect();
540     } catch (Exception   exc) {}
541 
542     stop();
543 
544     qout = null;
545     ois = null;
546     oos = null;
547   }
548 }
549
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags