src/uk/ac/vamsas/client/simpleclient/EventGeneratorThread.java

   1 package uk.ac.vamsas.client.simpleclient;
   2
   3 import java.beans.PropertyChangeListener;
   4 import java.beans.PropertyChangeSupport;
   5 import java.util.Hashtable;
   6
   7 import org.apache.commons.logging.Log;
   8 import org.apache.commons.logging.LogFactory;
   9
  10 import uk.ac.vamsas.client.Events;
  11
  12 /**
  13  * monitors watcher objects and generates events.
  14  */
  15 public class EventGeneratorThread {
  16   private static Log log = LogFactory.getLog(EventGeneratorThread.class);
  17   private SimpleClient client;
  18   private Hashtable handlers; // manager object
  19   private VamsasSession session;
  20   /**
  21    * thread watching all the session's file objects
  22    */
  23   protected VamsasFileWatcherThread watchThread=null;
  24   /**
  25    * Watcher element for list of all the clientHandles for the session
  26    */
  27   protected SessionFileWatcherElement clientfile=null;
  28   /**
  29    * the session's vamsasDocument
  30    */
  31   protected VamsasFileWatcherElement vamsasfile=null;
  32   /**
  33    * written to by client when its app calls storeDocument.
  34    */
  35   protected SessionFileWatcherElement storeFile=null;
  36
  37   EventGeneratorThread(VamsasSession s, SimpleClient _client, Hashtable eventhandlers) {
  38     if (eventhandlers==null || s==null || _client==null)
  39       throw new Error("Null arguments to EventGeneratorThread constructor.");
  40     handlers = eventhandlers;
  41     session = s;
  42     client = _client;
  43     log.debug("Creating VamsasFileWatcherThread.");
  44     watchThread = new VamsasFileWatcherThread(this);
  45     initWatchers();
  46   }
  47
  48   private void initWatchers() {
  49     if (clientfile==null) {
  50       log.debug("Initializing clientfile Watcher");
  51       clientfile = session.getClientWatcherElement();
  52      // handler is set in the Vamsas session
  53 /* clientfile.setHandler(new WatcherCallBack() {
  54
  55             public boolean handleWatchEvent(WatcherElement watcher, Lock lock) {
  56               return clientListChanged(watcher, lock);
  57             }
  58       });*/
  59       watchThread.addElement(clientfile);
  60     }
  61     final EventGeneratorThread evgen=this;
  62
  63     if (vamsasfile ==null) {
  64       log.debug("Initializing VamsasFileWatcher");
  65       vamsasfile = new VamsasFileWatcherElement(session.vamArchive,
  66           new WatcherCallBack() {
  67          public boolean handleWatchEvent(WatcherElement watcher, Lock lock) {
  68            return evgen.documentChanged(lock);
  69          }
  70       });
  71       watchThread.addElement(vamsasfile);
  72     }
  73     if (storeFile == null) {
  74       storeFile = new SessionFileWatcherElement(session.getStoreDocFile(),
  75           new WatcherCallBack() {
  76         public boolean handleWatchEvent(WatcherElement watcher, Lock lock) {
  77           return evgen.storeDocRequest(lock);
  78         }
  79       });
  80       log.debug("Initializing storeDocFile flag watcher");
  81     }
  82     /*
  83     */
  84     log.debug("Watchers inited.");
  85   }
  86   /**
  87    * Call registered handlers for a vamsas session event
  88    * @param handlerEvent a named event
  89    * @param property property name to pass to handler
  90    * @param oldval old value of property to pass
  91    * @param newval new value of property to pass
  92    * @return true if event generation did not raise any exceptions.
  93    */
  94   boolean _raise(String handlerEvent, String property, Object oldval, Object newval) {
  95     PropertyChangeSupport h = (PropertyChangeSupport) handlers.get(handlerEvent);
  96     if (h!=null) {
  97       log.debug("Triggering:"+handlerEvent);
  98       try {
  99         h.firePropertyChange(property, oldval, newval);
 100       } catch (Exception e) {
 101         log.warn("Client Exception during handling of "+handlerEvent, e);
 102         return false;
 103       }
 104       catch (Error e)
 105       {
 106         log.error("Serious! Client Error during handling of "+handlerEvent, e);
 107         return false;
 108       }
 109       log.debug("Finished  :"+handlerEvent);
 110     } else
 111       log.debug("No handlers for raised "+handlerEvent);
 112     return true;
 113   }
 114   protected boolean storeDocRequest(Lock lock) {
 115     if (log.isDebugEnabled())
 116       log.debug("StoreDocRequest on "+(lock==null ? (lock.isLocked() ? "" : "Invalid ") : "Non-")+"Existing lock");
 117     // TODO: define the storeFile semaphore mechanism : file exists - all clients inform their apps, and then the client that wrote the file should delete the file (it should hold the lock to it).
 118     if (storeFile.getWatcher().exists) {
 119       _raise(Events.DOCUMENT_FINALIZEAPPDATA, client.getSessionUrn(), null, client);
 120       // expect client to write to document so update watcher state on return
 121       vamsasfile.getWatcher().setState();
 122       lock.release();
 123     }
 124     return true;
 125   }
 126
 127   protected boolean documentChanged(Lock doclock) {
 128     boolean continueWatching=true;
 129     if (!block_document_updates) {
 130       session.vamArchive.fileLock=doclock;
 131       if (client.pickmanager!=null)
 132         client.pickmanager.setPassThru(false);
 133       if (log.isDebugEnabled()) {
 134         log.debug("Initiating a documentChanged event. Document is "+(client.cdocument==null ? "closed" : "open"));
 135       }
 136       // TODO: decide if individual object update handlers are called as well as overall event handler
 137       if (!_raise(Events.DOCUMENT_UPDATE, client.getSessionUrn(), null, client))
 138       {
 139         log.info("Recovering from errors or exceptions generated by client application");
 140         if (client.cdocument!=null)
 141         {
 142           try {
 143             client.tidyAwaySessionDocumentState();
 144           }
 145           catch (Exception e)
 146           {
 147             log.warn("Exception generated by vamsas library - when tidying away session document:",e);
 148           }
 149           catch (Error e)
 150           {
 151             log.error("LIBRARY Implementation error - when tidying away session document:",e);
 152           }
 153         }
 154
 155       }
 156       if (client.pickmanager!=null)
 157         client.pickmanager.setPassThru(true);
 158       if (log.isDebugEnabled()) {
 159         log.debug("Finished handling a documentChanged event. Document is "+(client.cdocument==null ? "closed" : "open"));
 160       }
 161       if (client.cdocument!=null)
 162       {
 163         log.warn("Implementation Error ?  ClientDocument instance has not been closed or updated by handler!");
 164       }
 165       /*try {
 166         client._session.getVamsasDocument().closeArchive();
 167       } catch (Exception e) {log.warn("Unexpected exception when closing document after update.",e);};
 168       */
 169     } else {
 170       // TODO: check documentChanged */
 171       log.debug("Ignoring documentChanged event for "+client.getSessionUrn());
 172     }
 173     return continueWatching;
 174   }
 175   boolean ownsf = false;
 176   /**
 177    * Moved to SimpleClientSessionManager
 178    * scans all watchers and fires changeEvents if necessary
 179    * @return number of events generated.
 180    */
 181   private boolean clientListChanged(WatcherElement clientfile, Lock watchlock) {
 182     log.debug("ClientListChanged handler called for "+clientfile.getWatcher().getSubject());
 183     // could make this general - but for now keep simple
 184     if (watchlock!=null) {
 185       // TODO: compare new client list to old list version. is it changed ?
 186       // see what happened to the clientfile - compare our internal version with the one in the file, or just send the updated list out...?
 187       //
 188       /**
 189        * Generated when a new vamsas client is attached to a session (Handle is
 190        * passed) Note: the newly created client does not receive the event.
 191        *
 192       public static final String CLIENT_CREATION = "uk.ac.vamsas.client.events.clientCreateEvent";
 193   */ // as the test
 194       /**
 195        * Generated when a vamsas client leaves a session (Handle is passed to all
 196        * others).
 197       public static final String CLIENT_FINALIZATION = "uk.ac.vamsas.client.events.clientFinalizationEvent";
 198        */ // again - as the test.
 199     }
 200     return true;
 201   }
 202   /**
 203    * Events raised by IClient and propagated to others in session
 204    */
 205
 206   /**
 207    * number of milliseconds between any file state check.
 208    */
 209   long POLL_UNIT = 20;
 210   protected void wait(int u) {
 211     if (u<=0)
 212       u=1;
 213     long l = System.currentTimeMillis()+POLL_UNIT*u;
 214       while (System.currentTimeMillis()<l)
 215         ;
 216   }
 217
 218
 219   private boolean block_document_updates=false;
 220   int STORE_WAIT=5; // how many units before we decide all clients have finalized their appdatas
 221   private boolean in_want_to_store_phase=false;
 222   /**
 223    * client App requests offline storage of vamsas data.
 224    * Call blocks whilst other apps do any appData finalizing
 225    * and then returns (after locking the vamsasDocument in the session)
 226    * Note - the calling app may also receive events through the EventGeneratorThread for document updates.
 227    *
 228    * @return Lock for session.vamArchive
 229    * @param STORE_WAIT indicates how lock the call will block for when nothing appears to be happening to the session.
 230    */
 231   protected Lock want_to_store() {
 232     if (in_want_to_store_phase) {
 233       log.error("client error: want_to_store called again before first call has completed.");
 234       return null;
 235     }
 236     in_want_to_store_phase=true;
 237     // TODO: test the storeDocumentRequest mechanism
 238     /*/ watchThread.haltWatchers();
 239      */
 240     log.debug("Stopping document_update watcher");
 241     vamsasfile.haltWatch();
 242     // block_document_updates=true;
 243     log.debug("Cleared flag for ignoring document_update requests");
 244
 245     log.debug("Sending Store Document Request");
 246     try {
 247       session.addStoreDocumentRequest(client.getClientHandle(), client.getUserHandle());
 248     } catch (Exception e) {
 249       log.warn("Whilst writing StoreDocumentRequest for "+client.getClientHandle().getClientUrn()+" "+client.getUserHandle(),
 250           e);
 251       log.info("trying to continue after storeDocumentRequest exception.");
 252     }
 253     log.debug("Waiting for other apps to do FinalizeApp handling.");
 254     // LATER: refine this semaphore process
 255     // to make a robust signalling mechanism:
 256     // app1 requests, app1..n do something (or don't - they may be dead),
 257     // app1 realises all apps have done their thing, it then continues with synchronized data.
 258     // this probably needs two files - a request file,
 259     //  and a response file which is acknowledged by the app1 requestor for each app.
 260     //  eventually, no more responses are received for the request, and the app can then only continue with its store.
 261     FileWatcher sfwatcher=session.getStoreWatcher();
 262     FileWatcher vfwatcher=session.getDocWatcher();
 263     int units = 0; // zero if updates occured over a sleep period
 264     while (units<STORE_WAIT) {
 265       try {
 266       Thread.sleep(watchThread.WATCH_SLEEP);
 267       } catch (InterruptedException e) {
 268         log.debug("interrupted.");
 269       }
 270       if (sfwatcher.hasChanged() || vfwatcher.hasChanged()) {
 271         units=0;
 272       } else {
 273          units++;
 274       }
 275     }
 276
 277     block_document_updates=false;
 278     vamsasfile.enableWatch();
 279     log.debug("Cleared flag for ignoring document_update requests");
 280     // wait around again (until our own watcher has woken up and synchronized).
 281     while (units<STORE_WAIT) {
 282       try {
 283         Thread.sleep(watchThread.WATCH_SLEEP);
 284         } catch (InterruptedException e) {
 285           log.debug("interrupted.");
 286         }
 287       if (sfwatcher.hasChanged() || vfwatcher.hasChanged())
 288         units=0;
 289       else
 290         units++;
 291     }
 292
 293
 294     log.debug("finished waiting.");
 295     in_want_to_store_phase=false;
 296     return session.vamArchive.getLock();
 297   }
 298   /**
 299    * count handlers for a particular vamsas event
 300    * @param event string enumeration from uk.ac.vamsas.client.Events
 301    * @return -1 for an invalid event, otherwise the number of handlers
 302    */
 303   protected int countHandlersFor(String event) {
 304     if (handlers.containsKey(event)) {
 305       PropertyChangeSupport handler = (PropertyChangeSupport) handlers.get(event);
 306       PropertyChangeListener[] listeners;
 307       if (handler!=null)
 308         return ((listeners=handler.getPropertyChangeListeners())==null)
 309         ? -1 : listeners.length;
 310     }
 311     return -1;
 312   }
 313
 314   public void disableDocumentWatch() {
 315     vamsasfile.haltWatch();
 316   }
 317
 318   public boolean isDocumentWatchEnabled() {
 319     return (vamsasfile!=null) && vamsasfile.isWatchEnabled();
 320   }
 321
 322   public void enableDocumentWatch() {
 323     vamsasfile.enableWatch();
 324   }
 325
 326   public boolean isWatcherAlive() {
 327     return watchThread!=null && watchThread.running && watchThread.isAlive();
 328   }
 329
 330   public void interruptWatching() {
 331     if (watchThread!=null && watchThread.isAlive())
 332       watchThread.interrupt();
 333
 334   }
 335   /**
 336    * called to start the session watching thread which generates events
 337    */
 338   public void startWatching() {
 339     enableDocumentWatch();
 340     watchThread.start();
 341     while (!watchThread.running && watchThread.isAlive())
 342       log.debug("Waiting until watcher is really started.");
 343   }
 344
 345   public void stopWatching() {
 346     interruptWatching();
 347     watchThread.haltWatchers();
 348
 349   }
 350
 351
 352 }