bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/AsyncReadEntry.java

   1 /*******************************************************************************
   2  * Copyright (c) 2007, 2018 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.impl.query;
  13
  14 import java.util.Collection;
  15
  16 import org.simantics.databoard.Bindings;
  17 import org.simantics.db.AsyncReadGraph;
  18 import org.simantics.db.DevelopmentKeys;
  19 import org.simantics.db.exception.DatabaseException;
  20 import org.simantics.db.impl.BlockingAsyncProcedure;
  21 import org.simantics.db.impl.graph.ReadGraphImpl;
  22 import org.simantics.db.impl.query.QueryProcessor.SessionTask;
  23 import org.simantics.db.procedure.AsyncProcedure;
  24 import org.simantics.db.request.AsyncRead;
  25 import org.simantics.utils.Development;
  26 import org.slf4j.Logger;
  27 import org.slf4j.LoggerFactory;
  28
  29 final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> implements AsyncProcedure<T>, IPending {
  30
  31     private static final Logger LOGGER = LoggerFactory.getLogger(AsyncReadEntry.class);
  32
  33     protected AsyncRead<T> id;
  34     protected PendingTaskSupport pendingTaskSupport;
  35
  36     AsyncReadEntry(AsyncRead<T> request) {
  37         this.id = request;
  38                 if (Development.DEVELOPMENT) {
  39                         if(Development.<Boolean>getProperty(DevelopmentKeys.CACHE_ENTRY_STATE, Bindings.BOOLEAN)) {
  40                         System.err.println("[QUERY STATE]: created " + this);
  41                         }
  42                 }
  43     }
  44
  45     @Override
  46     int makeHash() {
  47         return id.hashCode();
  48     }
  49
  50     @Override
  51     public Object getOriginalRequest() {
  52         return id;
  53     }
  54
  55     @Override
  56     public void discard() {
  57         super.discard();
  58         setResult(null);
  59     }
  60
  61     public void except(AsyncReadGraph graph, Throwable t) {
  62
  63         assert (isPending());
  64
  65         synchronized (this) {
  66             except(t);
  67         }
  68
  69     }
  70
  71     @Override
  72     final public Query getQuery() {
  73
  74         return new Query() {
  75
  76             @Override
  77             public void recompute(ReadGraphImpl graph) {
  78
  79                 try {
  80
  81                     BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure(graph, AsyncReadEntry.this, new AsyncProcedure<T>() {
  82
  83                         @Override
  84                         public void execute(AsyncReadGraph graph, T result) {
  85                             setResult(result);
  86                             setReady();
  87                         }
  88
  89                         @Override
  90                         public void exception(AsyncReadGraph graph, Throwable t) {
  91                             except(t);
  92                         }
  93
  94                     }, id, true);
  95
  96                     proc.performSync(id);
  97
  98                 } catch (Throwable t) {
  99                     except(t);
 100                 }
 101
 102             }
 103
 104             @Override
 105             public void removeEntry(QueryProcessor qp) {
 106                 qp.cache.remove(AsyncReadEntry.this);
 107             }
 108
 109             @Override
 110             public int type() {
 111                 return id.getFlags();
 112             }
 113
 114             @Override
 115             public String toString() {
 116                 if (id == null)
 117                     return "DISCARDED";
 118                 else if (isExcepted())
 119                     return id.toString() + " " + getResult();
 120                 else
 121                     return id.toString() + " " + statusOrException;
 122             }
 123
 124         };
 125
 126     }
 127
 128     @Override
 129     public Object performFromCache(ReadGraphImpl graph, AsyncProcedure<T> proc) {
 130
 131         if (isExcepted()) {
 132
 133             try {
 134                 proc.exception(graph, (Throwable) getResult());
 135             } catch (Throwable t) {
 136                 LOGGER.error("performFromCache proc.exception failed", t);
 137             }
 138
 139         } else {
 140
 141             try {
 142                 T result = (T) getResult();
 143                 proc.execute(graph, result);
 144             } catch (Throwable t) {
 145                 LOGGER.error("performFromCache proc.execute failed", t);
 146             }
 147
 148         }
 149
 150         return getResult();
 151
 152     }
 153
 154     public static <T> T computeForEach(ReadGraphImpl callerGraph, AsyncRead<T> request, AsyncReadEntry<T> entry,
 155             AsyncProcedure<T> procedure_, boolean needsToBlock) throws DatabaseException {
 156
 157         BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure(callerGraph, entry, procedure_, request, needsToBlock);
 158         if(needsToBlock) {
 159             return proc.performSync(request);
 160         } else {
 161             proc.performAsync(request);
 162             return null;
 163         }
 164
 165     }
 166
 167     @Override
 168     public String toString() {
 169         if (isDiscarded())
 170             return "DISCARDED " + id.toString();
 171         else if (isExcepted())
 172             return id.toString() + " " + getResult();
 173         else
 174             return id.toString() + " " + statusOrException;
 175     }
 176
 177     @Override
 178     public void execute(AsyncReadGraph graph, T result) {
 179         Collection<SessionTask> tasks = null;
 180         synchronized(this) {
 181             setResult(result);
 182             setReady();
 183             if(pendingTaskSupport != null)
 184                 tasks = pendingTaskSupport.executePending();
 185         }
 186         if(tasks != null)
 187             for(SessionTask task : tasks)
 188                 ((ReadGraphImpl)graph).processor.scheduleNow(task);
 189     }
 190
 191     @Override
 192     public synchronized void exception(AsyncReadGraph graph, Throwable throwable) {
 193         Collection<SessionTask> tasks = null;
 194         synchronized(this) {
 195             except(throwable);
 196             if(pendingTaskSupport != null)
 197                 tasks = pendingTaskSupport.executePending();
 198         }
 199         if(tasks != null)
 200             for(SessionTask task : tasks)
 201                 ((ReadGraphImpl)graph).processor.scheduleNow(task);
 202     }
 203
 204     public void executeWhenResultIsAvailable(QueryProcessor processor, SessionTask task) {
 205         boolean ready = false;
 206         synchronized(this) {
 207             if(pendingTaskSupport == null)
 208                 pendingTaskSupport = new PendingTaskSupport(this);
 209             ready = pendingTaskSupport.executeWhenResultIsAvailable(task);
 210         }
 211         if(ready) {
 212             processor.scheduleNow(task);
 213         }
 214     }
 215
 216 }