/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.openejb.core.timer;

import org.apache.openejb.ApplicationException;
import org.apache.openejb.BeanContext;
import org.apache.openejb.InterfaceType;
import org.apache.openejb.OpenEJBException;
import org.apache.openejb.OpenEJBRuntimeException;
import org.apache.openejb.RpcContainer;
import org.apache.openejb.core.BaseContext;
import org.apache.openejb.core.timer.quartz.PatchedStdJDBCDelegate;
import org.apache.openejb.core.transaction.TransactionType;
import org.apache.openejb.loader.SystemInstance;
import org.apache.openejb.monitoring.LocalMBeanServer;
import org.apache.openejb.quartz.JobBuilder;
import org.apache.openejb.quartz.JobDataMap;
import org.apache.openejb.quartz.JobDetail;
import org.apache.openejb.quartz.Scheduler;
import org.apache.openejb.quartz.SchedulerException;
import org.apache.openejb.quartz.Trigger;
import org.apache.openejb.quartz.TriggerKey;
import org.apache.openejb.quartz.impl.StdSchedulerFactory;
import org.apache.openejb.quartz.impl.jdbcjobstore.JobStoreSupport;
import org.apache.openejb.quartz.impl.jdbcjobstore.StdJDBCDelegate;
import org.apache.openejb.quartz.impl.triggers.AbstractTrigger;
import org.apache.openejb.quartz.listeners.SchedulerListenerSupport;
import org.apache.openejb.quartz.simpl.RAMJobStore;
import org.apache.openejb.resource.quartz.QuartzResourceAdapter;
import org.apache.openejb.spi.ContainerSystem;
import org.apache.openejb.util.LogCategory;
import org.apache.openejb.util.Logger;
import org.apache.openejb.util.SetAccessible;

import javax.ejb.EJBContext;
import javax.ejb.EJBException;
import javax.ejb.ScheduleExpression;
import javax.ejb.Timer;
import javax.ejb.TimerConfig;
import javax.transaction.Status;
import javax.transaction.SystemException;
import javax.transaction.TransactionManager;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

public class EjbTimerServiceImpl implements EjbTimerService, Serializable {

    private static final long serialVersionUID = 1L;
    private static final Logger log = Logger.getInstance(LogCategory.TIMER, "org.apache.openejb.util.resources");

    public static final String QUARTZ_JMX = "org.apache.openejb.quartz.scheduler.jmx.export";
    public static final String QUARTZ_MAKE_SCHEDULER_THREAD_DAEMON = "org.apache.openejb.quartz.scheduler.makeSchedulerThreadDaemon";

    public static final String OPENEJB_TIMEOUT_JOB_NAME = "OPENEJB_TIMEOUT_JOB";
    public static final String OPENEJB_TIMEOUT_JOB_GROUP_NAME = "OPENEJB_TIMEOUT_GROUP";

    public static final String EJB_TIMER_RETRY_ATTEMPTS = "EjbTimer.RetryAttempts";
    public static final String OPENEJB_QUARTZ_USE_TCCL = "openejb.quartz.use-TCCL";

    private boolean transacted;
    private int retryAttempts;

    private transient TransactionManager transactionManager;
    private transient BeanContext deployment;
    private transient TimerStore timerStore;
    private transient Scheduler scheduler;

    public EjbTimerServiceImpl(final BeanContext deployment, final TimerStore timerStore) {
        this(deployment, getDefaultTransactionManager(), timerStore, -1);
        log.isDebugEnabled(); // touch logger to force it to be initialized
    }

    public static TransactionManager getDefaultTransactionManager() {
        return SystemInstance.get().getComponent(TransactionManager.class);
    }

    public EjbTimerServiceImpl(final BeanContext deployment, final TransactionManager transactionManager, final TimerStore timerStore, final int retryAttempts) {
        this.deployment = deployment;
        this.transactionManager = transactionManager;
        this.timerStore = timerStore;
        final TransactionType transactionType = deployment.getTransactionType(deployment.getEjbTimeout());
        this.transacted = transactionType == TransactionType.Required || transactionType == TransactionType.RequiresNew;
        this.retryAttempts = retryAttempts;
        if (retryAttempts < 0) {
            this.retryAttempts = deployment.getOptions().get(EJB_TIMER_RETRY_ATTEMPTS, 1);
        }
    }

