jobstoresupport.java

Quartz is a full-featured, open source job scheduling system that can be integrated with, or used al

     * Store the given <code>{@link org.quartz.JobDetail}</code>.
     * </p>
     * 
     * @param newJob
     *          The <code>JobDetail</code> to be stored.
     * @param replaceExisting
     *          If <code>true</code>, any <code>Job</code> existing in the
     *          <code>JobStore</code> with the same name & group should be
     *          over-written.
     * @throws ObjectAlreadyExistsException
     *           if a <code>Job</code> with the same name/group already
     *           exists, and replaceExisting is set to false.
     */
    public void storeJob(final SchedulingContext ctxt, final JobDetail newJob,
        final boolean replaceExisting) throws ObjectAlreadyExistsException, JobPersistenceException {
        executeInLock(
            (isLockOnInsert() || replaceExisting) ? LOCK_TRIGGER_ACCESS : null,
            new VoidTransactionCallback() {
                public void execute(Connection conn) throws JobPersistenceException {
                    storeJob(conn, ctxt, newJob, replaceExisting);
                }
            });
    }
    
    /**
     * <p>
     * Insert or update a job.
     * </p>
     */
    protected void storeJob(Connection conn, SchedulingContext ctxt,
            JobDetail newJob, boolean replaceExisting)
        throws ObjectAlreadyExistsException, JobPersistenceException {
        if (newJob.isVolatile() && isClustered()) {
            getLog().info(
                "note: volatile jobs are effectively non-volatile in a clustered environment.");
        }

        boolean existingJob = jobExists(conn, newJob.getName(), newJob
                .getGroup());
        try {
            if (existingJob) {
                if (!replaceExisting) { 
                    throw new ObjectAlreadyExistsException(newJob); 
                }
                getDelegate().updateJobDetail(conn, newJob);
            } else {
                getDelegate().insertJobDetail(conn, newJob);
            }
        } catch (IOException e) {
            throw new JobPersistenceException("Couldn't store job: "
                    + e.getMessage(), e);
        } catch (SQLException e) {
            throw new JobPersistenceException("Couldn't store job: "
                    + e.getMessage(), e);
        }
    }

    /**
     * <p>
     * Check existence of a given job.
     * </p>
     */
    protected boolean jobExists(Connection conn, String jobName,
            String groupName) throws JobPersistenceException {
        try {
            return getDelegate().jobExists(conn, jobName, groupName);
        } catch (SQLException e) {
            throw new JobPersistenceException(
                    "Couldn't determine job existence (" + groupName + "."
                            + jobName + "): " + e.getMessage(), e);
        }
    }


    /**
     * <p>
     * Store the given <code>{@link org.quartz.Trigger}</code>.
     * </p>
     * 
     * @param newTrigger
     *          The <code>Trigger</code> to be stored.
     * @param replaceExisting
     *          If <code>true</code>, any <code>Trigger</code> existing in
     *          the <code>JobStore</code> with the same name & group should
     *          be over-written.
     * @throws ObjectAlreadyExistsException
     *           if a <code>Trigger</code> with the same name/group already
     *           exists, and replaceExisting is set to false.
     */
    public void storeTrigger(final SchedulingContext ctxt, final Trigger newTrigger,
        final boolean replaceExisting) throws ObjectAlreadyExistsException,
            JobPersistenceException {
        executeInLock(
            (isLockOnInsert() || replaceExisting) ? LOCK_TRIGGER_ACCESS : null,
            new VoidTransactionCallback() {
                public void execute(Connection conn) throws JobPersistenceException {
                    storeTrigger(conn, ctxt, newTrigger, null, replaceExisting,
                        STATE_WAITING, false, false);
                }
            });
    }
    
    /**
     * <p>
     * Insert or update a trigger.
     * </p>
     */
    protected void storeTrigger(Connection conn, SchedulingContext ctxt,
            Trigger newTrigger, JobDetail job, boolean replaceExisting, String state,
            boolean forceState, boolean recovering)
        throws ObjectAlreadyExistsException, JobPersistenceException {
        if (newTrigger.isVolatile() && isClustered()) {
            getLog().info(
                "note: volatile triggers are effectively non-volatile in a clustered environment.");
        }

        boolean existingTrigger = triggerExists(conn, newTrigger.getName(),
                newTrigger.getGroup());

        if ((existingTrigger) && (!replaceExisting)) { 
            throw new ObjectAlreadyExistsException(newTrigger); 
        }
        
        try {

            boolean shouldBepaused = false;

            if (!forceState) {
                shouldBepaused = getDelegate().isTriggerGroupPaused(
                        conn, newTrigger.getGroup());

                if(!shouldBepaused) {
                    shouldBepaused = getDelegate().isTriggerGroupPaused(conn,
                            ALL_GROUPS_PAUSED);

                    if (shouldBepaused) {
                        getDelegate().insertPausedTriggerGroup(conn, newTrigger.getGroup());
                    }
                }

                if (shouldBepaused && (state.equals(STATE_WAITING) || state.equals(STATE_ACQUIRED))) {
                    state = STATE_PAUSED;
                }
            }

            if(job == null) {
                job = getDelegate().selectJobDetail(conn,
                    newTrigger.getJobName(), newTrigger.getJobGroup(),
                    getClassLoadHelper());
            }
            if (job == null) {
                throw new JobPersistenceException("The job ("
                        + newTrigger.getFullJobName()
                        + ") referenced by the trigger does not exist.");
            }
            if (job.isVolatile() && !newTrigger.isVolatile()) {
                throw new JobPersistenceException(
                        "It does not make sense to "
                                + "associate a non-volatile Trigger with a volatile Job!");
            }

            if (job.isStateful() && !recovering) { 
                state = checkBlockedState(conn, ctxt, job.getName(), 
                        job.getGroup(), state);
            }
            
            if (existingTrigger) {
                if (newTrigger instanceof SimpleTrigger && ((SimpleTrigger)newTrigger).hasAdditionalProperties() == false ) {
                    getDelegate().updateSimpleTrigger(conn,
                            (SimpleTrigger) newTrigger);
                } else if (newTrigger instanceof CronTrigger && ((CronTrigger)newTrigger).hasAdditionalProperties() == false ) {
                    getDelegate().updateCronTrigger(conn,
                            (CronTrigger) newTrigger);
                } else {
                    getDelegate().updateBlobTrigger(conn, newTrigger);
                }
                getDelegate().updateTrigger(conn, newTrigger, state, job);
            } else {
                getDelegate().insertTrigger(conn, newTrigger, state, job);
                if (newTrigger instanceof SimpleTrigger && ((SimpleTrigger)newTrigger).hasAdditionalProperties() == false ) {
                    getDelegate().insertSimpleTrigger(conn,
                            (SimpleTrigger) newTrigger);
                } else if (newTrigger instanceof CronTrigger && ((CronTrigger)newTrigger).hasAdditionalProperties() == false ) {
                    getDelegate().insertCronTrigger(conn,
                            (CronTrigger) newTrigger);
                } else {
                    getDelegate().insertBlobTrigger(conn, newTrigger);
                }
            }
        } catch (Exception e) {
            throw new JobPersistenceException("Couldn't store trigger '" + newTrigger.getName() + "' for '" 
                    + newTrigger.getJobName() + "' job:" + e.getMessage(), e);
        }
    }

    /**
     * <p>
     * Check existence of a given trigger.
     * </p>
     */
    protected boolean triggerExists(Connection conn, String triggerName,
            String groupName) throws JobPersistenceException {
        try {
            return getDelegate().triggerExists(conn, triggerName, groupName);
        } catch (SQLException e) {
            throw new JobPersistenceException(
                    "Couldn't determine trigger existence (" + groupName + "."
                            + triggerName + "): " + e.getMessage(), e);
        }
    }

    /**
     * <p>
     * Remove (delete) the <code>{@link org.quartz.Job}</code> with the given
     * name, and any <code>{@link org.quartz.Trigger}</code> s that reference
     * it.
     * </p>
     * 
     * <p>
     * If removal of the <code>Job</code> results in an empty group, the
     * group should be removed from the <code>JobStore</code>'s list of
     * known group names.
     * </p>
     * 
     * @param jobName
     *          The name of the <code>Job</code> to be removed.
     * @param groupName
     *          The group name of the <code>Job</code> to be removed.
     * @return <code>true</code> if a <code>Job</code> with the given name &
     *         group was found and removed from the store.
     */
    public boolean removeJob(final SchedulingContext ctxt, final String jobName,
        final String groupName) throws JobPersistenceException {
        return ((Boolean)executeInLock(
                LOCK_TRIGGER_ACCESS,
                new TransactionCallback() {
                    public Object execute(Connection conn) throws JobPersistenceException {
                        return removeJob(conn, ctxt, jobName, groupName, true) ? 
                                Boolean.TRUE : Boolean.FALSE;
                    }
                })).booleanValue();
    }
    
    protected boolean removeJob(Connection conn, SchedulingContext ctxt,
            String jobName, String groupName, boolean activeDeleteSafe)
        throws JobPersistenceException {

        try {
            Key[] jobTriggers = getDelegate().selectTriggerNamesForJob(conn,
                    jobName, groupName);
            for (int i = 0; i < jobTriggers.length; ++i) {
                deleteTriggerAndChildren(
                    conn, jobTriggers[i].getName(), jobTriggers[i].getGroup());
            }

            return deleteJobAndChildren(conn, ctxt, jobName, groupName);
        } catch (SQLException e) {
            throw new JobPersistenceException("Couldn't remove job: "
                    + e.getMessage(), e);
        }
    }

    /**
     * Delete a job and its listeners.
     * 
     * @see #removeJob(Connection, SchedulingContext, String, String, boolean)
     * @see #removeTrigger(Connection, SchedulingContext, String, String)
     */
    private boolean deleteJobAndChildren(Connection conn, 
            SchedulingContext ctxt, String jobName, String groupName)
        throws NoSuchDelegateException, SQLException {
        getDelegate().deleteJobListeners(conn, jobName, groupName);

        return (getDelegate().deleteJobDetail(conn, jobName, groupName) > 0);
    }
    
    /**
     * Delete a trigger, its listeners, and its Simple/Cron/BLOB sub-table entry.
     * 
     * @see #removeJob(Connection, SchedulingContext, String, String, boolean)
     * @see #removeTrigger(Connection, SchedulingContext, String, String)
     * @see #replaceTrigger(Connection, SchedulingContext, String, String, Trigger)
     */
    private boolean deleteTriggerAndChildren(
            Connection conn, String triggerName, String triggerGroupName)
        throws SQLException, NoSuchDelegateException {
        DriverDelegate delegate = getDelegate();

        // Once it succeeds in deleting one sub-table entry it will not try the others.
        if ((delegate.deleteSimpleTrigger(conn, triggerName, triggerGroupName) == 0) && 
            (delegate.deleteCronTrigger(conn, triggerName, triggerGroupName) == 0)) {
            delegate.deleteBlobTrigger(conn, triggerName, triggerGroupName);
        }
        
        delegate.deleteTriggerListeners(conn, triggerName, triggerGroupName);
        
        return (delegate.deleteTrigger(conn, triggerName, triggerGroupName) > 0);
    }
    
    /**
     * <p>
     * Retrieve the <code>{@link org.quartz.JobDetail}</code> for the given
     * <code>{@link org.quartz.Job}</code>.
     * </p>
     * 
     * @param jobName
     *          The name of the <code>Job</code> to be retrieved.
     * @param groupName
     *          The group name of the <code>Job</code> to be retrieved.
     * @return The desired <code>Job</code>, or null if there is no match.
     */
    public JobDetail retrieveJob(final SchedulingContext ctxt, final String jobName,
            final String groupName) throws JobPersistenceException {
        return (JobDetail)executeWithoutLock( // no locks necessary for read...
            new TransactionCallback() {
                public Object execute(Connection conn) throws JobPersistenceException {
                    return retrieveJob(conn, ctxt, jobName, groupName);
                }
            });
    }
    
    protected JobDetail retrieveJob(Connection conn, SchedulingContext ctxt,
            String jobName, String groupName) throws JobPersistenceException {
        try {
            JobDetail job = getDelegate().selectJobDetail(conn, jobName,
                    groupName, getClassLoadHelper());
            if (job != null) {
                String[] listeners = getDelegate().selectJobListeners(conn,
                        jobName, groupName);
                for (int i = 0; i < listeners.length; ++i) {
                    job.addJobListener(listeners[i]);
                }
            }

            return job;
        } catch (ClassNotFoundException e) {
            throw new JobPersistenceException(
                    "Couldn't retrieve job because a required class was not found: "
                            + e.getMessage(), e,
                    SchedulerException.ERR_PERSISTENCE_JOB_DOES_NOT_EXIST);
        } catch (IOException e) {
            throw new JobPersistenceException(
                    "Couldn't retrieve job because the BLOB couldn't be deserialized: "
                            + e.getMessage(), e,
                    SchedulerException.ERR_PERSISTENCE_JOB_DOES_NOT_EXIST);
        } catch (SQLException e) {
            throw new JobPersistenceException("Couldn't retrieve job: "
                    + e.getMessage(), e);
        }
    }

    /**