bundles/org.simantics.db.layer0/src/org/simantics/db/layer0/util/EvaluatingListener.java

   1 /*******************************************************************************
   2  * Copyright (c) 2012 Association for Decentralized Information Management in
   3  * 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.layer0.util;
  13
  14 import java.util.concurrent.Semaphore;
  15 import java.util.concurrent.TimeUnit;
  16
  17 import org.simantics.db.Session;
  18 import org.simantics.db.exception.DatabaseException;
  19 import org.simantics.db.procedure.Listener;
  20 import org.simantics.db.request.Read;
  21
  22 /**
  23  * A database listener that disposes itself based on a criterion evaluated using
  24  * the request results received by the listener. If the results pass the
  25  * criterion, the listener is disposed and the result is either accepted or
  26  * discarded based on the evaluation.
  27  *
  28  * <p>
  29  * The reason for the existence of this class is that sometimes database
  30  * requests always perform asynchronously, i.e. return empty results at first
  31  * and later get updated to their final results. For example, a the results of a
  32  * web request would take a while to be delivered. This listener makes it a bit
  33  * easier to deal with these situations when results are needed synchronously.
  34  *
  35  * <p>
  36  * {@link #trySyncRequest(Session, Read, Criterion, long, TimeUnit)} provides a
  37  * utility for executing a synchronous read request with this listener so that
  38  * the implementation will wait for a criterion accepted result for the
  39  * specified amount of time. If an accepted result is reached within the time
  40  * limit, it will be returned. If the time limit is reached and no accepted
  41  * result is attained, <code>null</code> will be returned. If the request
  42  * produces an exception, it will be thrown.
  43  *
  44  * @author Tuukka Lehtonen
  45  *
  46  * @param <T> database request result type
  47  *
  48  * @see EvaluatingListener.Evaluation
  49  * @see EvaluatingListener.Criterion
  50  */
  51 public class EvaluatingListener<T> implements Listener<T> {
  52
  53     public static enum Evaluation {
  54         /**
  55          * Keep on listening to further results.
  56          */
  57         IGNORE,
  58         /**
  59          * Dispose listener and discard the results.
  60          */
  61         DISCARD,
  62         /**
  63          * Dispose listener and return the latest result.
  64          */
  65         ACCEPT
  66     }
  67
  68     /**
  69      * An evaluable criterion for the result received by
  70      * {@link Listener#execute(Object)} to tell whether to accept the result,
  71      * wait for another result or to consider the listener disposed.
  72      *
  73      * @param <T> the type of the result
  74      */
  75     public static interface Criterion<T> {
  76         Evaluation evaluate(T result);
  77     }
  78
  79     /**
  80      * The criterion the listener evaluates. When it evaluates to
  81      * {@value Evaluation#DISCARD}, this field is nullified and the listener is
  82      * considered disposed.
  83      */
  84     private volatile Criterion<T> criterion;
  85     private T                     result;
  86     private Throwable             exception;
  87     private Semaphore             wait = new Semaphore(0);
  88
  89     public EvaluatingListener(Criterion<T> criterion) {
  90         if (criterion == null)
  91             throw new NullPointerException("null criterion");
  92         this.criterion = criterion;
  93     }
  94
  95     /**
  96      * @param session
  97      * @param request
  98      * @param criterion
  99      * @param timeout
 100      * @param unit
 101      * @return
 102      * @throws InterruptedException
 103      * @throws DatabaseException
 104      */
 105     public static <T> T trySyncRequest(Session session, Read<T> request, Criterion<T> criterion, long timeout, TimeUnit unit) throws InterruptedException, DatabaseException {
 106         EvaluatingListener<T> l = new EvaluatingListener<T>(criterion);
 107         session.asyncRequest(request, l);
 108         l.tryWaitForResult(timeout, unit);
 109         // Make sure the listener is disposed.
 110         l.dispose();
 111         l.throwPossibleException();
 112         return l.getResult();
 113     }
 114
 115     public T waitForResult() throws InterruptedException {
 116         wait.acquire();
 117         return getResult();
 118     }
 119
 120     public boolean tryWaitForResult(long timeout, TimeUnit unit) throws InterruptedException {
 121         return wait.tryAcquire(timeout, unit);
 122     }
 123
 124     public T getResult() {
 125         return result;
 126     }
 127
 128     public Throwable getException() {
 129         return exception;
 130     }
 131
 132     public void throwPossibleException() throws DatabaseException {
 133         if (exception != null) {
 134             if (exception instanceof DatabaseException)
 135                 throw (DatabaseException) exception;
 136             throw new DatabaseException(exception);
 137         }
 138     }
 139
 140     @Override
 141     public void execute(T result) {
 142         Criterion<T> crit = criterion;
 143         if (crit == null)
 144             return;
 145         EvaluatingListener.Evaluation e = crit.evaluate(result);
 146         switch (e) {
 147             case IGNORE:
 148                 ignored(result);
 149                 return;
 150             case ACCEPT:
 151                 this.result = result;
 152                 try {
 153                     accepted(result);
 154                 } finally {
 155                     dispose();
 156                     wait.release();
 157                 }
 158                 return;
 159             case DISCARD:
 160                 dispose();
 161                 wait.release();
 162                 return;
 163         }
 164     }
 165
 166     /**
 167      * Override to process results that were ignored.
 168      *
 169      * @param result
 170      */
 171     public void ignored(T result) {
 172     }
 173
 174     /**
 175      * Override this to immediately process an accepted result in the listener.
 176      * This method is invoked before the listener is disposed
 177      *
 178      * @param result
 179      */
 180     public void accepted(T result) {
 181     }
 182
 183     @Override
 184     public void exception(Throwable t) {
 185         this.exception = t;
 186         dispose();
 187         wait.release();
 188     }
 189
 190     private void dispose() {
 191         this.criterion = null;
 192     }
 193
 194     @Override
 195     public boolean isDisposed() {
 196         return criterion == null;
 197     }
 198
 199 }