jobschedulingdataprocessor.java

Quartz 是个开源的作业调度框架

     * @param validating    whether or not to validate XML.
     */
    protected void initSchemaValidation(boolean validatingSchema) {
        if (validatingSchema) {
            String schemaUri = null;
            URL url = getClass().getResource(QUARTZ_XSD);
            if (url != null) {
                schemaUri = url.toExternalForm();
            }
            else {
                schemaUri = QUARTZ_SCHEMA;
            }
            digester.setSchema(schemaUri);
        }
    }

    protected static Log getLog() {
        return LogFactory.getLog(JobSchedulingDataProcessor.class);
    }

    /**
     * Returns whether to use the context class loader.
     * 
     * @return whether to use the context class loader.
     */
    public boolean getUseContextClassLoader() {
        return digester.getUseContextClassLoader();
    }
    
    /**
     * Sets whether to use the context class loader.
     * 
     * @param useContextClassLoader boolean.
     */
    public void setUseContextClassLoader(boolean useContextClassLoader) {
        digester.setUseContextClassLoader(useContextClassLoader);
    }

    /**
     * Returns whether to overwrite existing jobs.
     * 
     * @return whether to overwrite existing jobs.
     */
    public boolean getOverWriteExistingJobs() {
        return overWriteExistingJobs;
    }
    
    /**
     * Sets whether to overwrite existing jobs.
     * 
     * @param overWriteExistingJobs boolean.
     */
    public void setOverWriteExistingJobs(boolean overWriteExistingJobs) {
        this.overWriteExistingJobs = overWriteExistingJobs;
    }

    /*
     * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     * 
     * Interface.
     * 
     * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     */

    /**
     * Process the xml file in the default location (a file named
     * "quartz_jobs.xml" in the current working directory).
     *  
     */
    public void processFile() throws Exception {
        processFile(QUARTZ_XML_FILE_NAME);
    }

    /**
     * Process the xml file named <code>fileName</code>.
     * 
     * @param fileName
     *          meta data file name.
     */
    public void processFile(String fileName) throws Exception {
        processFile(fileName, fileName);
    }

    /**
     * Process the xmlfile named <code>fileName</code> with the given system
     * ID.
     * 
     * @param fileName
     *          meta data file name.
     * @param systemId
     *          system ID.
     */
    public void processFile(String fileName, String systemId)
            throws ValidationException, ParserConfigurationException,
            SAXException, IOException, SchedulerException,
            ClassNotFoundException, ParseException {
        clearValidationExceptions();

        scheduledJobs.clear();
        jobsToSchedule.clear();
        calsToSchedule.clear();

        getLog().info("Parsing XML file: " + fileName +
                      " with systemId: " + systemId +
                      " validating: " + digester.getValidating() +
                      " validating schema: " + digester.getSchema());
        InputSource is = new InputSource(getInputStream(fileName));
        is.setSystemId(systemId);
        digester.push(this);
        digester.parse(is);

        maybeThrowValidationException();
    }
    
    /**
     * Process the xmlfile named <code>fileName</code> with the given system
     * ID.
     * 
     * @param stream
     *          an input stream containing the xml content.
     * @param systemId
     *          system ID.
     */
    public void processStream(InputStream stream, String systemId)
            throws ValidationException, ParserConfigurationException,
            SAXException, IOException, SchedulerException,
            ClassNotFoundException, ParseException {
        clearValidationExceptions();

        scheduledJobs.clear();
        jobsToSchedule.clear();
        calsToSchedule.clear();

        getLog().info("Parsing XML from stream with systemId: " + systemId +
                      " validating: " + digester.getValidating() +
                      " validating schema: " + digester.getSchema());
        InputSource is = new InputSource(stream);
        is.setSystemId(systemId);
        digester.push(this);
        digester.parse(is);

        maybeThrowValidationException();
    }
    
    /**
     * Process the xml file in the default location, and schedule all of the
     * jobs defined within it.
     *  
     */
    public void processFileAndScheduleJobs(Scheduler sched,
            boolean overWriteExistingJobs) throws SchedulerException, Exception {
        processFileAndScheduleJobs(QUARTZ_XML_FILE_NAME, sched,
                overWriteExistingJobs);
    }

    /**
     * Process the xml file in the given location, and schedule all of the
     * jobs defined within it.
     * 
     * @param fileName
     *          meta data file name.
     */
    public void processFileAndScheduleJobs(String fileName, Scheduler sched,
            boolean overWriteExistingJobs) throws Exception {
        processFileAndScheduleJobs(fileName, fileName, sched, overWriteExistingJobs);
    }
    
    /**
     * Process the xml file in the given location, and schedule all of the
     * jobs defined within it.
     * 
     * @param fileName
     *          meta data file name.
     */
    public void processFileAndScheduleJobs(String fileName, String systemId,
            Scheduler sched, boolean overWriteExistingJobs) throws Exception {
        schedLocal.set(sched);
        try {
            processFile(fileName, systemId);
        scheduleJobs(getScheduledJobs(), sched, overWriteExistingJobs);
        } finally {
            schedLocal.set(null);
        }
    }

    /**
     * Add the Jobs and Triggers defined in the given map of <code>JobSchedulingBundle</code>
     * s to the given scheduler.
     * 
     * @param jobBundles
     * @param sched
     * @param overWriteExistingJobs
     * @throws Exception
     */
    public void scheduleJobs(Map jobBundles, Scheduler sched,
            boolean overWriteExistingJobs) throws Exception {
        getLog().info("Scheduling " + jobsToSchedule.size() + " parsed jobs.");

        Iterator itr = calsToSchedule.iterator();
        while (itr.hasNext()) {
            CalendarBundle bndle = (CalendarBundle) itr.next();
            addCalendar(sched, bndle);
        }

        itr = jobsToSchedule.iterator();
        while (itr.hasNext()) {
            JobSchedulingBundle bndle = (JobSchedulingBundle) itr.next();
            scheduleJob(bndle, sched, overWriteExistingJobs);
        }
        
        itr = listenersToSchedule.iterator();
        while (itr.hasNext()) {
            JobListener listener = (JobListener) itr.next();
            getLog().info("adding listener "+listener.getName()+" of class "+listener.getClass().getName());
            sched.addJobListener(listener);
        }
        getLog().info(jobBundles.size() + " scheduled jobs.");        
    }

    /**
     * Returns a <code>Map</code> of scheduled jobs.
     * <p/>
     * The key is the job name and the value is a <code>JobSchedulingBundle</code>
     * containing the <code>JobDetail</code> and <code>Trigger</code>.
     * 
     * @return a <code>Map</code> of scheduled jobs.
     */
    public Map getScheduledJobs() {
        return Collections.unmodifiableMap(scheduledJobs);
    }

    /**
     * Returns a <code>JobSchedulingBundle</code> for the job name.
     * 
     * @param name
     *          job name.
     * @return a <code>JobSchedulingBundle</code> for the job name.
     */
    public JobSchedulingBundle getScheduledJob(String name) {
        return (JobSchedulingBundle) getScheduledJobs().get(name);
    }

    /**
     * Returns an <code>InputStream</code> from the fileName as a resource.
     * 
     * @param fileName
     *          file name.
     * @return an <code>InputStream</code> from the fileName as a resource.
     */
    protected InputStream getInputStream(String fileName) {
        ClassLoader cl = Thread.currentThread().getContextClassLoader();

        InputStream is = cl.getResourceAsStream(fileName);

        return is;
    }
    
    /**
     * Schedules a given job and trigger (both wrapped by a <code>JobSchedulingBundle</code>).
     * 
     * @param job
     *          job wrapper.
     * @exception SchedulerException
     *              if the Job or Trigger cannot be added to the Scheduler, or
     *              there is an internal Scheduler error.
     */
    public void scheduleJob(JobSchedulingBundle job)
        throws SchedulerException {
        scheduleJob(job, (Scheduler) schedLocal.get(), getOverWriteExistingJobs());
    }

    
    public void addJobToSchedule(JobSchedulingBundle job)
    {
        jobsToSchedule.add(job);
    }

    public void addCalendarToSchedule(CalendarBundle cal)
    {
        calsToSchedule.add(cal);
    }

    public void addListenerToSchedule(JobListener listener)
    {
        listenersToSchedule.add(listener);
    }
    
    /**
     * Schedules a given job and trigger (both wrapped by a <code>JobSchedulingBundle</code>).
     * 
     * @param job
     *          job wrapper.
     * @param sched
     *          job scheduler.
     * @param localOverWriteExistingJobs
     *          locally overwrite existing jobs.
     * @exception SchedulerException
     *              if the Job or Trigger cannot be added to the Scheduler, or
     *              there is an internal Scheduler error.
     */
    public void scheduleJob(JobSchedulingBundle job, Scheduler sched, boolean localOverWriteExistingJobs)
            throws SchedulerException {
        if ((job != null) && job.isValid()) {
            JobDetail detail = job.getJobDetail();
            
            JobDetail dupeJ = sched.getJobDetail(detail.getName(), detail.getGroup());

            if ((dupeJ != null) && !localOverWriteExistingJobs) {
                getLog().info("Not overwriting existing job: " + dupeJ.getFullName());
                return;
            }
            
            if (dupeJ != null) {
                getLog().info("Replacing job: " + detail.getFullName());
            }
            else {
                getLog().info("Adding job: " + detail.getFullName());
            }
            
            if (job.getTriggers().size() == 0 && !job.getJobDetail().isDurable()) {