    private void writeObject(final ObjectOutputStream out) throws IOException {
        out.writeUTF(deployment.getDeploymentID().toString());
        out.writeBoolean(transacted);
        out.writeInt(retryAttempts);
    }

    private void readObject(final ObjectInputStream in) throws IOException {
        final String dId = in.readUTF();
        transacted = in.readBoolean();
        retryAttempts = in.readInt();

        deployment = SystemInstance.get().getComponent(ContainerSystem.class).getBeanContext(dId);
        transactionManager = getDefaultTransactionManager();
        timerStore = deployment.getEjbTimerService().getTimerStore();
        scheduler = (Scheduler) Proxy.newProxyInstance(deployment.getClassLoader(), new Class<?>[]{Scheduler.class}, new LazyScheduler(deployment));
    }

    public static synchronized Scheduler getDefaultScheduler(final BeanContext deployment) {
        Scheduler scheduler = deployment.get(Scheduler.class);
        if (scheduler != null) {
            boolean valid;
            try {
                valid = !scheduler.isShutdown();
            } catch (final Exception ignored) {
                valid = false;
            }
            if (valid) {
                return scheduler;
            }
        }

        Scheduler thisScheduler;
        synchronized (deployment.getId()) { // should be done only once so no perf issues
            scheduler = deployment.get(Scheduler.class);
            if (scheduler != null) {
                return scheduler;
            }

            final Properties properties = new Properties();
            int quartzProps = 0;
            quartzProps += putAll(properties, SystemInstance.get().getProperties());
            quartzProps += putAll(properties, deployment.getModuleContext().getAppContext().getProperties());
            quartzProps += putAll(properties, deployment.getModuleContext().getProperties());
            quartzProps += putAll(properties, deployment.getProperties());

            // custom config -> don't use default/global scheduler
            // if one day we want to keep a global config for a global scheduler (SystemInstance.get().getProperties()) we'll need to manage resume/pause etc correctly by app
            // since we have a scheduler by ejb today in such a case we don't need
            final boolean newInstance = quartzProps > 0;

            final SystemInstance systemInstance = SystemInstance.get();

            scheduler = systemInstance.getComponent(Scheduler.class);

            if (scheduler == null || newInstance) {
                final boolean useTccl = "true".equalsIgnoreCase(properties.getProperty(OPENEJB_QUARTZ_USE_TCCL, "false"));

                defaultQuartzConfiguration(properties, deployment, newInstance, useTccl);

                try {
                    // start in container context to avoid thread leaks
                    final ClassLoader oldCl = Thread.currentThread().getContextClassLoader();
                    if (useTccl) {
                        Thread.currentThread().setContextClassLoader(deployment.getClassLoader());
                    } else {
                        Thread.currentThread().setContextClassLoader(EjbTimerServiceImpl.class.getClassLoader());
                    }
                    try {
                        thisScheduler = new StdSchedulerFactory(properties).getScheduler();
                        thisScheduler.start();
                    } finally {
                        Thread.currentThread().setContextClassLoader(oldCl);
                    }

                    //durability is configured with true, which means that the job will be kept in the store even if no trigger is attached to it.
                    //Currently, all the EJB beans share with the same job instance
                    final JobDetail job = JobBuilder.newJob(EjbTimeoutJob.class)
                        .withIdentity(OPENEJB_TIMEOUT_JOB_NAME, OPENEJB_TIMEOUT_JOB_GROUP_NAME)
                        .storeDurably(true)
                        .requestRecovery(false)
                        .build();
                    thisScheduler.addJob(job, true);
                } catch (final SchedulerException e) {
                    throw new OpenEJBRuntimeException("Fail to initialize the default scheduler", e);
                }

                if (!newInstance) {
                    systemInstance.setComponent(Scheduler.class, thisScheduler);
                }
            } else {
                thisScheduler = scheduler;
            }

            deployment.set(Scheduler.class, thisScheduler);
        }

        return thisScheduler;
    }

