jobstoresupport.java

非常好用的计划任务调度包,能定义复杂的任务

     * @see #pauseTrigger(Connection, SchedulingContext, String, String)
     */
    public void resumeTrigger(Connection conn, SchedulingContext ctxt,
            String triggerName, String groupName)
            throws JobPersistenceException {
        try {

            TriggerStatus status = getDelegate().selectTriggerStatus(conn,
                    triggerName, groupName);

            if (status == null || status.getNextFireTime() == null) return;

            String newState = getStatusForResumedTrigger(conn, ctxt, status);

            boolean misfired = false;

            if (status.getNextFireTime().before(new Date())) {
              misfired = updateMisfiredTrigger(conn, ctxt, triggerName, groupName,
                  newState, true);
            }

            if(!misfired) {
                getDelegate().updateTriggerStateFromOtherState(conn,
                        triggerName, groupName, newState, STATE_PAUSED);
            } 

        } catch (SQLException e) {
            throw new JobPersistenceException("Couldn't resume trigger '"
                    + groupName + "." + triggerName + "': " + e.getMessage(), e);
        }
    }

    /**
     * <p>
     * Pause all of the <code>{@link org.quartz.Trigger}s</code> in the
     * given group.
     * </p>
     * 
     * @see #resumeTriggerGroup(Connection, SchedulingContext, String)
     */
    public void pauseTriggerGroup(Connection conn, SchedulingContext ctxt,
            String groupName) throws JobPersistenceException {

        try {

            getDelegate().updateTriggerGroupStateFromOtherStates(
                    conn, groupName, STATE_PAUSED, STATE_ACQUIRED,
                    STATE_BLOCKED, STATE_WAITING);

            if (!getDelegate().isTriggerGroupPaused(conn, groupName)) {
                getDelegate().insertPausedTriggerGroup(conn, groupName);
            }

        } catch (SQLException e) {
            throw new JobPersistenceException("Couldn't pause trigger group '"
                    + groupName + "': " + e.getMessage(), e);
        }
    }

    /**
     * <p>
     * Pause all of the <code>{@link org.quartz.Trigger}s</code> in the
     * given group.
     * </p>
     * 
     * @see #resumeTriggerGroup(Connection, SchedulingContext, String)
     */
    public Set getPausedTriggerGroups(Connection conn, SchedulingContext ctxt) 
        throws JobPersistenceException {

        try {
            return getDelegate().selectPausedTriggerGroups(conn);
        } catch (SQLException e) {
            throw new JobPersistenceException(
                    "Couldn't determine paused trigger groups: " + e.getMessage(), e);
        }
    }
    
    /**
     * <p>
     * Resume (un-pause) all of the <code>{@link org.quartz.Trigger}s</code>
     * in the given group.
     * </p>
     * 
     * <p>
     * If any <code>Trigger</code> missed one or more fire-times, then the
     * <code>Trigger</code>'s misfire instruction will be applied.
     * </p>
     * 
     * @see #pauseTriggerGroup(Connection, SchedulingContext, String)
     */
    public void resumeTriggerGroup(Connection conn, SchedulingContext ctxt,
            String groupName) throws JobPersistenceException {

        try {

            getDelegate().deletePausedTriggerGroup(conn, groupName);

            String[] trigNames = getDelegate().selectTriggersInGroup(conn,
                    groupName);

            for (int i = 0; i < trigNames.length; i++) {
                resumeTrigger(conn, ctxt, trigNames[i], groupName);
            }

            // TODO: find an efficient way to resume triggers (better than the
            // above)... logic below is broken because of
            // findTriggersToBeBlocked()
            /*
             * int res =
             * getDelegate().updateTriggerGroupStateFromOtherState(conn,
             * groupName, STATE_WAITING, STATE_PAUSED);
             * 
             * if(res > 0) {
             * 
             * long misfireTime = System.currentTimeMillis();
             * if(getMisfireThreshold() > 0) misfireTime -=
             * getMisfireThreshold();
             * 
             * Key[] misfires =
             * getDelegate().selectMisfiredTriggersInGroupInState(conn,
             * groupName, STATE_WAITING, misfireTime);
             * 
             * List blockedTriggers = findTriggersToBeBlocked(conn, ctxt,
             * groupName);
             * 
             * Iterator itr = blockedTriggers.iterator(); while(itr.hasNext()) {
             * Key key = (Key)itr.next();
             * getDelegate().updateTriggerState(conn, key.getName(),
             * key.getGroup(), STATE_BLOCKED); }
             * 
             * for(int i=0; i < misfires.length; i++) {               String
             * newState = STATE_WAITING;
             * if(blockedTriggers.contains(misfires[i])) newState =
             * STATE_BLOCKED; updateMisfiredTrigger(conn, ctxt,
             * misfires[i].getName(), misfires[i].getGroup(), newState, true); } }
             */

        } catch (SQLException e) {
            throw new JobPersistenceException("Couldn't pause trigger group '"
                    + groupName + "': " + e.getMessage(), e);
        }
    }

    /**
     * <p>
     * Pause all triggers - equivalent of calling <code>pauseTriggerGroup(group)</code>
     * on every group.
     * </p>
     * 
     * <p>
     * When <code>resumeAll()</code> is called (to un-pause), trigger misfire
     * instructions WILL be applied.
     * </p>
     * 
     * @see #resumeAll(SchedulingContext)
     * @see #pauseTriggerGroup(SchedulingContext, String)
     */
    public void pauseAll(Connection conn, SchedulingContext ctxt)
            throws JobPersistenceException {

        String[] names = getTriggerGroupNames(conn, ctxt);

        for (int i = 0; i < names.length; i++) {
            pauseTriggerGroup(conn, ctxt, names[i]);
        }

        try {
            if (!getDelegate().isTriggerGroupPaused(conn, ALL_GROUPS_PAUSED)) {
                getDelegate().insertPausedTriggerGroup(conn, ALL_GROUPS_PAUSED);
            }

        } catch (SQLException e) {
            throw new JobPersistenceException(
                    "Couldn't pause all trigger groups: " + e.getMessage(), e);
        }

    }

    /**
     * protected
     * <p>
     * Resume (un-pause) all triggers - equivalent of calling <code>resumeTriggerGroup(group)</code>
     * on every group.
     * </p>
     * 
     * <p>
     * If any <code>Trigger</code> missed one or more fire-times, then the
     * <code>Trigger</code>'s misfire instruction will be applied.
     * </p>
     * 
     * @see #pauseAll(SchedulingContext)
     */
    public void resumeAll(Connection conn, SchedulingContext ctxt)
            throws JobPersistenceException {

        String[] names = getTriggerGroupNames(conn, ctxt);

        for (int i = 0; i < names.length; i++) {
            resumeTriggerGroup(conn, ctxt, names[i]);
        }

        try {
            getDelegate().deletePausedTriggerGroup(conn, ALL_GROUPS_PAUSED);
        } catch (SQLException e) {
            throw new JobPersistenceException(
                    "Couldn't resume all trigger groups: " + e.getMessage(), e);
        }

    }

    private static long ftrCtr = System.currentTimeMillis();

    protected synchronized String getFiredTriggerRecordId() {
        return getInstanceId() + ftrCtr++;
    }

    // TODO: this really ought to return something like a FiredTriggerBundle,
    // so that the fireInstanceId doesn't have to be on the trigger...
    protected Trigger acquireNextTrigger(Connection conn, SchedulingContext ctxt)
            throws JobPersistenceException {
        Trigger nextTrigger = null;

        boolean acquiredOne = false;

        do {
            try {
                getDelegate().updateTriggerStateFromOtherStatesBeforeTime(conn,
                        STATE_MISFIRED, STATE_WAITING, STATE_WAITING,
                        getMisfireTime()); // only waiting

                long nextFireTime = getDelegate().selectNextFireTime(conn);

                if (nextFireTime == 0) return null;

                Key triggerKey = null;
                do {
                    triggerKey = getDelegate().selectTriggerForFireTime(conn,
                            nextFireTime);
                    if (null != triggerKey) {

                        int res = getDelegate()
                                .updateTriggerStateFromOtherState(conn,
                                        triggerKey.getName(),
                                        triggerKey.getGroup(), STATE_ACQUIRED,
                                        STATE_WAITING);

                        if (res <= 0) continue;

                        nextTrigger = retrieveTrigger(conn, ctxt, triggerKey
                                .getName(), triggerKey.getGroup());

                        if(nextTrigger == null) continue;
                        
                        nextTrigger
                                .setFireInstanceId(getFiredTriggerRecordId());
                        getDelegate().insertFiredTrigger(conn, nextTrigger,
                                STATE_ACQUIRED, null);

                        acquiredOne = true;
                    }
                } while (triggerKey != null && !acquiredOne);
            } catch (Exception e) {
                throw new JobPersistenceException(
                        "Couldn't acquire next trigger: " + e.getMessage(), e);
            }

        } while (!acquiredOne);

        return nextTrigger;
    }

    protected void releaseAcquiredTrigger(Connection conn,
            SchedulingContext ctxt, Trigger trigger)
            throws JobPersistenceException {
        try {
            getDelegate().updateTriggerStateFromOtherState(conn,
                    trigger.getName(), trigger.getGroup(), STATE_WAITING,
                    STATE_ACQUIRED);
            getDelegate().deleteFiredTrigger(conn, trigger.getFireInstanceId());
        } catch (SQLException e) {
            throw new JobPersistenceException(
                    "Couldn't release acquired trigger: " + e.getMessage(), e);
        }
    }

    protected TriggerFiredBundle triggerFired(Connection conn,
            SchedulingContext ctxt, Trigger trigger)
            throws JobPersistenceException {
        JobDetail job = null;
        Calendar cal = null;

        // Make sure trigger wasn't deleted, paused, or completed...
        try { // if trigger was deleted, state will be STATE_DELETED
            String state = getDelegate().selectTriggerState(conn,
                    trigger.getName(), trigger.getGroup());
            if (!state.equals(STATE_ACQUIRED)) return null;
        } catch (SQLException e) {
            throw new JobPersistenceException("Couldn't select trigger state: "
                    + e.getMessage(), e);
        }

        try {
            job = retrieveJob(conn, ctxt, trigger.getJobName(), trigger
                    .getJobGroup());
            if (job == null) { return null; }
        } catch (JobPersistenceException jpe) {
            try {
                getDelegate().updateTriggerState(conn, trigger.getName(),
                        trigger.getGroup(), STATE_ERROR);
            } catch (SQLException sqle) {
                getLog().error("Unable to set trigger state to ERROR.", sqle);
            }
            throw jpe;
        }

        if (trigger.getCalendarName() != null) {
            cal = retrieveCalendar(conn, ctxt, trigger.getCalendarName());
            if (cal == null) { return null; }
        }

        try {
            getDelegate().deleteFiredTrigger(conn, trigger.getFireInstanceId());
            getDelegate().insertFiredTrigger(conn, trigger, STATE_EXECUTING,
                    job);
        } catch (SQLException e) {
            throw new JobPersistenceException("Couldn't insert fired trigger: "
                    + e.getMessage(), e);
        }

        Date prevFireTime = trigger.getPreviousFireTime();

        // call triggered - to update the trigger's next-fire-time state...
        trigger.triggered(cal);

        String state = STATE_WAITING;
        boolean force = true;
        
        if (job.isStateful()) {
            state = STATE_BLOCKED;
            force = false;
            try {
                getDelegate().updateTriggerStatesForJob(conn, job.getName(),
                        job.getGroup(), STATE_BLOCKED);
            } catch (SQLException e) {
                throw new JobPersistenceException(
                        "Couldn't update states of blocked triggers: "
                                + e.getMessage(), e);
            }
        } 
            
        if (trigger.getNextFireTime() == null) {
                state = STATE_COMPLETE;
                force = true;
        }

        storeTrigger(conn, ctxt, trigger, job, true, state, force, false);

        job.getJobDataMap().clearDirtyFlag();

        return new TriggerFiredBundle(job, trigger, cal, trigger.getGroup()
                .equals(Scheduler.DEFAULT_RECOVERY_GROUP), new Date(), trigger
                .getPreviousFireTime(), prevFireTime, trigger.getNextFireTime());
    }

    pro