bundles/org.simantics.db.common/src/org/simantics/db/common/service/ServiceActivityMonitorImpl.java

   1 package org.simantics.db.common.service;
   2
   3 import java.util.concurrent.Semaphore;
   4 import java.util.concurrent.TimeUnit;
   5 import java.util.concurrent.atomic.AtomicInteger;
   6
   7 import org.simantics.db.service.ServiceActivityMonitor;
   8 import org.slf4j.Logger;
   9 import org.slf4j.LoggerFactory;
  10
  11 import gnu.trove.map.TObjectIntMap;
  12 import gnu.trove.map.hash.TObjectIntHashMap;
  13
  14 public class ServiceActivityMonitorImpl implements ServiceActivityMonitor {
  15
  16     private static final Logger LOGGER = LoggerFactory.getLogger(ServiceActivityMonitorImpl.class);
  17
  18     public static final long REPORTING_PERIOD = 5L;
  19
  20     Semaphore sem = new Semaphore(1);
  21     AtomicInteger ref = new AtomicInteger();
  22     TObjectIntMap<Object> ids = new TObjectIntHashMap<>();
  23
  24     @Override
  25     public void registerActivity(Object id) {
  26         if (ref.getAndIncrement() == 0)
  27             sem.acquireUninterruptibly(1);
  28         synchronized (ids) {
  29             ids.adjustOrPutValue(id, 1, 1);
  30         }
  31     }
  32
  33     @Override
  34     public void unregisterActivity(Object id) {
  35         synchronized (ids) {
  36             int refs = ids.get(id);
  37             if (refs == 1) {
  38                 ids.remove(id);
  39             } else {
  40                 ids.put(id, refs-1);
  41             }
  42         }
  43         if (ref.decrementAndGet() == 0)
  44             sem.release();
  45     }
  46
  47     @Override
  48     public void waitForCompletion() throws InterruptedException {
  49         while(true) {
  50             if(waitForCompletion(REPORTING_PERIOD, TimeUnit.SECONDS))
  51                 return;
  52             synchronized (ids) {
  53                 LOGGER.info("waitForCompletion: " + ids);
  54             }
  55         }
  56     }
  57
  58     @Override
  59     public boolean waitForCompletion(long timeout, TimeUnit unit) throws InterruptedException {
  60         if(sem.tryAcquire(timeout, unit)) {
  61             sem.release();
  62             return true;
  63         }
  64         else
  65             return false;
  66     }
  67
  68 }