    private static void defaultQuartzConfiguration(final Properties properties, final BeanContext deployment, final boolean newInstance, final boolean tccl) {
        final String defaultThreadPool = DefaultTimerThreadPoolAdapter.class.getName();
        if (!properties.containsKey(StdSchedulerFactory.PROP_THREAD_POOL_CLASS)) {
            properties.put(StdSchedulerFactory.PROP_THREAD_POOL_CLASS, defaultThreadPool);
        }
        if (!properties.containsKey(StdSchedulerFactory.PROP_SCHED_INSTANCE_NAME)) {
            properties.put(StdSchedulerFactory.PROP_SCHED_INSTANCE_NAME, "OpenEJB-TimerService-Scheduler");
        }
        if (!properties.containsKey(StdSchedulerFactory.PROP_SCHED_SKIP_UPDATE_CHECK)) {
            properties.put(StdSchedulerFactory.PROP_SCHED_SKIP_UPDATE_CHECK, "true");
        }
        if (!properties.containsKey("org.terracotta.quartz.skipUpdateCheck")) {
            properties.put("org.terracotta.quartz.skipUpdateCheck", "true");
        }
        if (!properties.containsKey(StdSchedulerFactory.PROP_SCHED_INTERRUPT_JOBS_ON_SHUTDOWN)) {
            properties.put(StdSchedulerFactory.PROP_SCHED_INTERRUPT_JOBS_ON_SHUTDOWN, "true");
        }
        if (!properties.containsKey(StdSchedulerFactory.PROP_SCHED_INTERRUPT_JOBS_ON_SHUTDOWN_WITH_WAIT)) {
            properties.put(StdSchedulerFactory.PROP_SCHED_INTERRUPT_JOBS_ON_SHUTDOWN_WITH_WAIT, "true");
        }
        if (!properties.containsKey(QUARTZ_MAKE_SCHEDULER_THREAD_DAEMON)) {
            properties.put(QUARTZ_MAKE_SCHEDULER_THREAD_DAEMON, "true");
        }
        if (!properties.containsKey(QUARTZ_JMX) && LocalMBeanServer.isJMXActive()) {
            properties.put(QUARTZ_JMX, "true");
        }
        if (!properties.containsKey(StdSchedulerFactory.PROP_SCHED_INSTANCE_ID)) {
            if (!newInstance) {
                properties.setProperty(StdSchedulerFactory.PROP_SCHED_INSTANCE_ID, "OpenEJB");
            } else {
                properties.setProperty(StdSchedulerFactory.PROP_SCHED_INSTANCE_ID, deployment.getDeploymentID().toString());
            }
        }

        if (!tccl) {
            final String driverDelegate = properties.getProperty("org.apache.openejb.quartz.jobStore.driverDelegateClass");
            if (driverDelegate != null && StdJDBCDelegate.class.getName().equals(driverDelegate)) {
                properties.put("org.apache.openejb.quartz.jobStore.driverDelegateClass", PatchedStdJDBCDelegate.class.getName());
            } else if (driverDelegate != null) {
                log.info("You use " + driverDelegate + " driver delegate with quartz, ensure it doesn't use ObjectInputStream otherwise your custom TimerData can induce some issues");
            }

            // adding our custom persister
            if (properties.containsKey("org.apache.openejb.quartz.jobStore.class") && !properties.containsKey("org.apache.openejb.quartz.jobStore.driverDelegateInitString")) {
                try {
                    final Class<?> clazz = EjbTimerServiceImpl.class.getClassLoader().loadClass(properties.getProperty("org.apache.openejb.quartz.jobStore.class"));
                    if (JobStoreSupport.class.isAssignableFrom(clazz)) {
                        properties.put("org.apache.openejb.quartz.jobStore.driverDelegateInitString",
                            "triggerPersistenceDelegateClasses=" + EJBCronTriggerPersistenceDelegate.class.getName());
                    }
                } catch (final Throwable th) {
                    // no-op
                }
            }
        }

        if (defaultThreadPool.equals(properties.get(StdSchedulerFactory.PROP_THREAD_POOL_CLASS))) {
            if (properties.containsKey("org.apache.openejb.quartz.threadPool.threadCount")
                && !properties.containsKey(DefaultTimerThreadPoolAdapter.OPENEJB_TIMER_POOL_SIZE)) {
                log.info("Found property 'org.apache.openejb.quartz.threadPool.threadCount' for default thread pool, please use '"
                    + DefaultTimerThreadPoolAdapter.OPENEJB_TIMER_POOL_SIZE + "' instead");
                properties.put(DefaultTimerThreadPoolAdapter.OPENEJB_TIMER_POOL_SIZE, properties.getProperty("org.apache.openejb.quartz.threadPool.threadCount"));
            }
            if (properties.containsKey("org.quartz.threadPool.threadCount")
                && !properties.containsKey(DefaultTimerThreadPoolAdapter.OPENEJB_TIMER_POOL_SIZE)) {
                log.info("Found property 'org.quartz.threadPool.threadCount' for default thread pool, please use '"
                    + DefaultTimerThreadPoolAdapter.OPENEJB_TIMER_POOL_SIZE + "' instead");
                properties.put(DefaultTimerThreadPoolAdapter.OPENEJB_TIMER_POOL_SIZE, properties.getProperty("org.quartz.threadPool.threadCount"));
            }
        }

        // to ensure we can shutdown correctly, default doesn't support such a configuration
        if (!properties.getProperty(StdSchedulerFactory.PROP_JOB_STORE_CLASS, RAMJobStore.class.getName()).equals(RAMJobStore.class.getName())) {
            properties.put("org.apache.openejb.quartz.jobStore.makeThreadsDaemons", properties.getProperty("org.apache.openejb.quartz.jobStore.makeThreadsDaemon", "true"));
        }
    }

