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