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.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(16);

        p.setLoginTimeout(5);

        try {

            Connection c = p.getConnection();

            Statement s = c.createStatement();

            s.execute(dbinit);

            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");

            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.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 LINK UNION SELECT " + "btx_event_type, btx_id, btx_obj_type, btx_obj_id, btx_timestamp, btx_event from LINK2;";

            s.execute(insert);

            int count = s.getUpdateCount();

            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;");

            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 = "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 recoverSQL = "INSERT INTO LINK 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

        startEmitterImmediate(egres, (Scheduler) context.getAttribute("scheduler"));

    }

    private synchronized void startEmitterImmediate(AtomicLong egres_counter, 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(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 readConfiguration(ServletContext ctx) {

                /* Configure ServletContext attributes using configuration object*/

                EventmanagerConfiguration c = EventmanagerConfiguration.getInstance().refreshConfig();

                ctx.setAttribute(EventmanagerConfiguration.KEY, c);

    }

}