    private static int putAll(final Properties a, final Properties b) {
        int number = 0;
        for (final Map.Entry<Object, Object> entry : b.entrySet()) {
            final String key = entry.getKey().toString();
            if (key.startsWith("org.quartz.")
                || key.startsWith("org.apache.openejb.quartz.")
                || key.startsWith("openejb.quartz.")
                || DefaultTimerThreadPoolAdapter.OPENEJB_TIMER_POOL_SIZE.equals(key)
                || "org.terracotta.quartz.skipUpdateCheck".equals(key)) {
                number++;
            }

            final Object value = entry.getValue();
            if (String.class.isInstance(value)) {
                if (!key.startsWith("org.quartz")) {
                    a.put(key, value);
                } else {
                    a.put("org.apache.openejb.quartz" + key.substring("org.quartz".length()), value);
                }
            }
        }
        return number;
    }

    @Override
    public void stop() {
        cleanTimerData();
        shutdownMyScheduler();
    }

    private void cleanTimerData() {
        if (timerStore == null || scheduler == null || deployment == null) {
            return;
        }

        final Collection<TimerData> timerDatas = timerStore.getTimers(deployment.getDeploymentID().toString());
        if (timerDatas == null) {
            return;
        }

        for (final TimerData data : timerDatas) {
            final Trigger trigger = data.getTrigger();
            if (trigger == null) {
                continue;
            }

            final TriggerKey key = trigger.getKey();
            try {
                data.stop();
            } catch (final EJBException ignored) {
                log.warning("An error occured deleting trigger '" + key + "' on bean " + deployment.getDeploymentID());
            }
        }
    }

    private void shutdownMyScheduler() {
        if (scheduler == null) {
            return;
        }

        boolean defaultScheduler = false;
        final Scheduler ds = SystemInstance.get().getComponent(Scheduler.class);
        try { // == is the faster way to test, we rely on name (key in quartz registry) only for serialization
            defaultScheduler = ds == scheduler || scheduler.getSchedulerName().equals(ds.getSchedulerName());
        } catch (final Exception e) {
            // no-op: default should be fine
        }

        // if specific instance
        if (!defaultScheduler) {
            shutdown(scheduler);
        }
    }

    public static void shutdown() {
        shutdown(SystemInstance.get().getComponent(Scheduler.class));
    }

