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       watchlock.release();
 200     }
 201     return true;
 202   }
 203   /**
 204    * Events raised by IClient and propagated to others in session
 205    */
 206
 207   /**
 208    * number of milliseconds between any file state check.
 209    */
 210   long POLL_UNIT = 20;
 211   protected void wait(int u) {
 212     if (u<=0)
 213       u=1;
 214     long l = System.currentTimeMillis()+POLL_UNIT*u;
 215       while (System.currentTimeMillis()<l)
 216         ;
 217   }
 218
 219
 220   private boolean block_document_updates=false;
 221   int STORE_WAIT=5; // how many units before we decide all clients have finalized their appdatas
 222   private boolean in_want_to_store_phase=false;
 223   /**
 224    * client App requests offline storage of vamsas data.
 225    * Call blocks whilst other apps do any appData finalizing
 226    * and then returns (after locking the vamsasDocument in the session)
 227    * Note - the calling app may also receive events through the EventGeneratorThread for document updates.
 228    *
 229    * @return Lock for session.vamArchive
 230    * @param STORE_WAIT indicates how lock the call will block for when nothing appears to be happening to the session.
 231    */
 232   protected Lock want_to_store() {
 233     if (in_want_to_store_phase) {
 234       log.error("client error: want_to_store called again before first call has completed.");
 235       return null;
 236     }
 237     in_want_to_store_phase=true;
 238     // TODO: test the storeDocumentRequest mechanism
 239     /*/ watchThread.haltWatchers();
 240      */
 241     log.debug("Stopping document_update watcher");
 242     vamsasfile.haltWatch();
 243     // block_document_updates=true;
 244     log.debug("Cleared flag for ignoring document_update requests");
 245
 246     log.debug("Sending Store Document Request");
 247     try {
 248       session.addStoreDocumentRequest(client.getClientHandle(), client.getUserHandle());
 249     } catch (Exception e) {
 250       log.warn("Whilst writing StoreDocumentRequest for "+client.getClientHandle().getClientUrn()+" "+client.getUserHandle(),
 251           e);
 252       log.info("trying to continue after storeDocumentRequest exception.");
 253     }
 254     log.debug("Waiting for other apps to do FinalizeApp handling.");
 255     // LATER: refine this semaphore process
 256     // to make a robust signalling mechanism:
 257     // app1 requests, app1..n do something (or don't - they may be dead),
 258     // app1 realises all apps have done their thing, it then continues with synchronized data.
 259     // this probably needs two files - a request file,
 260     //  and a response file which is acknowledged by the app1 requestor for each app.
 261     //  eventually, no more responses are received for the request, and the app can then only continue with its store.
 262     FileWatcher sfwatcher=session.getStoreWatcher();
 263     FileWatcher vfwatcher=session.getDocWatcher();
 264     int units = 0; // zero if updates occured over a sleep period
 265     while (units<STORE_WAIT) {
 266       try {
 267       Thread.sleep(watchThread.WATCH_SLEEP);
 268       } catch (InterruptedException e) {
 269         log.debug("interrupted.");
 270       }
 271       if (sfwatcher.hasChanged() || vfwatcher.hasChanged()) {
 272         units=0;
 273       } else {
 274          units++;
 275       }
 276     }
 277
 278     block_document_updates=false;
 279     vamsasfile.enableWatch();
 280     log.debug("Cleared flag for ignoring document_update requests");
 281     // wait around again (until our own watcher has woken up and synchronized).
 282     while (units<STORE_WAIT) {
 283       try {
 284         Thread.sleep(watchThread.WATCH_SLEEP);
 285         } catch (InterruptedException e) {
 286           log.debug("interrupted.");
 287         }
 288       if (sfwatcher.hasChanged() || vfwatcher.hasChanged())
 289         units=0;
 290       else
 291         units++;
 292     }
 293
 294
 295     log.debug("finished waiting.");
 296     in_want_to_store_phase=false;
 297     return session.vamArchive.getLock();
 298   }
 299   /**
 300    * count handlers for a particular vamsas event
 301    * @param event string enumeration from uk.ac.vamsas.client.Events
 302    * @return -1 for an invalid event, otherwise the number of handlers
 303    */
 304   protected int countHandlersFor(String event) {
 305     if (handlers.containsKey(event)) {
 306       PropertyChangeSupport handler = (PropertyChangeSupport) handlers.get(event);
 307       PropertyChangeListener[] listeners;
 308       if (handler!=null)
 309         return ((listeners=handler.getPropertyChangeListeners())==null)
 310         ? -1 : listeners.length;
 311     }
 312     return -1;
 313   }
 314
 315   public void disableDocumentWatch() {
 316     vamsasfile.haltWatch();
 317   }
 318
 319   public boolean isDocumentWatchEnabled() {
 320     return (vamsasfile!=null) && vamsasfile.isWatchEnabled();
 321   }
 322
 323   public void enableDocumentWatch() {
 324     vamsasfile.enableWatch();
 325   }
 326
 327   public boolean isWatcherAlive() {
 328     return watchThread!=null && watchThread.running && watchThread.isAlive();
 329   }
 330
 331   public void interruptWatching() {
 332     if (watchThread!=null && watchThread.isAlive())
 333     {
 334       // TODO: find a way of interrupting watcher in a way that prevents file IO being interrupted
 335       watchThread.interrupt();
 336     }
 337
 338   }
 339   /**
 340    * called to start the session watching thread which generates events
 341    */
 342   public void startWatching() {
 343     enableDocumentWatch();
 344     watchThread.start();
 345     while (!watchThread.running && watchThread.isAlive())
 346       log.debug("Waiting until watcher is really started.");
 347   }
 348
 349   public void stopWatching() {
 350     interruptWatching();
 351     watchThread.haltWatchers();
 352
 353   }
 354
 355
 356 }