pre-release 2020.3 sql rs refactoring

Author: yuriy-glotanov

src/main/java/su/interference/core/Chunk.java

@@ -45,6 +45,8 @@ public interface Chunk extends Comparable {
     int getBytesAmount();
     Comparable getId (Session s) throws InvocationTargetException, NoSuchMethodException, IllegalAccessException;
     ValueSet getDcs();
+    boolean isTerminate();
+    void setTerminate(boolean terminate);
     Object getEntity() throws ClassNotFoundException, InstantiationException, IllegalAccessException, NoSuchMethodException, InvocationTargetException;
     Object getUndoEntity() throws ClassNotFoundException, InstantiationException, IllegalAccessException, NoSuchMethodException, InvocationTargetException;
     void updateEntity(Object o) throws InternalException, ClassNotFoundException, InstantiationException, IllegalAccessException, NoSuchMethodException, InvocationTargetException, MalformedURLException;

src/main/java/su/interference/core/DataChunk.java

@@ -63,6 +63,7 @@ public class DataChunk implements Chunk {
     private Object undoentity;
     private DataChunk source;
     private UndoChunk uc;
+    private boolean terminate;
     private final CustomSerializer sr = new CustomSerializer();
 
     //returns datacolumn set
@@ -882,6 +883,14 @@ protected synchronized void undo(Session s, LLT llt) throws Exception {
         }
     }
 
+    public boolean isTerminate() {
+        return terminate;
+    }
+
+    public void setTerminate(boolean terminate) {
+        this.terminate = terminate;
+    }
+
     private byte[] getBytesFromHexString(String s) {
         byte b[] = new byte[s.length()/2];
         for (int i=0; i<s.length(); i+=2) {

src/main/java/su/interference/core/Instance.java

@@ -435,8 +435,7 @@ private void startProcesses(Session s) throws Exception {
     }
 
     private void stopProcesses(Session s) throws Exception {
-        final Table t = getTableByName("su.interference.persistent.Process");
-        final ArrayList<Object> pss = t.getAll(s, 0);
+        final List<Process> pss = getProcesses();
         for (Object p : pss) {
             final Process ps = (Process) p;
             ps.stop();

src/main/java/su/interference/core/ManagedCallable.java

@@ -0,0 +1,7 @@
+package su.interference.core;
+
+import java.util.concurrent.Callable;
+
+public interface ManagedCallable<T> extends Callable<T> {
+    void stop();
+}

…/su/interference/core/ManageProcess.java …su/interference/core/ManagedProcess.javasrc/main/java/su/interference/core/ManageProcess.java renamed to src/main/java/su/interference/core/ManagedProcess.java src/main/java/su/interference/core/ManageProcess.java renamed to src/main/java/su/interference/core/ManagedProcess.java

@@ -1,34 +1,34 @@
-/**
- The MIT License (MIT)
-
- Copyright (c) 2010-2019 head systems, ltd
-
- Permission is hereby granted, free of charge, to any person obtaining a copy of
- this software and associated documentation files (the "Software"), to deal in
- the Software without restriction, including without limitation the rights to
- use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
- the Software, and to permit persons to whom the Software is furnished to do so,
- subject to the following conditions:
-
- The above copyright notice and this permission notice shall be included in all
- copies or substantial portions of the Software.
-
- THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
- FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
- COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
- IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
- CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
-
- */
-
-package su.interference.core;
-
-/**
- * @author Yuriy Glotanov
- * @since 1.0
- */
-
-public interface ManageProcess {
-    void stop() throws InterruptedException;
-}
+/**
+ The MIT License (MIT)
+
+ Copyright (c) 2010-2019 head systems, ltd
+
+ Permission is hereby granted, free of charge, to any person obtaining a copy of
+ this software and associated documentation files (the "Software"), to deal in
+ the Software without restriction, including without limitation the rights to
+ use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ the Software, and to permit persons to whom the Software is furnished to do so,
+ subject to the following conditions:
+
+ The above copyright notice and this permission notice shall be included in all
+ copies or substantial portions of the Software.
+
+ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+ IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+ */
+
+package su.interference.core;
+
+/**
+ * @author Yuriy Glotanov
+ * @since 1.0
+ */
+
+public interface ManagedProcess {
+    void stop() throws InterruptedException;
+}

src/main/java/su/interference/core/RetrieveQueue.java

@@ -0,0 +1,65 @@
+package su.interference.core;
+
+import java.util.concurrent.LinkedBlockingQueue;
+
+public class RetrieveQueue {
+    private final LinkedBlockingQueue<Chunk> q;
+    private final ManagedCallable r;
+    private boolean retrieve = true;
+
+    public RetrieveQueue(LinkedBlockingQueue<Chunk> q, ManagedCallable r) {
+        this.q = q;
+        this.r = r;
+    }
+
+    public Object poll() {
+        try {
+            if (retrieve) {
+                Chunk c = q.take();
+                if (c.isTerminate()) {
+                    retrieve = false;
+                    return null;
+                }
+                return c.getEntity();
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return null;
+    }
+
+    public Chunk cpoll() {
+        try {
+            if (retrieve) {
+                Chunk c = q.take();
+                if (c.isTerminate()) {
+                    retrieve = false;
+                    return null;
+                }
+                return c;
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return null;
+    }
+
+    public void stop() {
+        retrieve = false;
+        if (r != null) {
+            r.stop();
+        }
+    }
+
+    public LinkedBlockingQueue<Chunk> getQ() {
+        return q;
+    }
+
+    public ManagedCallable getR() {
+        return r;
+    }
+
+    public boolean isRetrieve() {
+        return retrieve;
+    }
+}

src/main/java/su/interference/core/SyncQueue.java

@@ -41,7 +41,7 @@ this software and associated documentation files (the "Software"), to deal in
  * @since 1.0
  */
 
-public class SyncQueue implements Runnable, ManageProcess {
+public class SyncQueue implements Runnable, ManagedProcess {
 
     private volatile boolean f = true;
     private volatile boolean running = false;

src/main/java/su/interference/persistent/Process.java

@@ -94,7 +94,7 @@ public void start (Runnable r, Session s) {
 
     public void stop () throws InterruptedException {
         if (th!=null&&ro!=null) {
-            ((ManageProcess)ro).stop();
+            ((ManagedProcess)ro).stop();
             th.join();
         }
     }

src/main/java/su/interference/persistent/Session.java

@@ -45,6 +45,7 @@ this software and associated documentation files (the "Software"), to deal in
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import javax.persistence.*;
 
 /**
@@ -111,7 +112,11 @@ public static int getCLASS_ID() {
     private static ThreadLocal<Session> contextSession = new ThreadLocal<Session>();
 
     @Transient
-    private final ExecutorService streampool = Executors.newCachedThreadPool();
+    private static final ExecutorService rqpool = Executors.newCachedThreadPool();
+    @Transient
+    private volatile RetrieveQueue retrieveQueue;
+    @Transient
+    private static final ExecutorService streampool = Executors.newCachedThreadPool();
     @Transient
     private final Map<Long, Integer> streammap = new ConcurrentHashMap<>();
     @Transient
@@ -168,7 +173,7 @@ private void startStatement(LLT llt) throws Exception {
         //this.getTransaction().startStatement(this, llt);
     }
 
-    public void setTransaction(Transaction transaction) {
+    protected void setTransaction(Transaction transaction) {
         this.transaction = transaction;
     }
 
@@ -269,27 +274,36 @@ public EntityFactory getEntityFactory(Class c) {
         return new EntityFactory(c, this);
     }
 
-    public List<Object> getAll (Class c) throws Exception {
+    public Object find (Class c, long id) throws Exception {
         final Table t = Instance.getInstance().getTableByName(c.getName());
         if (t != null) {
             this.startStatement();
-//            t.lockTable(this);
-            final List<Object> r = t.getAll(this, 0);
-//            t.unlockTable(this);
-            return r;
+            return t.getChunkById(id, this).getEntity();
         }
         return null;
     }
 
-    public Object find (Class c, long id) throws Exception {
-        final Table t = Instance.getInstance().getTableByName(c.getName());
+    protected synchronized RetrieveQueue getContentQueue(Table t) {
         if (t != null) {
-            this.startStatement();
-            return t.getChunkById(id, this).getEntity();
+            retrieveQueue = t.getContentQueue(this);
+            Future f = rqpool.submit(retrieveQueue.getR());
+            return retrieveQueue;
         }
         return null;
     }
 
+    public void closeQueue() {
+        if (this.retrieveQueue != null) {
+            retrieveQueue.stop();
+        }
+        retrieveQueue = null;
+    }
+
+    public void close() {
+        rqpool.shutdownNow();
+        streampool.isShutdown();
+    }
+
     //todo uncommitted data not retrieved
     public void stream (Class c, StreamCallable task) {
         final Map<Long, Long> retrieved = new HashMap<>();
@@ -318,17 +332,7 @@ public void run() {
         });
     }
 
-    public List<Object> ngetAll (Class c) throws Exception {
-        final Table t = Instance.getInstance().getTableByName(c.getName());
-        if (t != null) {
-//            t.lockTable(this);
-            final List<Object> r = t.getAll(this, 0);
-//            t.unlockTable(this);
-            return r;
-        }
-        return null;
-    }
-
+    @Deprecated
     public Object nfind (Class c, long id) throws InternalException, NoSuchMethodException, InvocationTargetException, IOException, InvalidFrameHeader, InvalidFrame, EmptyFrameHeaderFound, IncorrectUndoChunkFound, ClassNotFoundException, InstantiationException, IllegalAccessException {
         final Table t = Instance.getInstance().getTableByName(c.getName());
         if (t != null) {
@@ -544,6 +548,10 @@ public static void setDntmSession(Session dntmSession) {
         Session.dntmSession = dntmSession;
     }
 
+    public RetrieveQueue getRetrieveQueue() {
+        return retrieveQueue;
+    }
+
     public void streamFramePtr(Frame f, int ptr) {
         streammap.put(f.getAllocFile()+f.getAllocPointer(), ptr);
     }