WorkerQueue


1   /*
2     * JBoss, Home of Professional Open Source
3     * Copyright 2005, JBoss Inc., and individual contributors as indicated
4     * by the @authors tag. See the copyright.txt in the distribution for a
5     * full listing of individual contributors.
6     *
7     * This is free software; you can redistribute it and/or modify it
8     * under the terms of the GNU Lesser General Public License as
9     * published by the Free Software Foundation; either version 2.1 of
10    * the License, or (at your option) any later version.
11    *
12    * This software is distributed in the hope that it will be useful,
13    * but WITHOUT ANY WARRANTY; without even the implied warranty of
14    * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15    * Lesser General Public License for more details.
16    *
17    * You should have received a copy of the GNU Lesser General Public
18    * License along with this software; if not, write to the Free
19    * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
20    * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
21    */
22  package org.jboss.util;
23  
24  /**
25   * Class that queues {@link Executable} jobs that are executed sequentially 
26   * by a single thread.
27   *
28   * @see Executable
29   * 
30   * @author <a HREF="mailto:simone.bordet@compaq.com">Simone Bordet</a>
31   * @version $Revision: 1958 $
32   */
33  public class WorkerQueue
34  {
35     /** The thread that runs the Executable jobs */
36     protected Thread   m_queueThread;
37     
38     /** The job that will be executed by the worker thread */
39     private JobItem m_currentJob;
40  
41     /**
42      * Creates a new worker queue with default thread name of "Worker Thread"
43      */
44     public WorkerQueue() 
45     {
46        this("Worker Thread");
47     }
48  
49     /**
50      * Creates a new worker queue with the specified thread name
51      */
52     public WorkerQueue(String   threadName) 
53     {
54        m_queueThread = new Thread  (createQueueLoop(), threadName);
55     }
56     
57     /**
58      * Creates a new worker queue with the specified thread name
59      * and daemon mode flag
60      */
61     public WorkerQueue(String   threadName, boolean isDaemon) 
62     {
63        m_queueThread = new Thread  (createQueueLoop(), threadName);
64        m_queueThread.setDaemon(isDaemon);
65     }
66  
67     /**
68      * Starts the worker queue.
69      * @see #stop
70      */
71     public void start() 
72     {
73        if (m_queueThread != null) {m_queueThread.start();}
74     }
75  
76     /**
77      * Stops nicely the worker queue. <br> 
78      * After this call trying to put a new job will result in a 
79      * InterruptedException to be thrown. The jobs queued before and not 
80      * yet processed are processed until the queue is empty, then this 
81      * worker queue is cleared.
82      * @see #clear
83      * @see #start
84      * @see #isInterrupted
85      */
86     public synchronized void stop() 
87     {
88        if (m_queueThread != null) {m_queueThread.interrupt();}
89     }
90     
91     /**
92      * Called by a thread that is not the WorkerQueue thread, this method 
93      * queues the job and, if necessary, wakes up this worker queue that is 
94      * waiting in {@link #getJob}.
95      */
96     public synchronized void putJob(Executable job)
97     {
98        // Preconditions
99        if (m_queueThread == null || !m_queueThread.isAlive()) {
100          throw new IllegalStateException  ("Can't put job, thread is not alive or not present");
101       }
102       
103       if (isInterrupted()) {
104          throw new IllegalStateException  ("Can't put job, thread was interrupted");
105       }
106         
107       putJobImpl(job);
108    }
109    
110    /**
111     * Returns whether the worker thread has been interrupted. <br>
112     * When this method returns true, it is not possible to put new jobs in the
113     * queue and the already present jobs are executed and removed from the 
114     * queue, then the thread exits.
115     * 
116     * @see #stop
117     */
118    protected boolean isInterrupted() 
119    {
120       return m_queueThread.isInterrupted();
121    }
122 
123    /**
124     * Called by this class, this method checks if the queue is empty; 
125     * if it is, then waits, else returns the current job.
126     * 
127     * @see #putJob
128     */
129    protected synchronized Executable getJob() throws InterruptedException  
130    {
131       // Preconditions
132       if (m_queueThread == null || !m_queueThread.isAlive()) {
133          throw new IllegalStateException  ();
134       }
135         
136       return getJobImpl();
137    }
138    
139    /**
140     * Never call this method, only override in subclasses to perform
141     * job getting in a specific way, normally tied to the data structure 
142     * holding the jobs.
143     */
144    protected Executable getJobImpl() throws InterruptedException  
145    {
146       // While the queue is empty, wait();
147       // when notified take an event from the queue and return it.
148       while (m_currentJob == null) {wait();}
149       // This one is the job to return
150       JobItem item = m_currentJob;
151       // Go on to the next object for the next call. 
152       m_currentJob = m_currentJob.m_next;
153       return item.m_job;
154    }
155    
156    /**
157     * Never call this method, only override in subclasses to perform
158     * job adding in a specific way, normally tied to the data structure
159     * holding the jobs.
160     */
161    protected void putJobImpl(Executable job) 
162    {
163       JobItem posted = new JobItem(job);      
164       if (m_currentJob == null) 
165       {
166          // The queue is empty, set the current job to process and
167          // wake up the thread waiting in method getJob
168          m_currentJob = posted;
169          notifyAll();
170       }
171       else 
172       {
173          JobItem item = m_currentJob;
174          // The queue is not empty, find the end of the queue ad add the
175          // posted job at the end
176          while (item.m_next != null) {item = item.m_next;}
177          item.m_next = posted;           
178       }
179    }
180 
181    /**
182     * Clears the running thread after the queue has been stopped. <br> 
183     * After this call, this worker queue is unusable and can be garbaged.
184     */
185    protected void clear() 
186    {
187       m_queueThread = null;
188       m_currentJob = null;
189    }
190    
191    /**
192     * Creates the loop that will get the next job and process it. <br>
193     * Override in subclasses to create a custom loop.
194     */
195    protected Runnable   createQueueLoop() {
196       return new QueueLoop();
197    }
198    
199    /**
200     * Class that loops getting the next job to be executed and then 
201     * executing it, in the worker thread.
202     */
203    protected class QueueLoop
204       implements Runnable   
205    {
206       public void run() 
207       {
208          try
209          {
210             while (true) 
211             {
212                try 
213                {
214                   if (isInterrupted()) 
215                   {
216                      flush();
217                      break;
218                   }
219                   else 
220                   {
221                      getJob().execute();
222                   }
223                }
224                catch (InterruptedException   e) 
225                {
226                   try {
227                      flush();
228                   }
229                   catch (Exception   ignored) {}
230                   break;
231                }
232                catch (Exception   e) {
233                   ThrowableHandler.add(ThrowableHandler.Type.ERROR, e);
234                }
235             }
236          }
237          finally {
238             clear();
239          }
240       }
241       
242       protected void flush() throws Exception  
243       {
244          // Empty the queue of the posted jobs and exit
245          while (m_currentJob != null) 
246          {
247             m_currentJob.m_job.execute();
248             m_currentJob = m_currentJob.m_next;
249          }
250       }
251    }
252    
253    /**
254     * Simple linked cell, that has only a reference to the next job.
255     */
256    private class JobItem 
257    {
258       private Executable m_job;
259       private JobItem m_next;
260       private JobItem(Executable job) {m_job = job;}
261    }
262 }
263
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags