View Javadoc

1   package org.apache.turbine.services.schedule;
2   
3   /*
4    * Licensed to the Apache Software Foundation (ASF) under one
5    * or more contributor license agreements.  See the NOTICE file
6    * distributed with this work for additional information
7    * regarding copyright ownership.  The ASF licenses this file
8    * to you under the Apache License, Version 2.0 (the
9    * "License"); you may not use this file except in compliance
10   * with the License.  You may obtain a copy of the License at
11   *
12   *   http://www.apache.org/licenses/LICENSE-2.0
13   *
14   * Unless required by applicable law or agreed to in writing,
15   * software distributed under the License is distributed on an
16   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
17   * KIND, either express or implied.  See the License for the
18   * specific language governing permissions and limitations
19   * under the License.
20   */
21  
22  import java.util.Iterator;
23  import java.util.List;
24  
25  import javax.servlet.ServletConfig;
26  
27  import org.apache.commons.logging.Log;
28  import org.apache.commons.logging.LogFactory;
29  import org.apache.torque.TorqueException;
30  import org.apache.torque.util.Criteria;
31  import org.apache.turbine.services.InitializationException;
32  import org.apache.turbine.services.TurbineBaseService;
33  import org.apache.turbine.util.TurbineException;
34  
35  /**
36   * Service for a cron like scheduler.
37   *
38   * @author <a href="mailto:mbryson@mont.mindspring.com">Dave Bryson</a>
39   * @author <a href="mailto:quintonm@bellsouth.net">Quinton McCombs</a>
40   * @version $Id: TurbineSchedulerService.java 534527 2007-05-02 16:10:59Z tv $
41   */
42  public class TurbineSchedulerService
43          extends TurbineBaseService
44          implements ScheduleService
45  {
46      /** Logging */
47      protected static Log log = LogFactory.getLog(ScheduleService.LOGGER_NAME);
48  
49      /** The queue */
50      protected JobQueue scheduleQueue = null;
51  
52      /** Current status of the scheduler */
53      protected boolean enabled = false;
54  
55      /** The main loop for starting jobs. */
56      protected MainLoop mainLoop;
57  
58      /** The thread used to process commands.  */
59      protected Thread thread;
60  
61      /**
62       * Creates a new instance.
63       */
64      public TurbineSchedulerService()
65      {
66          mainLoop = null;
67          thread = null;
68      }
69  
70      /**
71       * Initializes the SchedulerService.
72       *
73       * @throws InitializationException Something went wrong in the init
74       *         stage
75       */
76      @Override
77      public void init()
78              throws InitializationException
79      {
80          try
81          {
82              setEnabled(getConfiguration().getBoolean("enabled", true));
83              scheduleQueue = new JobQueue();
84              mainLoop = new MainLoop();
85  
86              // Load all from cold storage.
87              List<JobEntry> jobs = JobEntryPeer.doSelect(new Criteria());
88  
89              if (jobs != null && jobs.size() > 0)
90              {
91                  Iterator<JobEntry> it = jobs.iterator();
92                  while (it.hasNext())
93                  {
94                      it.next().calcRunTime();
95                  }
96                  scheduleQueue.batchLoad(jobs);
97  
98                  restart();
99              }
100 
101             setInit(true);
102         }
103         catch (Exception e)
104         {
105             String errorMessage = "Could not initialize the scheduler service";
106             log.error(errorMessage, e);
107             throw new InitializationException(errorMessage, e);
108         }
109     }
110 
111     /**
112      * Called the first time the Service is used.<br>
113      *
114      * Load all the jobs from cold storage.  Add jobs to the queue
115      * (sorted in ascending order by runtime) and start the scheduler
116      * thread.
117      *
118      * @param config A ServletConfig.
119      * @deprecated use init() instead.
120      */
121     @Deprecated
122     public void init(ServletConfig config) throws InitializationException
123     {
124         init();
125     }
126 
127     /**
128      * Shutdowns the service.
129      *
130      * This methods interrupts the housekeeping thread.
131      */
132     @Override
133     public void shutdown()
134     {
135         if (getThread() != null)
136         {
137             getThread().interrupt();
138         }
139     }
140 
141     /**
142      * Get a specific Job from Storage.
143      *
144      * @param oid The int id for the job.
145      * @return A JobEntry.
146      * @exception TurbineException job could not be retreived.
147      */
148     public JobEntry getJob(int oid)
149             throws TurbineException
150     {
151         try
152         {
153             JobEntry je = JobEntryPeer.retrieveByPK(oid);
154             return scheduleQueue.getJob(je);
155         }
156         catch (TorqueException e)
157         {
158             String errorMessage = "Error retrieving job from persistent storage.";
159             log.error(errorMessage, e);
160             throw new TurbineException(errorMessage, e);
161         }
162     }
163 
164     /**
165      * Add a new job to the queue.
166      *
167      * @param je A JobEntry with the job to add.
168      * @throws TurbineException job could not be added
169      */
170     public void addJob(JobEntry je)
171             throws TurbineException
172     {
173         updateJob(je);
174     }
175 
176     /**
177      * Remove a job from the queue.
178      *
179      * @param je A JobEntry with the job to remove.
180      * @exception TurbineException job could not be removed
181      */
182     public void removeJob(JobEntry je)
183             throws TurbineException
184     {
185         try
186         {
187             // First remove from DB.
188             Criteria c = new Criteria().add(JobEntryPeer.JOB_ID, je.getPrimaryKey());
189             JobEntryPeer.doDelete(c);
190 
191             // Remove from the queue.
192             scheduleQueue.remove(je);
193 
194             // restart the scheduler
195             restart();
196         }
197         catch (Exception e)
198         {
199             String errorMessage = "Problem removing Scheduled Job: " + je.getTask();
200             log.error(errorMessage, e);
201             throw new TurbineException(errorMessage, e);
202         }
203     }
204 
205     /**
206      * Add or update a job.
207      *
208      * @param je A JobEntry with the job to modify
209      * @throws TurbineException job could not be updated
210      */
211     public void updateJob(JobEntry je)
212             throws TurbineException
213     {
214         try
215         {
216             je.calcRunTime();
217 
218             // Update the queue.
219             if (je.isNew())
220             {
221                 scheduleQueue.add(je);
222             }
223             else
224             {
225                 scheduleQueue.modify(je);
226             }
227 
228             je.save();
229 
230             restart();
231         }
232         catch (Exception e)
233         {
234             String errorMessage = "Problem updating Scheduled Job: " + je.getTask();
235             log.error(errorMessage, e);
236             throw new TurbineException(errorMessage, e);
237         }
238     }
239 
240     /**
241      * List jobs in the queue.  This is used by the scheduler UI.
242      *
243      * @return A List of jobs.
244      */
245     public List<JobEntry> listJobs()
246     {
247         return scheduleQueue.list();
248     }
249 
250     /**
251      * Sets the enabled status of the scheduler
252      *
253      * @param enabled
254      *
255      */
256     protected void setEnabled(boolean enabled)
257     {
258         this.enabled = enabled;
259     }
260 
261     /**
262      * Determines if the scheduler service is currently enabled.
263      *
264      * @return Status of the scheduler service.
265      */
266     public boolean isEnabled()
267     {
268         return enabled;
269     }
270 
271     /**
272      * Starts or restarts the scheduler if not already running.
273      */
274     public synchronized void startScheduler()
275     {
276         setEnabled(true);
277         restart();
278     }
279 
280     /**
281      * Stops the scheduler if it is currently running.
282      */
283     public synchronized void stopScheduler()
284     {
285         log.info("Stopping job scheduler");
286         Thread thread = getThread();
287         if (thread != null)
288         {
289             thread.interrupt();
290         }
291         enabled = false;
292     }
293 
294     /**
295      * Return the thread being used to process commands, or null if
296      * there is no such thread.  You can use this to invoke any
297      * special methods on the thread, for example, to interrupt it.
298      *
299      * @return A Thread.
300      */
301     public synchronized Thread getThread()
302     {
303         return thread;
304     }
305 
306     /**
307      * Set thread to null to indicate termination.
308      */
309     protected synchronized void clearThread()
310     {
311         thread = null;
312     }
313 
314     /**
315      * Start (or restart) a thread to process commands, or wake up an
316      * existing thread if one is already running.  This method can be
317      * invoked if the background thread crashed due to an
318      * unrecoverable exception in an executed command.
319      */
320     public synchronized void restart()
321     {
322         if (enabled)
323         {
324             log.info("Starting job scheduler");
325             if (thread == null)
326             {
327                 // Create the the housekeeping thread of the scheduler. It will wait
328                 // for the time when the next task needs to be started, and then
329                 // launch a worker thread to execute the task.
330                 thread = new Thread(mainLoop, ScheduleService.SERVICE_NAME);
331                 // Indicate that this is a system thread. JVM will quit only when there
332                 // are no more enabled user threads. Settings threads spawned internally
333                 // by Turbine as daemons allows commandline applications using Turbine
334                 // to terminate in an orderly manner.
335                 thread.setDaemon(true);
336                 thread.start();
337             }
338             else
339             {
340                 notify();
341             }
342         }
343     }
344 
345     /**
346      *  Return the next Job to execute, or null if thread is
347      *  interrupted.
348      *
349      * @return A JobEntry.
350      * @exception TurbineException a generic exception.
351      */
352     protected synchronized JobEntry nextJob()
353             throws TurbineException
354     {
355         try
356         {
357             while (!Thread.interrupted())
358             {
359                 // Grab the next job off the queue.
360                 JobEntry je = scheduleQueue.getNext();
361 
362                 if (je == null)
363                 {
364                     // Queue must be empty. Wait on it.
365                     wait();
366                 }
367                 else
368                 {
369                     long now = System.currentTimeMillis();
370                     long when = je.getNextRuntime();
371 
372                     if (when > now)
373                     {
374                         // Wait till next runtime.
375                         wait(when - now);
376                     }
377                     else
378                     {
379                         // Update the next runtime for the job.
380                         scheduleQueue.updateQueue(je);
381                         // Return the job to run it.
382                         return je;
383                     }
384                 }
385             }
386         }
387         catch (InterruptedException ex)
388         {
389             // ignore
390         }
391 
392         // On interrupt.
393         return null;
394     }
395 
396     /**
397      * Inner class.  This is isolated in its own Runnable class just
398      * so that the main class need not implement Runnable, which would
399      * allow others to directly invoke run, which is not supported.
400      */
401     protected class MainLoop
402             implements Runnable
403     {
404         /**
405          * Method to run the class.
406          */
407         public void run()
408         {
409             String taskName = null;
410             try
411             {
412                 while (enabled)
413                 {
414                     JobEntry je = nextJob();
415                     if (je != null)
416                     {
417                         taskName = je.getTask();
418 
419                         // Start the thread to run the job.
420                         Runnable wt = new WorkerThread(je);
421                         Thread helper = new Thread(wt);
422                         helper.start();
423                     }
424                     else
425                     {
426                         break;
427                     }
428                 }
429             }
430             catch (Exception e)
431             {
432                 log.error("Error running a Scheduled Job: " + taskName, e);
433                 enabled = false;
434             }
435             finally
436             {
437                 clearThread();
438             }
439         }
440     }
441 }