]> gerrit.simantics Code Review - simantics/platform.git/blobdiff - bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/ReadEntry.java
Multiple reader thread support for db client
[simantics/platform.git] / bundles / org.simantics.db.impl / src / org / simantics / db / impl / query / ReadEntry.java
index e4ff1ea874bb2c3b9f8e288d413eaedc49f51c51..c95242defc77a45ae859f49042a2fc1e13df4f20 100644 (file)
@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2007, 2010 Association for Decentralized Information Management
+ * Copyright (c) 2007, 2018 Association for Decentralized Information Management
  * in Industry THTH ry.
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
@@ -14,153 +14,171 @@ package org.simantics.db.impl.query;
 import org.simantics.db.AsyncReadGraph;
 import org.simantics.db.exception.DatabaseException;
 import org.simantics.db.impl.graph.ReadGraphImpl;
-import org.simantics.db.impl.graph.WriteGraphImpl;
 import org.simantics.db.procedure.AsyncProcedure;
 import org.simantics.db.request.Read;
 import org.simantics.db.request.ReadExt;
 import org.simantics.db.request.RequestFlags;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-final public class ReadEntry<T> extends CacheEntryBase {
+public final class ReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> implements AsyncProcedure<T> {
 
-       protected Read<T> request;
+    private static final Logger LOGGER = LoggerFactory.getLogger(ReadEntry.class);
 
-       public ReadEntry(Read<T> request) {
-       this.request = request;
+    protected Read<T> request;
+
+    public ReadEntry(Read<T> request) {
+        this.request = request;
     }
-    
+
     @Override
     int makeHash() {
-       return request.hashCode();
+        return request.hashCode();
     }
-       
+
     @Override
     public Object getOriginalRequest() {
         return request;
     }
-    
+
     @Override
     public void discard() {
-       super.discard();
-       setResult(null);
-    }
-       
-    final public void addOrSet(AsyncReadGraph graph, Object item) {
-
-       assert(assertPending());
-       
-//        ArrayList<Pair<AsyncProcedure<T>, AsyncBarrier>> p = null;
-
-       synchronized(this) {
-           
-            setResult(item);
-               setReady();
-//             p = procs;
-//             procs = null;
-               
-       }
-
-//        if(p != null)
-//             for(Pair<AsyncProcedure<T>, AsyncBarrier> proc : p) {
-//                     proc.first.execute(graph, (T)item);
-//                     proc.second.dec();
-//             }
-        
+        super.discard();
+        setResult(null);
     }
 
     @Override
     final public Query getQuery() {
-        
+
         return new Query() {
 
-                       @Override
-                       public void recompute(ReadGraphImpl graph_, Object provider, CacheEntry entry) {
-                               
-                               QueryProcessor qp = (QueryProcessor)provider;
-                               
-                       WriteGraphImpl write = qp.getCore().getSession().getService(WriteGraphImpl.class);
-                               
-                       ReadGraphImpl graph = write.newSync(entry);
+            @Override
+            public void recompute(ReadGraphImpl graph) {
+
+                try {
 
-                               try {
+                    T result = request.perform(graph);
+                    setResult(result);
+                    setReady();
 
-                                       entry.setPending();
-                                   T result = request.perform(graph);
-                                   addOrSet(graph, result);
+                } catch (Throwable t) {
 
-                               } catch (Throwable t) {
+                    except(t);
+
+                }
+
+            }
+
+            @Override
+            public void removeEntry(QueryProcessor processor) {
+                processor.cache.remove(ReadEntry.this);
+            }
 
-                                       except(t);
-                    
+            @Override
+            public int type() {
+                if (request instanceof ReadExt) {
+                    return ((ReadExt) request).getType();
+                } else {
+                    return RequestFlags.INVALIDATE;
                 }
-                               
-                       }
-
-                       @Override
-                       public void removeEntry(QueryProcessor processor) {
-                               processor.readMap.remove(request);
-                       }
-
-                       @Override
-                       public int type() {
-                               if(request instanceof ReadExt) {
-                                       return ((ReadExt)request).getType();
-                               } else {
-                                       return RequestFlags.INVALIDATE;
-                               }
-                       }
-                       
-                       @Override
-                       public String toString() {
-                               if(request == null) return "DISCARDED";
-                               else return request.toString() + statusOrException;
-                       }
-               
+            }
+
+            @Override
+            public String toString() {
+                if (request == null)
+                    return "DISCARDED";
+                else
+                    return request.toString() + statusOrException;
+            }
+
         };
-        
+
     }
-    
-       public void performFromCache(ReadGraphImpl graph, Object provider,      Object procedure) {
-           
-        AsyncProcedure<T> proc = (AsyncProcedure<T>)procedure;
 
-           if(isExcepted()) {
+    public static <T> void computeForEach(ReadGraphImpl graph, Read<T> request, ReadEntry<T> entry,
+            AsyncProcedure<T> procedure_) throws DatabaseException {
 
-            try {
-                proc.exception(graph, (Throwable)getResult());
-            } catch (Throwable t) {
-                t.printStackTrace();
-            }
-               
-           } else {
-               
-            try {
-                proc.execute(graph, (T)getResult());
-            } catch (Throwable t) {
-                t.printStackTrace();
+        AsyncProcedure<T> procedure = entry != null ? entry : procedure_;
+
+        ReadGraphImpl queryGraph = entry != null ? graph.withParent(entry) : graph;
+
+        try {
+
+            T result = request.perform(queryGraph);
+            procedure.execute(graph, result);
+
+        } catch (DatabaseException e) {
+
+            procedure.exception(graph, e);
+
+        } catch (Throwable t) {
+
+            DatabaseException dbe = new DatabaseException(t);
+            procedure.exception(graph, dbe);
+
+        }
+
+        if (entry != null)
+            entry.performFromCache(queryGraph, procedure_);
+
+    }
+
+    public Object performFromCache(ReadGraphImpl graph, AsyncProcedure<T> procedure) {
+
+        AsyncProcedure<T> proc = (AsyncProcedure<T>) procedure;
+
+        if (proc != null) {
+            if (isExcepted()) {
+                try {
+                    proc.exception(graph, (Throwable) getResult());
+                } catch (Throwable t) {
+                    LOGGER.error("performFromCache proc.exception failed", t);
+                }
+            } else {
+                try {
+                    proc.execute(graph, (T) getResult());
+                } catch (Throwable t) {
+                    LOGGER.error("performFromCache proc.execute failed", t);
+                }
             }
+        }
+
+        return (T) getResult();
 
-           }
-               
-       }
-       
-       @Override
-       public String toString() {
-               if(request == null) return "DISCARDED";
-               else return request.toString() + " - " + statusOrException;
-       }
-       
-       public Object get(ReadGraphImpl graph, QueryProcessor processor, Object procedure) throws DatabaseException {
-               if(procedure != null) performFromCache(graph, processor, procedure);
-               checkAndThrow();
-               return getResult();
-       }
-       
-       @Override
-       boolean isImmutable(ReadGraphImpl graph) throws DatabaseException {
-               if(request instanceof ReadExt) {
-                       return ((ReadExt)request).isImmutable(graph);
-               }
-               return false;
-       }
+    }
+
+    @Override
+    public String toString() {
+        if (request == null)
+            return "DISCARDED";
+        else
+            return request.toString() + " - " + statusOrException;
+    }
+
+    public Object get(ReadGraphImpl graph, AsyncProcedure<T> procedure) throws DatabaseException {
+        if (procedure != null)
+            performFromCache(graph, procedure);
+        checkAndThrow();
+        return getResult();
+    }
+
+    @Override
+    boolean isImmutable(ReadGraphImpl graph) throws DatabaseException {
+        if (request instanceof ReadExt) {
+            return ((ReadExt) request).isImmutable(graph);
+        }
+        return false;
+    }
+
+    @Override
+    public void execute(AsyncReadGraph graph, T result) {
+        setResult(result);
+        setReady();
+    }
+
+    @Override
+    public void exception(AsyncReadGraph graph, Throwable throwable) {
+        except(throwable);
+    }
 
 }