Subversion Repositories XServices

package net.brutex.xservices.util;

import lombok.extern.slf4j.Slf4j;

import org.apache.commons.io.IOUtils;

import org.h2.jdbcx.JdbcConnectionPool;

import org.quartz.*;

import java.io.IOException;

import java.io.Reader;

import java.io.StringReader;

import java.sql.*;

import java.time.Instant;

import java.util.concurrent.atomic.AtomicBoolean;

import java.util.concurrent.atomic.AtomicLong;

@Slf4j

@DisallowConcurrentExecution

public class EventLogCleanerJob implements Job, InterruptableJob {

    private final AtomicBoolean isInterrupted = new AtomicBoolean(false);

    @Override

    public void execute(JobExecutionContext context) throws JobExecutionException {

        final Instant d = Instant.now();

        final long ts = d.toEpochMilli();

        log.info("EventLogCleaner is executing now.");

        final EventmanagerConfiguration conf = (EventmanagerConfiguration) context.getMergedJobDataMap()

                .get(EventmanagerConfiguration.KEY);

        final JdbcConnectionPool fpool = (JdbcConnectionPool) context.getMergedJobDataMap().get("fdbConnection");

        final JdbcConnectionPool mpool = (JdbcConnectionPool) context.getMergedJobDataMap().get("mdbConnection");

        final String moveSQL = "INSERT INTO brutex.tbl_events_all DIRECT SELECT * FROM MEM_ALL_EVENTS " +

                "where btx_timestamp < " + (ts-5000) + " ";

        final String deleteTable = "DELETE FROM MEM_ALL_EVENTS where btx_timestamp < " + (ts-5000);

        final String deleteMemTable = "DELETE FROM brutex.tbl_events_all where btx_timestamp < " + (ts-5000);

        /**

         * Move event table data to all events log

         */

        try (Connection fcon = fpool.getConnection();

            Connection mcon = mpool.getConnection()){

            fcon.setTransactionIsolation(Connection.TRANSACTION_SERIALIZABLE);

            fcon.setAutoCommit(false);

            Statement stmt = fcon.createStatement();

            stmt.execute(moveSQL);

            int count = stmt.getUpdateCount();

            fcon.commit();

            Statement mstm = mcon.createStatement();

            mstm.execute(deleteMemTable);

            int count2 = mstm.getUpdateCount();

            mcon.commit();

            log.info("EventLogCleaner moved '{}/ deleted {}' events into the persistence space.", count, count2);

        } catch (SQLException e) {

            log.error("Exception in SQL execution: {}", e.getMessage());

            throw new JobExecutionException(e);

        }

    }

        /**

     * <p>

     * Called by the <code>{@link Scheduler}</code> when a user

     * interrupts the <code>Job</code>.

     * </p>

     *

     * @throws UnableToInterruptJobException if there is an exception while interrupting the job.

     */

    @Override

    public synchronized void interrupt() throws UnableToInterruptJobException {

        isInterrupted.set(true);

        log.warn("EventLogCleaner received and interrupt.");

        Thread.currentThread().interrupt();

    }

}