Subversion Repositories XServices

package net.brutex.xservices.util;


import lombok.extern.slf4j.Slf4j;
import org.h2.jdbcx.JdbcConnectionPool;
import org.quartz.*;
import org.quartz.impl.StdSchedulerFactory;

import javax.servlet.ServletContext;
import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;
import javax.servlet.annotation.WebListener;
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;

//For Servlet container 3.x, you can annotate the listener with @WebListener, no need to declares in web.xml.

/**
 * Handle servlet lifecycle actions for the MiscService servlet, such as
 * initializing in-memory database, persist on shutdown etc.
 */


@WebListener
@Slf4j
public class MiscServiceServletContextListener implements ServletContextListener {

    /**
     * SQL initialization for in-memory database
     * INIT=RUNSCRIPT FROM 'classpath:scripts/create.sql'"
     */
    private final static String dbinit = "RUNSCRIPT FROM 'classpath:ddl/BRTX_schema.ddl';";

    private final EventmanagerConfiguration configuration = EventmanagerConfiguration.getInstance().refreshConfig();
    private final JdbcConnectionPool mempool = getDbPool(configuration.getJdbc_memdb());
    private final JdbcConnectionPool fdbpool = getDbPool(configuration.getJdbc_filedb());

    /**
     * Create DB connection pool and initialize the in-memory database with schema.
     *
     * @return connection pool
     */
    private static JdbcConnectionPool getDbPool(String dbConnectString) {
        JdbcConnectionPool p = JdbcConnectionPool.create(dbConnectString, "", "");
        p.setMaxConnections(256);
        p.setLoginTimeout(20);
        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());
            throw new RuntimeException(e);
        }
        return p;
    }

    @Override
    public void contextDestroyed(ServletContextEvent arg0) {
        log.trace("contextDestroyed called.");
        try {
            Scheduler scheduler = (Scheduler) arg0.getServletContext().getAttribute("scheduler");
            log.debug("Active jobs to be terminated: {}", scheduler.getCurrentlyExecutingJobs());

            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);
                }
            }
        } catch (SchedulerException e) {
            log.error("Failed to stop the ALFEmitter job: {}", e.getMessage());
            throw new RuntimeException(e);
        }

        log.info("ServletContextListener destroyed. Saving in-memory database to file based database.");
        int act_i = mempool.getActiveConnections();
        if (act_i > 0) {
            log.warn("There are still {} connections to the XServices in-memory database active.", act_i);
        }

        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 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("Shutting down in-memory database. Closing file-based database. Please wait ...");
            s.execute("SHUTDOWN;");

            con.close();
            log.info("Shutting down databases complete.");
        } catch (SQLException e) {
            log.error("An error occurred during database persistence: {}", e.getMessage());
            throw new RuntimeException(e);
        }
        log.debug("Handled {} egress events.", arg0.getServletContext().getAttribute("egres_counter"));
        log.debug("Handled {} ingress events.", arg0.getServletContext().getAttribute("ingres_counter"));
    }

    //Run this before web application is started
    @Override
    public void contextInitialized(ServletContextEvent arg0) {
        log.debug("ServletContextListener started");
        ServletContext context = arg0.getServletContext();
        readConfiguration(context);

        context.setAttribute("mdbConnection", mempool);
        context.setAttribute("fdbConnection", fdbpool);
        context.setAttribute("ingres_counter", 0);
        AtomicLong egres = new AtomicLong(0);
        context.setAttribute("egres_counter", egres);
        context.setAttribute("ingres_counter", new AtomicLong(0));
        try {
            StdSchedulerFactory fact = new StdSchedulerFactory();
            fact.initialize("MiscServicesScheduler-quartz.properties");
            Scheduler scheduler = fact.getScheduler();
            scheduler.start();
            context.setAttribute("scheduler", scheduler);
        } catch (SchedulerException e) {
            log.error("Error creating scheduler within ServletContext: {}", e.getMessage());
            throw new RuntimeException(e);
        }

        //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 = getLinkSQL();
            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);
            Statement statement = con.createStatement();
            statement.execute(link);
            con.commit();
            ResultSet rs = statement.executeQuery("SELECT COUNT(1) FROM brutex.tbl_events");
            if(rs.next()) count = rs.getInt(1);
            statement.execute(recoverSQL);
            log.info("Recovered {} events and loaded them into in-memory database.", count);
            statement.execute(truncate);
            con.commit();
            con.close();
        } 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"));
        }

        //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, 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);
                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();
                ctx.setAttribute(EventmanagerConfiguration.KEY, c);
    }

}