QueueWorker


1   /**
2    * Redistribution and use of this software and associated documentation
3    * ("Software"), with or without modification, are permitted provided
4    * that the following conditions are met:
5    *
6    * 1. Redistributions of source code must retain copyright
7    *    statements and notices.  Redistributions must also contain a
8    *    copy of this document.
9    *
10   * 2. Redistributions in binary form must reproduce the
11   *    above copyright notice, this list of conditions and the
12   *    following disclaimer in the documentation and/or other
13   *    materials provided with the distribution.
14   *
15   * 3. The name "Exolab" must not be used to endorse or promote
16   *    products derived from this Software without prior written
17   *    permission of Exoffice Technologies.  For written permission,
18   *    please contact info@exolab.org.
19   *
20   * 4. Products derived from this Software may not be called "Exolab"
21   *    nor may "Exolab" appear in their names without prior written
22   *    permission of Exoffice Technologies. Exolab is a registered
23   *    trademark of Exoffice Technologies.
24   *
25   * 5. Due credit should be given to the Exolab Project
26   *    (http://www.exolab.org/).
27   *
28   * THIS SOFTWARE IS PROVIDED BY EXOFFICE TECHNOLOGIES AND CONTRIBUTORS
29   * ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT
30   * NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
31   * FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL
32   * EXOFFICE TECHNOLOGIES OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
33   * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
34   * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
35   * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
36   * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
37   * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
38   * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
39   * OF THE POSSIBILITY OF SUCH DAMAGE.
40   *
41   * Copyright 2002-2004 (C) Exoffice Technologies Inc. All Rights Reserved.
42   *
43   * $Id: QueueWorker.java,v 1.1 2004/11/26 01:50:35 tanderson Exp $
44   */
45  package org.exolab.jms.common.threads;
46  
47  import java.util.LinkedList  ;
48  
49  import org.apache.commons.logging.Log;
50  import org.apache.commons.logging.LogFactory;
51  
52  
53  /**
54   * This class creates an internal thread, adds it to the idleThread queue
55   * and starts it running. The thread will then block on an internal queue
56   * waitUntilWork for a runnable object to be passed to it. Once invoked
57   * the thread will then call the objects run method.
58   *
59   * @version     $Revision: 1.1 $ $Date: 2004/11/26 01:50:35 $
60   * @author      <a HREF="mailto:tma@netspace.net.au">Tim Anderson</a>
61   * @see         ThreadPool
62   */
63  class QueueWorker {
64  
65      /**
66       * The owning thread pool
67       */
68      private ThreadPool _pool;
69  
70      /**
71       * The queue of pending work
72       */
73      private LinkedList   _queue = new LinkedList  ();
74  
75      /**
76       * The worker thread
77       */
78      private Thread   _worker;
79  
80      /**
81       * A flag used to notify the thread to stop
82       */
83      private volatile boolean _stop = false;
84  
85      /**
86       * The logger
87       */
88      private static final Log _log = LogFactory.getLog(QueueWorker.class);
89  
90  
91      /**
92       * The constructor creates the worker thread and runs it.
93       * The new thread is given a unique id and added to the idle work queue.
94       * The thread will then sleep until it is allocated work.
95       *
96       * @param pool the owning thread pool
97       * @param group the thread group to add the worker thread to
98       * @param name the name of the thread. This must be unique.
99       */
100     public QueueWorker(ThreadPool pool, ThreadGroup   group, String   name) {
101         _pool = pool;
102 
103         Runnable   worker = new Runnable  () {
104             public void run() {
105                 try {
106                     runWork();
107                 } catch (Exception   exception) {
108                     _log.error("Thread " + _worker.getName()
109                                + ": terminating on exception", exception);
110                 }
111             }
112         };
113 
114         // Create the new internal thread and start it all off.
115         _worker = new Thread  (group, worker, name);
116         _worker.setDaemon(true);
117         _worker.start();
118     }
119 
120     /**
121      * Queues a {@link Runnable} object to be executed.
122      *
123      * @param target the object to execute
124      * @param listener the listener to notify when execution completes
125      * (may be null)
126      */
127     public void add(Runnable   target, CompletionListener listener) {
128         synchronized (_queue) {
129             _queue.add(new Executor(target, listener));
130             _queue.notify();
131         }
132     }
133 
134     /**
135      * Request the worker thread to stop
136      */
137     public void stop() {
138         _stop = true;
139         _worker.interrupt();
140     }
141 
142     /**
143      * Determines if the worker is alive
144      *
145      * @return <code>true</code> if the worker is alive
146      */
147     public boolean isAlive() {
148         return _worker.isAlive();
149     }
150 
151     /**
152      * Executes queued {@link Runnable} instances, until {@link #stop} is
153      * invoked.
154      */
155     private void runWork() {
156         while (!_stop) {
157             Runnable   target = null;
158             synchronized (_queue) {
159                 if (_queue.isEmpty()) {
160                     try {
161                         _queue.wait();
162                     } catch (InterruptedException   ignore) {
163                     }
164                 } else {
165                     target = (Executor) _queue.removeFirst();
166                 }
167             }
168 
169             if (!_stop && target != null) {
170                 try {
171                     _pool.execute(target);
172                 } catch (InterruptedException   ignore) {
173                 }
174             }
175         }
176     }
177 
178     /**
179      * Executes a {@link Runnable}
180      */
181     private static class Executor implements Runnable   {
182 
183         /**
184          * The target to execute
185          */
186         private Runnable   _target;
187 
188         /**
189          * The listener to notify on completion. May be <code>null</code>
190          */
191         private CompletionListener _listener;
192 
193         /**
194          * Construct a new <code>Executor</code>
195          *
196          * @param target the target to run
197          * @param listener the listener to notify on completion.
198          * May be <code>null</code>
199          */
200         public Executor(Runnable   target, CompletionListener listener) {
201             _target = target;
202             _listener = listener;
203         }
204 
205         /**
206          * Run the target
207          */
208         public void run() {
209             try {
210                 _target.run();
211             } catch (Throwable   exception) {
212                 _log.error("Thread " + Thread.currentThread().getName()
213                            + " - uncaught exception fell through from run",
214                            exception);
215             } finally {
216                 if (_listener != null) {
217                     _listener.completed(_target);
218                 }
219             }
220         }
221     }
222 
223 }
224
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags