1 /**
2 * Copyright (C) 2005-2009 Alfresco Software Limited.
3 *
4 * This file is part of the Spring Surf Extension project.
5 *
6 * Licensed under the Apache License, Version 2.0 (the "License");
7 * you may not use this file except in compliance with the License.
8 * You may obtain a copy of the License at
9 *
10 * http://www.apache.org/licenses/LICENSE-2.0
11 *
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS,
14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 * See the License for the specific language governing permissions and
16 * limitations under the License.
17 */
18
19 package org.springframework.extensions.surf.task;
20
21 import org.apache.commons.logging.Log;
22 import org.apache.commons.logging.LogFactory;
23
24 /**
25 * Worker thread implementation that is responsible for executing a task,
26 * updating task state and then falling back asleep until called upon once more.
27 *
28 * @author muzquiano
29 */
30 public class TaskWorkerThread extends Thread
31 {
32 private static Log logger = LogFactory.getLog(TaskWorkerThread.class);
33
34 private AbstractTask task = null;
35 private TaskManager taskManager = null;
36
37 public TaskWorkerThread(TaskManager taskManager)
38 {
39 this.taskManager = taskManager;
40 }
41
42 public AbstractTask getTask()
43 {
44 return this.task;
45 }
46
47 public void setTask(AbstractTask task)
48 {
49 this.task = task;
50 }
51
52 public Object getMutex()
53 {
54 return taskManager.getClass();
55 }
56
57 public void run()
58 {
59 while (true)
60 {
61 try
62 {
63 synchronized (getMutex())
64 {
65 // wait until we are told that there is some work to do
66 getMutex().wait();
67
68 // let's make sure there is a task available to do
69 AbstractTask task = (AbstractTask) taskManager.getTaskQueue().peek();
70 if (task != null)
71 {
72 task = (AbstractTask) taskManager.getTaskQueue().poll();
73
74 // some debug info
75 if (logger.isDebugEnabled())
76 {
77 logger.debug("Found a job to start (" + task.getName() + ")");
78 }
79
80 // set this as the active task
81 setTask(task);
82 }
83 }
84 }
85 catch (InterruptedException ie)
86 {
87 }
88
89 // if we have a task, let's run it
90 if (getTask() != null)
91 {
92 // reset the task
93 getTask().progress = 0;
94 getTask().setStatus("Starting...");
95
96 // mark that the job is running
97 getTask().isRunning = true;
98 getTask().isFinished = false;
99 getTask().isError = false;
100 getTask().isSuccess = false;
101
102 if (logger.isDebugEnabled())
103 {
104 logger.debug("Running job " + getTask().getId());
105 }
106
107 try
108 {
109 getTask().execute();
110 getTask().isSuccess = true;
111 }
112 catch (Throwable t)
113 {
114 getTask().throwable = t;
115 t.printStackTrace();
116 getTask().isError = true;
117 }
118
119 if (getTask().isCancelled() && logger.isDebugEnabled())
120 {
121 getTask().isSuccess = false;
122
123 logger.debug("Finished with a cancelled job");
124 }
125
126 // mark that the job has completed
127 if (logger.isDebugEnabled())
128 {
129 logger.debug("Completing job " + getTask().getId());
130 }
131
132 getTask().setStatus("Finished");
133 getTask().isFinished = true;
134 getTask().isRunning = false;
135
136 // now unbind the job from this thread, we're done working
137 setTask(null);
138 }
139 }
140 }
141 }