WebSVN – XServices – Diff – /xservices/trunk/src/main/java/net/brutex/xservices/util/MiscServiceServletContextListener.java

 import java.sql.Connection;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Date;
 import java.util.concurrent.atomic.AtomicLong;
 import static org.quartz.TriggerBuilder.newTrigger;
+     *
      * @return connection pool
      */
     private static JdbcConnectionPool getDbPool(String dbConnectString) {
         JdbcConnectionPool p = JdbcConnectionPool.create(dbConnectString, "", "");
-        p.setMaxConnections(16);
+        p.setMaxConnections(256);
-        p.setLoginTimeout(5);
+        p.setLoginTimeout(20);
-        try {
-            Connection c = p.getConnection();
+        try (Connection c = p.getConnection();){
             Statement s = c.createStatement();
             s.execute(dbinit);
+            c.commit();
             log.trace("Running SQL against database '{}': '{}'", dbConnectString, dbinit);
             c.close();
             log.debug("Successfully created schema for database 'Brutex' at '{}'.", dbConnectString);
         } catch (SQLException e) {
             log.error("Error creating the schema for database 'Brutex' using '{}': {}", dbConnectString, e.getMessage());
         try {
             Scheduler scheduler = (Scheduler) arg0.getServletContext().getAttribute("scheduler");
             log.debug("Active jobs to be terminated: {}", scheduler.getCurrentlyExecutingJobs());
+            JobKey key = JobKey.jobKey("ALFEmitter");
-            JobKey key = JobKey.jobKey("ALFEmitter");
+            JobKey cleanerkey = JobKey.jobKey("EventLogCleaner");
             synchronized (scheduler) {
                 if (!scheduler.isShutdown() && scheduler.checkExists(key) ) {
                     scheduler.interrupt(key);
                     scheduler.deleteJob(key);
                     log.info("Gracefully stopped the ALFEventEmitter job.");
+                }
+                if (!scheduler.isShutdown() && scheduler.checkExists(cleanerkey) ) {
+                    scheduler.interrupt(cleanerkey);
+                    scheduler.deleteJob(cleanerkey);
+                    log.info("Gracefully stopped the ALFEventEmitter job.");
+                }
                 if (!scheduler.isShutdown()) {
                     scheduler.shutdown(true);
+                }
+            }
         try {
             log.info("Create/Re-open file based database to persist memory database.");
             Connection con = fdbpool.getConnection();
             Statement s = con.createStatement();
-            final String insert = "INSERT INTO brutex.tbl_events SELECT * from LINK UNION SELECT " + "btx_event_type, btx_id, btx_obj_type, btx_obj_id, btx_timestamp, btx_event from LINK2;";
+            final String insert = "INSERT INTO brutex.tbl_events SELECT * from MEM_INBOUND UNION SELECT " + "btx_event_type, btx_id, btx_obj_type, btx_obj_id, btx_timestamp, btx_event from MEM_OUTBOUND;";
             s.execute(insert);
             int count = s.getUpdateCount();
+            log.info("Persisted {} active event rows in file-based database.", count);
+            final String save_all = "INSERT INTO brutex.tbl_events_all SELECT * from MEM_ALL_EVENTS;";
+            s.execute(save_all);
+            count = s.getUpdateCount();
+            log.info("Persisted {} event rows from all_events log in file-based database", count);
             log.info("Persisted {} rows in file-based database.", count);
             log.info("Shutting down in-memory database. Closing file-based database. Please wait ...");
+            s.execute("SHUTDOWN;");
             s.execute("SHUTDOWN;");
             con.close();
             log.info("Shutting down databases complete.");
         } catch (SQLException e) {
             log.error("An error occurred during database persistence: {}", e.getMessage());
+        }
         //Load events from file based database into in-memory database
         try {
             log.info("Start recovery of previously unsend alf events. Trying to load them into in-memory database.");
-            final String link = "CREATE LINKED TABLE IF NOT EXISTS LINK('org.h2.Driver', '"+configuration.getJdbc_memdb()+"', '', '', 'brutex.tbl_events'); " +
-                                "CREATE LINKED TABLE IF NOT EXISTS LINK2('org.h2.Driver', '"+configuration.getJdbc_memdb()+"', '', '', 'brutex.tbl_events_snap');";
+            final String link = getLinkSQL();
-            final String recoverSQL = "INSERT INTO LINK DIRECT SELECT * FROM brutex.tbl_events;";
+            final String recoverSQL = "INSERT INTO MEM_INBOUND DIRECT SELECT * FROM brutex.tbl_events;";
             final String truncate = "TRUNCATE TABLE brutex.tbl_events;";
             int count = 0;
             Connection con = fdbpool.getConnection();
             con.setAutoCommit(false);
         } catch (SQLException e) {
             log.error("Exception during recovery of events from previous runs: {}", e.getMessage());
             throw new RuntimeException(e);
+        }
         //Start initial run of the emitter
+        if(configuration.isEmitterActive()) {
-        startEmitterImmediate(egres, (Scheduler) context.getAttribute("scheduler"));
+            startEmitterImmediate(egres, (Scheduler) context.getAttribute("scheduler"));
+        }
+        //Start mem db log cleaner
+        if(configuration.getCleaner_interval()>0) {
+            startEventLogCleaner((Scheduler) context.getAttribute("scheduler"));
+        }
+    }
+    private String getLinkSQL() {
+        final String dbDriverClass = "org.h2.Driver";
+        final String link = "CREATE LINKED TABLE IF NOT EXISTS MEM_INBOUND('"+dbDriverClass+"', '"+configuration.getJdbc_memdb()+"', '', '', 'brutex.tbl_events'); " +
+                            "CREATE LINKED TABLE IF NOT EXISTS MEM_OUTBOUND('"+dbDriverClass+"', '"+configuration.getJdbc_memdb()+"', '', '', 'brutex.tbl_events_snap'); " +
+                            "CREATE LINKED TABLE IF NOT EXISTS MEM_ALL_EVENTS('"+dbDriverClass+"', '"+configuration.getJdbc_memdb()+"', '', '', 'brutex.tbl_events_all') AUTOCOMMIT OFF; " +
+                            "";
+        return link;
+    }
-    private synchronized void startEmitterImmediate(AtomicLong egres_counter, Scheduler scheduler) {
+    private synchronized void startEmitterImmediate(AtomicLong egres, Scheduler scheduler) {
         try {
             if (!scheduler.checkExists(JobKey.jobKey("ALFEmitter"))) {
                 JobDetail job2 = JobBuilder.newJob(EventEmitter.class).withIdentity("ALFEmitter").build();
                 job2.getJobDataMap().put("mdbConnection", mempool);
                 job2.getJobDataMap().put("fdbConnection", fdbpool);
                 job2.getJobDataMap().put("run_key", Instant.now().toEpochMilli());
-                job2.getJobDataMap().put("egres_counter", egres_counter);
+                job2.getJobDataMap().put("egres_counter", egres);
                 job2.getJobDataMap().put(EventmanagerConfiguration.KEY, EventmanagerConfiguration.getInstance());
                 SimpleTrigger t = (SimpleTrigger) newTrigger().withIdentity("ALFEmitter").startNow().build();
                 scheduler.scheduleJob(job2, t);
+            }
         } catch (SchedulerException ex) {
             log.error("Could not start EventEmitter to process existing queue directly after startup: {}", ex.getMessage());
+        }
+    }
+    private void startEventLogCleaner(Scheduler scheduler) {
+        try {
+            if (!scheduler.checkExists(JobKey.jobKey("EventLogCleaner"))) {
+                JobDetail job2 = JobBuilder.newJob(EventLogCleanerJob.class).withIdentity("EventLogCleaner").build();
+                job2.getJobDataMap().put("mdbConnection", mempool);
+                job2.getJobDataMap().put("fdbConnection", fdbpool);
+                job2.getJobDataMap().put("run_key", Instant.now().toEpochMilli());
+                job2.getJobDataMap().put(EventmanagerConfiguration.KEY, EventmanagerConfiguration.getInstance());
+                SimpleTrigger t = (SimpleTrigger) newTrigger().withIdentity("EventLogCleaner")
+                        .withSchedule(SimpleScheduleBuilder.simpleSchedule()
+                                .withIntervalInMinutes(configuration.getCleaner_interval())
+                                .repeatForever())
+                        .startAt(Date.from(Instant.now().plus(configuration.getCleaner_interval(), ChronoUnit.MINUTES)))
+                        .build();
+                scheduler.scheduleJob(job2, t);
+            }
+        } catch (SchedulerException ex) {
+            log.error("Could not start EventLogCleaner to process existing queue directly after startup: {}", ex.getMessage());
+        }
+    }
     private void readConfiguration(ServletContext ctx) {
                 /* Configure ServletContext attributes using configuration object*/
                 EventmanagerConfiguration c = EventmanagerConfiguration.getInstance().refreshConfig();

Subversion Repositories XServices

(root)/xservices/trunk/src/main/java/net/brutex/xservices/util/MiscServiceServletContextListener.java – Rev 201 → 203