bundles/org.simantics.db.common/src/org/simantics/db/common/procedure/BlockingAsyncProcedure.java

   1 /*******************************************************************************
   2  * Copyright (c) 2007, 2010 Association for Decentralized Information Management
   3  * in Industry THTH ry.
   4  * All rights reserved. This program and the accompanying materials
   5  * are made available under the terms of the Eclipse Public License v1.0
   6  * which accompanies this distribution, and is available at
   7  * http://www.eclipse.org/legal/epl-v10.html
   8  *
   9  * Contributors:
  10  *     VTT Technical Research Centre of Finland - initial API and implementation
  11  *******************************************************************************/
  12 package org.simantics.db.common.procedure;
  13
  14 import java.util.concurrent.Semaphore;
  15 import java.util.concurrent.TimeUnit;
  16 import java.util.concurrent.atomic.AtomicBoolean;
  17
  18 import org.simantics.db.AsyncReadGraph;
  19 import org.simantics.db.common.utils.Logger;
  20 import org.simantics.db.exception.DatabaseException;
  21 import org.simantics.db.procedure.AsyncProcedure;
  22
  23 public class BlockingAsyncProcedure<Result> implements AsyncProcedure<Result> {
  24
  25         final private Object key;
  26     private Result result = null;
  27     private Throwable exception = null;
  28     final private AsyncReadGraph graph;
  29     final private AsyncProcedure<Result> procedure;
  30     final private Semaphore semaphore = new Semaphore(0);
  31 //    final private AtomicBoolean latch;
  32
  33     public BlockingAsyncProcedure(AsyncReadGraph graph, AsyncProcedure<Result> procedure, Object key) {
  34 //      assert(procedure != null);
  35         this.graph = graph;
  36         this.key = key;
  37         this.procedure = procedure;
  38         if(key == null)
  39                 System.err.println("asd");
  40         //System.err.println("BlockingAsyncProcedure " + key);
  41 //        latch = new AtomicBoolean(false);
  42     }
  43
  44     @Override
  45     public void execute(AsyncReadGraph graph, Result result) {
  46         this.result = result;
  47         semaphore.release();
  48 //        if(latch.compareAndSet(false, true)) {
  49                 try {
  50                         if(procedure != null) procedure.execute(graph, result);
  51                 } catch (Throwable throwable) {
  52                 Logger.defaultLogError("AsyncProcedure.execute threw for " + procedure, throwable);
  53                 }
  54 //              } finally {
  55 ////                    System.err.println("ResultCallWrappedSingleQueryProcedure4 dec " + key);
  56 //              }
  57 //        } else {
  58 //              Logger.defaultLogError("Procedure was called many times (this time is execute)");
  59 //        }
  60     }
  61
  62     @Override
  63     public void exception(AsyncReadGraph graph, Throwable t) {
  64         this.exception = t;
  65         semaphore.release();
  66 //        if(latch.compareAndSet(false, true)) {
  67                 try {
  68                         if(procedure != null) procedure.exception(graph, t);
  69                 } catch (Throwable throwable) {
  70                 Logger.defaultLogError("AsyncProcedure.exception threw for " + procedure, throwable);
  71                 } finally {
  72                 }
  73 //        } else {
  74 //              Logger.defaultLogError("Procedure was called many times (this time is exception)");
  75 //        }
  76
  77     }
  78
  79     private void waitFor() throws DatabaseException {
  80
  81         boolean success = false;
  82         success = semaphore.tryAcquire();
  83         if(success) return;
  84
  85         while(!success) {
  86
  87                 if(graph.performPending()) {
  88                         // Some task was done
  89                         success = semaphore.tryAcquire();
  90                 } else {
  91                         // Nothing to do - just wait
  92                 try {
  93                                 success = semaphore.tryAcquire(10, TimeUnit.SECONDS);
  94                                 if(!success) throw new DatabaseException("Timeout while waiting for async request to complete: " + key);
  95                         } catch (InterruptedException e) {
  96                                 throw new DatabaseException(e);
  97                         }
  98                 }
  99
 100         }
 101
 102     }
 103
 104     public Result get() throws DatabaseException {
 105
 106         waitFor();
 107
 108         if(exception != null) {
 109                 if(exception instanceof DatabaseException) throw (DatabaseException)exception;
 110                 throw new DatabaseException(exception);
 111         } else {
 112                 return result;
 113         }
 114
 115     }
 116
 117     public Result getResult() {
 118         return result;
 119     }
 120
 121     public Throwable getException() {
 122         return exception;
 123     }
 124
 125     @Override
 126     public String toString() {
 127         return "." + procedure;
 128     }
 129
 130 }