]> gerrit.simantics Code Review - simantics/platform.git/blobdiff - bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/AsyncReadEntry.java
Multiple reader thread support for db client
[simantics/platform.git] / bundles / org.simantics.db.impl / src / org / simantics / db / impl / query / AsyncReadEntry.java
index 010f554463373ace1edf3ac21cef13a4cc17122a..00abbebf65859e76d0e8dad481d01097cab04621 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
  *******************************************************************************/
 package org.simantics.db.impl.query;
 
-import java.util.ArrayList;
-
 import org.simantics.db.AsyncReadGraph;
+import org.simantics.db.exception.DatabaseException;
+import org.simantics.db.exception.RuntimeDatabaseException;
+import org.simantics.db.impl.BlockingAsyncProcedure;
 import org.simantics.db.impl.DebugPolicy;
 import org.simantics.db.impl.graph.ReadGraphImpl;
 import org.simantics.db.procedure.AsyncProcedure;
 import org.simantics.db.request.AsyncRead;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> implements AsyncProcedure<T> {
 
-final public class AsyncReadEntry<T> extends CacheEntryBase {
+    private static final Logger LOGGER = LoggerFactory.getLogger(AsyncReadEntry.class);
 
     protected AsyncRead<T> request;
 
-    public AsyncReadEntry(AsyncRead<T> request) {
-       this.request = request;
-       if(DebugPolicy.QUERY_STATE) System.out.println("[QUERY STATE]: created " + this);
+    AsyncReadEntry(AsyncRead<T> request) {
+        this.request = request;
+        if (DebugPolicy.QUERY_STATE)
+            System.out.println("[QUERY STATE]: created " + this);
     }
 
     @Override
     int makeHash() {
-       return request.hashCode();
+        return request.hashCode();
     }
-    
+
     @Override
     public Object getOriginalRequest() {
         return request;
     }
-    
+
     @Override
     public void discard() {
-       super.discard();
-       //request = null;
-       setResult(null);
+        super.discard();
+        setResult(null);
     }
-    
-    final public void addOrSet(AsyncReadGraph graph, Object item) {
 
-       assert(isPending());
-        
-//        ArrayList<AsyncProcedure<T>> p = null;
-
-        synchronized(this) {
-               
-            setResult(item);
-               setReady();
-//            p = procs;
-//            procs = null;
-            
-        }
-
-//        if(p != null)
-//            for(AsyncProcedure<T> proc : p) {
-//             proc.execute(graph, (T)item);
-////                proc.first.execute(graph, (T)item);
-////                proc.second.dec();
-//            }
-        
-    }
-    
-    
     public void except(AsyncReadGraph graph, Throwable t) {
-       
-       assert(isPending());
-        
-//        ArrayList<AsyncProcedure<T>> p = null;
 
-        synchronized(this) {
-               
+        assert (isPending());
+
+        synchronized (this) {
             except(t);
-////            p = procs;
-//            procs = null;
-            
         }
 
-//        if(p != null)
-//            for(AsyncProcedure<T> proc : p) {
-//             proc.exception(graph, t);
-//            }
-       
     }
-    
-    
+
     @Override
     final public Query getQuery() {
-       
-        return new Query() {
 
-                       @Override
-                       public void recompute(ReadGraphImpl graph, Object provider, CacheEntry entry) {
-                               
-                               QueryProcessor qp = (QueryProcessor)provider;
+        return new Query() {
 
-                               final ReadGraphImpl parentGraph = ReadGraphImpl.forRecompute(entry, qp); 
+            @Override
+            public void recompute(ReadGraphImpl graph) {
 
-                               try {
+                try {
 
-                                   request.perform(parentGraph , new AsyncProcedure<T>() {
+                    BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure<>(graph, new AsyncProcedure<T>() {
 
                         @Override
                         public void execute(AsyncReadGraph graph, T result) {
-                            addOrSet(graph, result);
+                            setResult(result);
+                            setReady();
                         }
-                               
-                               @Override
-                               public void exception(AsyncReadGraph graph, Throwable t) {
-                                   except(t);
+
+                        @Override
+                        public void exception(AsyncReadGraph graph, Throwable t) {
+                            except(t);
                         }
 
-                    });
+                    }, request);
 
-                               } catch (Throwable t) {
-                                   except(t);
+                    request.perform(graph, proc);
+
+                    proc.get();
+
+                } catch (Throwable t) {
+                    except(t);
                 }
-                               
-                       }
-
-                       @Override
-                       public void removeEntry(QueryProcessor qp) {
-                       qp.asyncReadMap.remove(request);
-                       }
-
-                       @Override
-                       public int type() {
-                               return request.getFlags();
-                       }
-                       
-                       @Override
-                       public String toString() {
-                               if(request == null) return "DISCARDED";
-                               else if(isExcepted()) return request.toString() + " " + getResult();
-                               else return request.toString() + " " + statusOrException;
-                       }
-               
+
+            }
+
+            @Override
+            public void removeEntry(QueryProcessor qp) {
+                qp.cache.remove(AsyncReadEntry.this);
+            }
+
+            @Override
+            public int type() {
+                return request.getFlags();
+            }
+
+            @Override
+            public String toString() {
+                if (request == null)
+                    return "DISCARDED";
+                else if (isExcepted())
+                    return request.toString() + " " + getResult();
+                else
+                    return request.toString() + " " + statusOrException;
+            }
+
         };
-        
+
     }
 
-       public void performFromCache(ReadGraphImpl graph, Object provider, Object procedure) {
-               
-        AsyncProcedure<T> proc = (AsyncProcedure<T>)procedure;
+    @Override
+    public Object performFromCache(ReadGraphImpl graph, AsyncProcedure<T> proc) {
+
+        if (isExcepted()) {
 
-           if(isExcepted()) {
-            
             try {
-                proc.exception(graph, (Throwable)getResult());
+                proc.exception(graph, (Throwable) getResult());
             } catch (Throwable t) {
-                t.printStackTrace();
+                LOGGER.error("performFromCache proc.exception failed", t);
             }
-            
+
         } else {
-            
+
             try {
-                proc.execute(graph, (T)getResult());
+                proc.execute(graph, (T) getResult());
             } catch (Throwable t) {
-                t.printStackTrace();
+                LOGGER.error("performFromCache proc.execute failed", t);
             }
-            
+
         }
-               
-       }
-
-       @Override
-       public String toString() {
-               if(isDiscarded()) return "DISCARDED " + request.toString();
-               else if(isExcepted()) return request.toString() + " " + getResult();
-               else return request.toString() + " " + statusOrException;
-       }
+
+        return getResult();
+
+    }
+
+    public static <T> void computeForEach(ReadGraphImpl parentGraph, AsyncRead<T> request, AsyncReadEntry<T> entry,
+            AsyncProcedure<T> procedure_) throws DatabaseException {
+
+        AsyncProcedure<T> procedure = entry != null ? entry : procedure_;
+
+        ReadGraphImpl queryGraph = parentGraph.withParent(entry);
+        
+        BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure<>(queryGraph, new AsyncProcedure<T>() {
+
+            @Override
+            public void execute(AsyncReadGraph returnGraph, T result) {
+                try {
+                    procedure.execute(parentGraph, result);
+                } catch (Throwable t) {
+                    LOGGER.error("computeForEach procedure.execute failed", t);
+                }
+            }
+
+            @Override
+            public void exception(AsyncReadGraph returnGraph, Throwable t) {
+                try {
+                    procedure.exception(parentGraph, t);
+                } catch (Throwable t2) {
+                    LOGGER.error("computeForEach procedure.exception failed", t2);
+                }
+            }
+
+            @Override
+            public String toString() {
+                return procedure.toString();
+            }
+
+        }, request);
+
+        request.perform(queryGraph, proc);
+        
+        proc.get();
+
+        if (entry != null)
+            entry.performFromCache(parentGraph, procedure_);
+
+    }
+
+    @Override
+    public String toString() {
+        if (isDiscarded())
+            return "DISCARDED " + request.toString();
+        else if (isExcepted())
+            return request.toString() + " " + getResult();
+        else
+            return request.toString() + " " + statusOrException;
+    }
+
+    @Override
+    public void execute(AsyncReadGraph graph, T result) {
+        setResult(result);
+        setReady();
+    }
+
+    @Override
+    public void exception(AsyncReadGraph graph, Throwable throwable) {
+        except(throwable);
+    }
 
 }