    private static void shutdown(final Scheduler s) throws OpenEJBRuntimeException {

        try {
            if (null != s && !s.isShutdown() && s.isStarted()) {

                try {
                    s.pauseAll();
                } catch (final SchedulerException e) {
                    // no-op
                }

                long timeout = SystemInstance.get().getOptions().get(QuartzResourceAdapter.OPENEJB_QUARTZ_TIMEOUT, 10000L);

                if (timeout < 1000L) {
                    timeout = 1000L;
                }

                final CountDownLatch shutdownWait = new CountDownLatch(1);
                final AtomicReference<Throwable> ex = new AtomicReference<>();

                String n = "Unknown";
                try {
                    n = s.getSchedulerName();
                } catch (final SchedulerException e) {
                    log.warning("EjbTimerService scheduler has no name");
                }

                final String name = n;

                Thread stopThread = new Thread(name + " shutdown wait") {

                    @Override
                    public void run() {
                        try {
                            s.getListenerManager().addSchedulerListener(new SchedulerListenerSupport() {
                                @Override
                                public void schedulerShutdown() {
                                    shutdownWait.countDown();
                                }
                            });

                            //Shutdown, but give running jobs a chance to complete.
                            //User scheduled jobs should really implement InterruptableJob
                            s.shutdown(true);
                        } catch (final Throwable e) {
                            ex.set(e);
                            shutdownWait.countDown();
                        }
                    }
                };

                stopThread.setDaemon(true);
                stopThread.start();

                boolean stopped = false;
                try {
                    stopped = shutdownWait.await(timeout, TimeUnit.MILLISECONDS);
                } catch (final InterruptedException e) {
                    //Ignore
                }

                try {
                    if (!stopped || !s.isShutdown()) {

                        stopThread = new Thread(name + " shutdown forced") {

                            @Override
                            public void run() {
                                try {
                                    //Force a shutdown without waiting for jobs to complete.
                                    s.shutdown(false);
                                    log.warning("Forced " + name + " shutdown - Jobs may be incomplete");
                                } catch (final Throwable e) {
                                    ex.set(e);
                                }
                            }
                        };

                        stopThread.setDaemon(true);
                        stopThread.start();

                        try {
                            //Give the forced shutdown a chance to complete
                            stopThread.join(timeout);
                        } catch (final InterruptedException e) {
                            //Ignore
                        }
                    }
                } catch (final Throwable e) {
                    ex.set(e);
                }

                final Throwable t = ex.get();
                if (null != t) {
                    throw new OpenEJBRuntimeException("Unable to shutdown " + name + " scheduler", t);
                }
            }
        } catch (final SchedulerException e) {
            //Ignore - This can only be a shutdown issue that we have no control over.
        }
    }

    @Override
    public void start() throws TimerStoreException {

        if (isStarted()) {
            return;
        }

        scheduler = getDefaultScheduler(deployment);

        // load saved timers
        final Collection<TimerData> timerDatas = timerStore.loadTimers(this, (String) deployment.getDeploymentID());
        // schedule the saved timers
        for (final TimerData timerData : timerDatas) {
            initializeNewTimer(timerData);
        }
    }

    public TransactionManager getTransactionManager() {
        return transactionManager;
    }

    /**
     * Called from TimerData and start when a timer should be scheduled with the java.util.Timer.
     *
     * @param timerData the timer to schedule
     */
    public void schedule(final TimerData timerData) throws TimerStoreException {

        start();

        if (scheduler == null) {
            throw new TimerStoreException("Scheduler is not configured properly");
        }

        timerData.setScheduler(scheduler);

        final Trigger trigger = timerData.getTrigger();

        if (null == trigger) {

            try {
                if (!scheduler.isShutdown()) {
                    log.warning("Failed to schedule: " + timerData.getInfo());
                }
            } catch (final SchedulerException e) {
                //Ignore
            }
        }

        final AbstractTrigger<?> atrigger;
        if (trigger instanceof AbstractTrigger) { // is the case
            atrigger = (AbstractTrigger<?>) trigger;
            atrigger.setJobName(OPENEJB_TIMEOUT_JOB_NAME);
            atrigger.setJobGroup(OPENEJB_TIMEOUT_JOB_GROUP_NAME);
        } else {
            throw new OpenEJBRuntimeException("the trigger was not an AbstractTrigger - Should not be possible: " + trigger);
        }

        final JobDataMap triggerDataMap = trigger.getJobDataMap();
        triggerDataMap.put(EjbTimeoutJob.EJB_TIMERS_SERVICE, this);
        triggerDataMap.put(EjbTimeoutJob.TIMER_DATA, timerData);

        try {
            final TriggerKey triggerKey = new TriggerKey(atrigger.getName(), atrigger.getGroup());
            if (!scheduler.checkExists(triggerKey)) {
                scheduler.scheduleJob(trigger);
            } else if (Trigger.TriggerState.PAUSED.equals(scheduler.getTriggerState(triggerKey))) { // redeployment
                // more consistent in the semantic than a resume but resume would maybe be more relevant here
                scheduler.unscheduleJob(triggerKey);
                scheduler.scheduleJob(trigger);
            }
        } catch (final Exception e) {
            //TODO Any other actions we could do ?
            log.error("Could not schedule timer " + timerData, e);
        }
    }

    /**
     * Call back from TimerData and ejbTimeout when a timer has been cancelled (or is complete) and should be removed from stores.
     *
     * @param timerData the timer that was cancelled
     */
    public void cancelled(final TimerData timerData) {
        // make sure it was removed from the store
        timerStore.removeTimer(timerData.getId());
    }

    /**
     * Returns a timerData to the TimerStore, if a cancel() is rolled back.
     *
     * @param timerData the timer to be returned to the timer store
     */
    public void addTimerData(final TimerData timerData) {
        try {
            timerStore.addTimerData(timerData);
        } catch (final Exception e) {
            log.warning("Could not add timer of type " + timerData.getType().name() + " due to " + e.getMessage());
        }
    }

    @Override
    public Timer getTimer(final long timerId) {
        final TimerData timerData = timerStore.getTimer((String) deployment.getDeploymentID(), timerId);
        if (timerData != null) {
            return timerData.getTimer();
        } else {
            return null;
        }
    }

    @Override
    public Collection<Timer> getTimers(final Object primaryKey) throws IllegalStateException {
        checkState();

        final Collection<Timer> timers = new ArrayList<>();
        for (final TimerData timerData : timerStore.getTimers((String) deployment.getDeploymentID())) {
            // if (!CalendarTimerData.class.isInstance(timerData) || !CalendarTimerData.class.cast(timerData).isAutoCreated()) {
                timers.add(timerData.getTimer());
            // }
        }
        return timers;
    }

    @Override
    public Timer createTimer(final Object primaryKey,
                             final Method timeoutMethod,
                             final long duration,
                             final TimerConfig timerConfig) throws IllegalArgumentException, IllegalStateException, EJBException {
        if (duration < 0) {
            throw new IllegalArgumentException("duration is negative: " + duration);
        }
        checkState();

        final Date expiration = new Date(System.currentTimeMillis() + duration);
        try {
            final TimerData timerData = timerStore.createSingleActionTimer(this, (String) deployment.getDeploymentID(), primaryKey, timeoutMethod, expiration, timerConfig);
            initializeNewTimer(timerData);
            return timerData.getTimer();
        } catch (final TimerStoreException e) {
            throw new EJBException(e);
        }
    }

    @Override
    public Timer createTimer(final Object primaryKey,
                             final Method timeoutMethod,
                             final long initialDuration,
                             final long intervalDuration,
                             final TimerConfig timerConfig) throws IllegalArgumentException, IllegalStateException, EJBException {
        if (initialDuration < 0) {
            throw new IllegalArgumentException("initialDuration is negative: " + initialDuration);
        }
        if (intervalDuration < 0) {
            throw new IllegalArgumentException("intervalDuration is negative: " + intervalDuration);
        }
        checkState();

        final Date initialExpiration = new Date(System.currentTimeMillis() + initialDuration);
        try {
            final TimerData timerData = timerStore.createIntervalTimer(this,
                (String) deployment.getDeploymentID(),
                primaryKey,
                timeoutMethod,
                initialExpiration,
                intervalDuration,
                timerConfig);
            initializeNewTimer(timerData);
            return timerData.getTimer();
        } catch (final TimerStoreException e) {
            throw new EJBException(e);
        }
    }

    @Override
    public Timer createTimer(final Object primaryKey,
                             final Method timeoutMethod,
                             final Date expiration,
                             final TimerConfig timerConfig) throws IllegalArgumentException, IllegalStateException, EJBException {
        if (expiration == null) {
            throw new IllegalArgumentException("expiration is null");
        }
        if (expiration.getTime() < 0) {
            throw new IllegalArgumentException("expiration is negative: " + expiration.getTime());
        }
        checkState();

        try {
            final TimerData timerData = timerStore.createSingleActionTimer(this, (String) deployment.getDeploymentID(), primaryKey, timeoutMethod, expiration, timerConfig);
            initializeNewTimer(timerData);
            return timerData.getTimer();
        } catch (final TimerStoreException e) {
            throw new EJBException(e);
        }
    }

    @Override
    public Timer createTimer(final Object primaryKey,
                             final Method timeoutMethod,
                             final Date initialExpiration,
                             final long intervalDuration,
                             final TimerConfig timerConfig) throws IllegalArgumentException, IllegalStateException, EJBException {
        if (initialExpiration == null) {
            throw new IllegalArgumentException("initialExpiration is null");
        }
        if (initialExpiration.getTime() < 0) {
            throw new IllegalArgumentException("initialExpiration is negative: " + initialExpiration.getTime());
        }
        if (intervalDuration < 0) {
            throw new IllegalArgumentException("intervalDuration is negative: " + intervalDuration);
        }
        checkState();

        try {
            final TimerData timerData = timerStore.createIntervalTimer(this,
                (String) deployment.getDeploymentID(),
                primaryKey,
                timeoutMethod,
                initialExpiration,
                intervalDuration,
                timerConfig);
            initializeNewTimer(timerData);
            return timerData.getTimer();
        } catch (final TimerStoreException e) {
            throw new EJBException(e);
        }
    }

    @Override
    public Timer createTimer(final Object primaryKey, final Method timeoutMethod, final ScheduleExpression scheduleExpression, final TimerConfig timerConfig) {
        if (scheduleExpression == null) {
            throw new IllegalArgumentException("scheduleExpression is null");
        }
        //TODO add more schedule expression validation logic ?
        checkState();
        try {
            final TimerData timerData = timerStore.createCalendarTimer(this,
                (String) deployment.getDeploymentID(),
                primaryKey,
                timeoutMethod,
                scheduleExpression,
                timerConfig,
                false);
            initializeNewTimer(timerData);
            return timerData.getTimer();
        } catch (final TimerStoreException e) {
            throw new EJBException(e);
        }
    }

    @Override
    public TimerStore getTimerStore() {
        return timerStore;
    }

    @Override
    public boolean isStarted() {
        return scheduler != null;
    }

    public Scheduler getScheduler() {
        return scheduler;
    }

    private void initializeNewTimer(final TimerData timerData) {
        // mark this as a new timer... when the transaction completes it will schedule the timer
        timerData.newTimer();
    }

    /**
     * Insure that timer methods can be invoked for the current operation on this Context.
     */
    private void checkState() throws IllegalStateException {
        final BaseContext context = (BaseContext) deployment.get(EJBContext.class);
        context.doCheck(BaseContext.Call.timerMethod);
    }

    /**
     * This method calls the ejbTimeout method and starts a transaction if the timeout is transacted.
     *
     * This method will retry failed ejbTimeout calls until retryAttempts is exceeded.
     *
     * @param timerData the timer to call.
     */
    @SuppressWarnings("ReturnInsideFinallyBlock")
    public void ejbTimeout(final TimerData timerData) {
        final Thread thread = Thread.currentThread();
        final ClassLoader loader = thread.getContextClassLoader(); // container loader
        try {
            Timer timer = getTimer(timerData.getId());
            // quartz can be backed by some advanced config (jdbc for instance)
            if (timer == null && timerStore instanceof MemoryTimerStore && timerData.getTimer() != null) {
                try {
                    timerStore.addTimerData(timerData);
                    timer = timerData.getTimer(); // TODO: replace memoryjobstore by the db one?
                } catch (final TimerStoreException e) {
                    // shouldn't occur
                }
                // return;
            }
            for (int tries = 0; tries < 1 + retryAttempts; tries++) {
                boolean retry = false;
                // if transacted, begin the transaction
                if (transacted) {
                    try {
                        transactionManager.begin();
                    } catch (final Exception e) {
                        log.warning("Exception occured while starting container transaction", e);
                        return;
                    }
                }
                // call the timeout method
                try {
                    final RpcContainer container = (RpcContainer) deployment.getContainer();
                    if (container == null) {
                        return;
                    }

                    final Method ejbTimeout = timerData.getTimeoutMethod();
                    if (ejbTimeout == null) {
                        return;
                    }

                    // if app registered Synchronization we need it for commit()/rollback()
                    // so forcing it and not relying on container for it
                    thread.setContextClassLoader(deployment.getClassLoader() != null ? deployment.getClassLoader() : loader);

                    SetAccessible.on(ejbTimeout);
                    container.invoke(deployment.getDeploymentID(),
                        InterfaceType.TIMEOUT,
                        ejbTimeout.getDeclaringClass(),
                        ejbTimeout,
                        new Object[]{timer},
                        timerData.getPrimaryKey());
                } catch (final RuntimeException e) {
                    retry = true;
                    // exception from a timer does not necessairly mean failure
                    log.warning("RuntimeException from ejbTimeout on " + deployment.getDeploymentID(), e);
                    try {
                        transactionManager.setRollbackOnly();
                    } catch (final SystemException e1) {
                        log.warning("Exception occured while setting RollbackOnly for container transaction", e1);
                    }
                } catch (final OpenEJBException e) {
                    retry = true;
                    if (ApplicationException.class.isInstance(e)) { // we don't want to pollute logs
                        log.debug("Exception from ejbTimeout on " + deployment.getDeploymentID(), e);
                    } else {
                        log.warning("Exception from ejbTimeout on " + deployment.getDeploymentID(), e);
                    }
                    if (transacted) {
                        try {
                            transactionManager.setRollbackOnly();
                        } catch (final SystemException e1) {
                            log.warning("Exception occured while setting RollbackOnly for container transaction", e1);
                        }
                    }
                } finally {
                    try {
                        if (!transacted) {
                            if (!retry) {
                                return;
                            }
                        } else if (transactionManager.getStatus() == Status.STATUS_ACTIVE) {
                            transactionManager.commit();
                            return;
                        } else {
                            // tx was marked rollback, so roll it back and retry.
                            transactionManager.rollback();
                        }
                    } catch (final Exception e) {
                        log.warning("Exception occured while completing container transaction", e);
                    }
                }
            }
            log.warning("Failed to execute ejbTimeout on " + timerData.getDeploymentId() + " successfully within " + retryAttempts + " attempts");
        } catch (final RuntimeException e) {
            log.warning("RuntimeException occured while calling ejbTimeout", e);
            throw e;
        } catch (final Error e) {
            log.warning("Error occured while calling ejbTimeout", e);
            throw e;
        } finally {
            thread.setContextClassLoader(loader);

            // clean up the timer store
            //TODO shall we do all this via Quartz listener ???
            if (timerData.getType() == TimerType.SingleAction) {
                timerStore.removeTimer(timerData.getId());
                timerData.setExpired(true);
            } else if (timerData.getType() == TimerType.Calendar && timerData.getNextTimeout() == null) {
                timerStore.removeTimer(timerData.getId());
                timerData.setExpired(true);
            } else {
                timerStore.updateIntervalTimer(timerData);
            }
        }
    }

    /**
     * The timer task registered with the java.util.Timer.  The run method of this class
     * simply adds an execution of the ejbTimeout method to the thread pool.  It is
     * important to use the thread pool, since the java.util.Timer is single threaded.
     */
    /*private class EjbTimeoutTimerTask extends TimerTask {
        private final TimerData timerData;

        public EjbTimeoutTimerTask(TimerData timerData) {
            this.timerData = timerData;
        }

        public void run() {
            threadPool.execute(new Runnable() {
                public void run() {
                    ejbTimeout(timerData);
                }
            });
        }
    }*/

    private static class LazyScheduler implements InvocationHandler {

        private final BeanContext ejb;

        public LazyScheduler(final BeanContext deployment) {
            ejb = deployment;
        }

        @Override
        public Object invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {
            return method.invoke(getDefaultScheduler(ejb), args);
        }
    }
}