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

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