Merge pull request #20 from yuriy-glotanov/2020.1beta

release 2020.1 beta (unstable)

Author: yuriy-glotanov

README.md

@@ -2,7 +2,7 @@
 
 ##### simple distributed persistent layer for java applications
 ##### (c) 2010 - 2020 head systems, ltd
-##### current revision: 2019.3
+##### current revision: 2020.1 beta (unstable)
 ##### for detailed information see doc/InterferenceManual.pdf
 
 ##### contacts: info@inteference.su
@@ -13,6 +13,7 @@
 
 - supports Base JPA annotations
 - supports local & distributed SQL queries
+- supports complex event processing and streaming SQL
 - supports transactions
 - supports unique constraints
 - supports persistent indexes
@@ -25,7 +26,7 @@
   included in the cluster
 - does not contain any coordination nodes and does not require 
   the launch of any additional coordinators. All cluster nodes are equivalent.
-- supports complex event processing and streaming SQL (in next release)
+
 
 ## NOTE:
 
@@ -53,7 +54,7 @@ like this:
     <dependency>
         <groupId>su.interference</groupId>
         <artifactId>interference</artifactId>
-        <version>2019.3</version>
+        <version>2020.1</version>
     </dependency>
     ...
 </dependencies>

pom.xml

@@ -7,11 +7,11 @@
 
     <groupId>su.interference</groupId>
     <artifactId>interference</artifactId>
-    <version>2019.3</version>
+    <version>2020.1</version>
     <packaging>jar</packaging>
 
     <name>interference</name>
-    <description>distributed persistent layer with JPA/SQL support</description>
+    <description>distributed persistent layer with JPA/SQL/CEP support</description>
     <url>https://github.com/interference-project/interference</url>
 
     <licenses>

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/Config.java

@@ -59,6 +59,7 @@ public class Config {
     private static final String P_DISKIO_MODE = "diskio.mode";
     private static final String P_SYNC_LOCK_ENABLE = "sync.lock.enable";
     private static final String P_SYNC_PERIOD = "sync.period";
+    private static final String P_RETRIEVE_QUEUE_SIZE = "retrieve.queue.size";
     private static final String P_CODEPAGE = "codepage";
     private static final String P_DATEFORMAT = "dateformat";
     private static final int MAX_NODE_ID = 64;
@@ -80,7 +81,8 @@ public class Config {
     private static final int FILES_AMOUNT_DEFAULT=4;
     private static final String DISKIO_MODE_DEFAULT="rws";
     private static final boolean SYNC_LOCK_ENABLE_DEFAULT=false;
-    private static final int SYNC_PERIOD_DEFAULT=5;
+    private static final int SYNC_PERIOD_DEFAULT=2000;
+    private static final int RETRIEVE_QUEUE_SIZE_DEFAULT=10000;
     private static final String CODEPAGE_DEFAULT="UTF-8";
     private static final String DATEFORMAT_DEFAULT="dd.MM.yyyy";
 
@@ -101,6 +103,7 @@ public class Config {
     public final String DISKIO_MODE;
     public final boolean SYNC_LOCK_ENABLE;
     public final int SYNC_PERIOD;
+    public final int RETRIEVE_QUEUE_SIZE;
     public final String CODEPAGE;
     public final String DATEFORMAT;
 
@@ -158,6 +161,7 @@ private Config() {
             DISKIO_MODE = validateDiskioMode(p.getProperty(P_DISKIO_MODE));
             SYNC_LOCK_ENABLE = validateSyncLock(p.getProperty(P_SYNC_LOCK_ENABLE));
             SYNC_PERIOD = validateSyncPeriod(p.getProperty(P_SYNC_PERIOD));
+            RETRIEVE_QUEUE_SIZE = validateQueueSize(p.getProperty(P_RETRIEVE_QUEUE_SIZE));
         } else {
             DB_PATH = DB_PATH_DEFAULT;
             JOURNAL_PATH = JOURNAL_PATH_DEFAULT;
@@ -178,6 +182,7 @@ private Config() {
             DISKIO_MODE = DISKIO_MODE_DEFAULT;
             SYNC_LOCK_ENABLE = SYNC_LOCK_ENABLE_DEFAULT;
             SYNC_PERIOD = SYNC_PERIOD_DEFAULT;
+            RETRIEVE_QUEUE_SIZE = RETRIEVE_QUEUE_SIZE_DEFAULT;
         }
         System.setProperty("com.sun.management.jmxremote.port","8111");
         System.setProperty("com.sun.management.jmxremote.authenticate","false");
@@ -324,7 +329,19 @@ private static String validateDiskioMode(String mode) {
     private static int validateSyncPeriod(String value) {
         try {
             int p = Integer.valueOf(value);
-            if (p > 0 && p < 10000) {
+            if (p > 9 && p < 60001) {
+                return p;
+            }
+        } catch(NumberFormatException e) {
+            logger.error("sync period value is not valid - use default value");
+        }
+        return SYNC_PERIOD_DEFAULT;
+    }
+
+    private static int validateQueueSize(String value) {
+        try {
+            int p = Integer.valueOf(value);
+            if (p > 0 && p < 100000000) {
                 return p;
             }
         } catch(NumberFormatException e) {

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,36 @@
+/**
+ 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;
+
+import java.util.concurrent.Callable;
+
+/**
+ * @author Yuriy Glotanov
+ * @since 1.0
+ */
+
+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,94 @@
+/**
+ 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;
+
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * @author Yuriy Glotanov
+ * @since 1.0
+ */
+
+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()) {
+                    stop();
+                    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()) {
+                    stop();
+                    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;
@@ -141,7 +141,7 @@ public void run () {
             }
 
             try {
-                final int period = Config.getConfig().SYNC_PERIOD*1000;
+                final int period = Config.getConfig().SYNC_PERIOD;
                 Thread.sleep(period);
             } catch (InterruptedException e) {
                 e.printStackTrace();