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       clientfile.setHandler(new WatcherCallBack() {
  53
  54             public boolean handleWatchEvent(WatcherElement watcher, Lock lock) {
  55               // TODO Auto-generated method stub
  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   void _raise(String handlerEvent, String property, Object oldval, Object newval) {
  87     PropertyChangeSupport h = (PropertyChangeSupport) handlers.get(handlerEvent);
  88     if (h!=null) {
  89       log.debug("Triggering:"+handlerEvent);
  90       h.firePropertyChange(property, oldval, newval);
  91       log.debug("Finished  :"+handlerEvent);
  92     } else
  93       log.debug("No handlers for raised "+handlerEvent);
  94   }
  95   protected boolean storeDocRequest(Lock lock) {
  96     if (log.isDebugEnabled())
  97       log.debug("StoreDocRequest on "+(lock==null ? (lock.isLocked() ? "" : "Invalid ") : "Non-")+"Existing lock");
  98     // 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).
  99     if (storeFile.getWatcher().exists) {
 100       _raise(Events.DOCUMENT_FINALIZEAPPDATA, client.getSessionUrn(), null, client);
 101       // expect client to write to document so update watcher state on return
 102       vamsasfile.getWatcher().setState();
 103       lock.release();
 104     }
 105     return true;
 106   }
 107
 108   protected boolean documentChanged(Lock doclock) {
 109     boolean continueWatching=true;
 110     if (!block_document_updates) {
 111       session.vamArchive.fileLock=doclock;
 112       // TODO: decide if individual object update handlers are called as well as overall event handler
 113       _raise(Events.DOCUMENT_UPDATE, client.getSessionUrn(), null, client);
 114       if (log.isDebugEnabled()) {
 115         log.debug("Finished handling a documentChanged event. Document is "+(client.cdocument==null ? "closed" : "open"));
 116       }
 117       /*try {
 118         client._session.getVamsasDocument().closeArchive();
 119       } catch (Exception e) {log.warn("Unexpected exception when closing document after update.",e);};
 120       */
 121     } else {
 122       // TODO: check documentChanged */
 123       log.debug("Ignoring documentChanged event for "+client.getSessionUrn());
 124     }
 125     return continueWatching;
 126   }
 127   boolean ownsf = false;
 128   /**
 129    * scans all watchers and fires changeEvents if necessary
 130    * @return number of events generated.
 131    */
 132   private boolean clientListChanged(WatcherElement clientfile, Lock watchlock) {
 133     log.debug("ClientListChanged handler called for "+clientfile.getWatcher().getSubject());
 134     // could make this general - but for now keep simple
 135     if (watchlock!=null) {
 136       // TODO: compare new client list to old list version. is it changed ?
 137       // see what happened to the clientfile - compare our internal version with the one in the file, or just send the updated list out...?
 138       //
 139       /**
 140        * Generated when a new vamsas client is attached to a session (Handle is
 141        * passed) Note: the newly created client does not receive the event.
 142        *
 143       public static final String CLIENT_CREATION = "uk.ac.vamsas.client.events.clientCreateEvent";
 144   */ // as the test
 145       /**
 146        * Generated when a vamsas client leaves a session (Handle is passed to all
 147        * others).
 148       public static final String CLIENT_FINALIZATION = "uk.ac.vamsas.client.events.clientFinalizationEvent";
 149        */ // again - as the test.
 150     }
 151     return true;
 152   }
 153   /**
 154    * Events raised by IClient and propagated to others in session
 155    */
 156
 157   /**
 158    * number of milliseconds between any file state check.
 159    */
 160   long POLL_UNIT = 20;
 161   protected void wait(int u) {
 162     if (u<=0)
 163       u=1;
 164     long l = System.currentTimeMillis()+POLL_UNIT*u;
 165       while (System.currentTimeMillis()<l)
 166         ;
 167   }
 168
 169
 170   private boolean block_document_updates=false;
 171   int STORE_WAIT=5; // how many units before we decide all clients have finalized their appdatas
 172   private boolean in_want_to_store_phase=false;
 173   /**
 174    * client App requests offline storage of vamsas data.
 175    * Call blocks whilst other apps do any appData finalizing
 176    * and then returns (after locking the vamsasDocument in the session)
 177    * Note - the calling app may also receive events through the EventGeneratorThread for document updates.
 178    *
 179    * @return Lock for session.vamArchive
 180    * @param STORE_WAIT indicates how lock the call will block for when nothing appears to be happening to the session.
 181    */
 182   protected Lock want_to_store() {
 183     if (in_want_to_store_phase) {
 184       log.error("client error: want_to_store called again before first call has completed.");
 185       return null;
 186     }
 187     in_want_to_store_phase=true;
 188     // TODO: test the storeDocumentRequest mechanism
 189     /*/ watchThread.haltWatchers();
 190      */
 191     log.debug("Stopping document_update watcher");
 192     vamsasfile.haltWatch();
 193     // block_document_updates=true;
 194     log.debug("Cleared flag for ignoring document_update requests");
 195
 196     log.debug("Sending Store Document Request");
 197     try {
 198       session.addStoreDocumentRequest(client.getClientHandle(), client.getUserHandle());
 199     } catch (Exception e) {
 200       log.warn("Whilst writing StoreDocumentRequest for "+client.getClientHandle().getClientUrn()+" "+client.getUserHandle(),
 201           e);
 202       log.info("trying to continue after storeDocumentRequest exception.");
 203     }
 204     log.debug("Waiting for other apps to do FinalizeApp handling.");
 205     // LATER: refine this semaphore process
 206     // to make a robust signalling mechanism:
 207     // app1 requests, app1..n do something (or don't - they may be dead),
 208     // app1 realises all apps have done their thing, it then continues with synchronized data.
 209     // this probably needs two files - a request file,
 210     //  and a response file which is acknowledged by the app1 requestor for each app.
 211     //  eventually, no more responses are received for the request, and the app can then only continue with its store.
 212     FileWatcher sfwatcher=session.getStoreWatcher();
 213     FileWatcher vfwatcher=session.getDocWatcher();
 214     int units = 0; // zero if updates occured over a sleep period
 215     while (units<STORE_WAIT) {
 216       try {
 217       Thread.sleep(watchThread.WATCH_SLEEP);
 218       } catch (InterruptedException e) {
 219         log.debug("interrupted.");
 220       }
 221       if (sfwatcher.hasChanged() || vfwatcher.hasChanged()) {
 222         units=0;
 223       } else {
 224          units++;
 225       }
 226     }
 227
 228     block_document_updates=false;
 229     vamsasfile.enableWatch();
 230     log.debug("Cleared flag for ignoring document_update requests");
 231     // wait around again (until our own watcher has woken up and synchronized).
 232     while (units<STORE_WAIT) {
 233       try {
 234         Thread.sleep(watchThread.WATCH_SLEEP);
 235         } catch (InterruptedException e) {
 236           log.debug("interrupted.");
 237         }
 238       if (sfwatcher.hasChanged() || vfwatcher.hasChanged())
 239         units=0;
 240       else
 241         units++;
 242     }
 243
 244
 245     log.debug("finished waiting.");
 246     in_want_to_store_phase=false;
 247     return session.vamArchive.getLock();
 248   }
 249   /**
 250    * count handlers for a particular vamsas event
 251    * @param event string enumeration from uk.ac.vamsas.client.Events
 252    * @return -1 for an invalid event, otherwise the number of handlers
 253    */
 254   protected int countHandlersFor(String event) {
 255     if (handlers.containsKey(event)) {
 256       PropertyChangeSupport handler = (PropertyChangeSupport) handlers.get(event);
 257       PropertyChangeListener[] listeners;
 258       if (handler!=null)
 259         return ((listeners=handler.getPropertyChangeListeners())==null)
 260         ? -1 : listeners.length;
 261     }
 262     return -1;
 263   }
 264
 265   public void disableDocumentWatch() {
 266     vamsasfile.haltWatch();
 267   }
 268
 269   public boolean isDocumentWatchEnabled() {
 270     return (vamsasfile!=null) && vamsasfile.isWatchEnabled();
 271   }
 272
 273   public void enableDocumentWatch() {
 274     vamsasfile.enableWatch();
 275   }
 276
 277   public boolean isWatcherAlive() {
 278     return watchThread!=null && watchThread.running && watchThread.isAlive();
 279   }
 280
 281   public void interruptWatching() {
 282     if (watchThread!=null && watchThread.isAlive())
 283       watchThread.interrupt();
 284
 285   }
 286   /**
 287    * called to start the session watching thread which generates events
 288    */
 289   public void startWatching() {
 290     enableDocumentWatch();
 291     watchThread.start();
 292     while (!watchThread.running && watchThread.isAlive())
 293       log.debug("Waiting until watcher is really started.");
 294   }
 295
 296   public void stopWatching() {
 297     interruptWatching();
 298     watchThread.haltWatchers();
 299
 300   }
 301
 302
 303 }