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
   9 import gnu.trove.map.TObjectIntMap;
  10 import gnu.trove.map.hash.TObjectIntHashMap;
  11
  12 public class ServiceActivityMonitorImpl implements ServiceActivityMonitor {
  13
  14     public static final long REPORTING_PERIOD = 5L;
  15
  16     Semaphore sem = new Semaphore(1);
  17     AtomicInteger ref = new AtomicInteger();
  18     TObjectIntMap<Object> ids = new TObjectIntHashMap<>();
  19
  20     @Override
  21     public void registerActivity(Object id) {
  22         if (ref.getAndIncrement() == 0)
  23             sem.acquireUninterruptibly(1);
  24         synchronized (ids) {
  25             ids.adjustOrPutValue(id, 1, 1);
  26         }
  27     }
  28
  29     @Override
  30     public void unregisterActivity(Object id) {
  31         synchronized (ids) {
  32             int refs = ids.get(id);
  33             if (refs == 1) {
  34                 ids.remove(id);
  35             } else {
  36                 ids.put(id, refs-1);
  37             }
  38         }
  39         if (ref.decrementAndGet() == 0)
  40             sem.release();
  41     }
  42
  43     @Override
  44     public void waitForCompletion() throws InterruptedException {
  45         while(true) {
  46             if(waitForCompletion(REPORTING_PERIOD, TimeUnit.SECONDS))
  47                 return;
  48             synchronized (ids) {
  49                 System.err.println("ServiceActivityMonitorImpl.waitForCompletion: " + ids);
  50             }
  51         }
  52     }
  53
  54     @Override
  55     public boolean waitForCompletion(long timeout, TimeUnit unit) throws InterruptedException {
  56         if(sem.tryAcquire(timeout, unit)) {
  57             sem.release();
  58             return true;
  59         }
  60         else
  61             return false;
  62     }
  63
  64 }