Index: common/src/main/org/jboss/util/threadpool/BasicThreadPool.java
===================================================================
RCS file: /cvsroot/jboss/jboss-common/src/main/org/jboss/util/threadpool/BasicThreadPool.java,v
retrieving revision 1.7.2.1
diff -b -B -U8 -r1.7.2.1 BasicThreadPool.java
--- common/src/main/org/jboss/util/threadpool/BasicThreadPool.java	20 Dec 2004 17:42:35 -0000	1.7.2.1
+++ common/src/main/org/jboss/util/threadpool/BasicThreadPool.java	26 Aug 2005 23:18:40 -0000
@@ -54,17 +54,17 @@
 
    /** The pooled executor */
    private MinPooledExecutor executor;
 
    /** The queue */
    private BoundedLinkedQueue queue;
 
    /** The thread group */
-   private ThreadGroup threadGroup = JBOSS_THREAD_GROUP;
+   private ThreadGroup threadGroup;
 
    /** The last thread number */
    private SynchronizedInt lastThreadNumber = new SynchronizedInt(0);
 
    /** Has the pool been stopped? */
    private SynchronizedBoolean stopped = new SynchronizedBoolean(false);
    /** The Heap<TimeoutInfo> of tasks ordered by their completion timeout */
    private Heap tasksWithTimeouts = new Heap(13);
@@ -81,36 +81,49 @@
     * Create a new thread pool
     */
    public BasicThreadPool()
    {
       this("ThreadPool");
    }
 
    /**
-    * Create a new thread pool with a default queue size of 1024, min/max pool
-    * sizes of 100 and a keep alive of 60 seconds.
+    * Create a new thread pool with a default queue size of 1024, max pool
+    * size of 100, min pool size of 4, and a keep alive of 60 seconds.
     *
     * @param name the pool name
     */
    public BasicThreadPool(String name)
    {
+      this(name, JBOSS_THREAD_GROUP);
+   }
+
+   /**
+    * Create a new thread pool with a default queue size of 1024, max pool
+    * size of 100, min pool size of 4, and a keep alive of 60 seconds.
+    *
+    * @param name the pool name
+    * @param threadGroup threadGroup
+    */
+   public BasicThreadPool(String name, ThreadGroup threadGroup)
+   {
       trace = log.isTraceEnabled();
       ThreadFactory factory = new ThreadPoolThreadFactory();
 
       queue = new BoundedLinkedQueue(1024);
 
       executor = new MinPooledExecutor(queue, 100);
-      executor.setMinimumPoolSize(100);
+      executor.setMinimumPoolSize(4);
       executor.setKeepAliveTime(60 * 1000);
       executor.setThreadFactory(factory);
       executor.abortWhenBlocked();
 
       poolNumber = lastPoolNumber.increment();
       setName(name);
+      this.threadGroup = threadGroup;
    }
 
    // Public --------------------------------------------------------
 
    // ThreadPool ----------------------------------------------------
 
    public void stop(boolean immediate)
    {
@@ -173,22 +186,28 @@
       BasicTaskWrapper wrapper = new BasicTaskWrapper(task);
       runTaskWrapper(wrapper);
    }
 
    public void run(Runnable runnable)
    {
       run(runnable, 0, 0);
    }
+
    public void run(Runnable runnable, long startTimeout, long completeTimeout)
    {
       RunnableTaskWrapper wrapper = new RunnableTaskWrapper(runnable, startTimeout, completeTimeout);
       runTaskWrapper(wrapper);      
    }
 
+   public ThreadGroup getThreadGroup()
+   {
+      return threadGroup;
+   }
+
    // ThreadPoolMBean implementation --------------------------------
 
    public String getName()
    {
       return name;
    }
 
    public void setName(String name)
Index: common/src/main/org/jboss/util/threadpool/MinPooledExecutor.java
===================================================================
RCS file: /cvsroot/jboss/jboss-common/src/main/org/jboss/util/threadpool/MinPooledExecutor.java,v
retrieving revision 1.3.2.1
diff -b -B -U8 -r1.3.2.1 MinPooledExecutor.java
--- common/src/main/org/jboss/util/threadpool/MinPooledExecutor.java	6 Apr 2005 21:10:47 -0000	1.3.2.1
+++ common/src/main/org/jboss/util/threadpool/MinPooledExecutor.java	26 Aug 2005 23:18:40 -0000
@@ -43,28 +43,28 @@
 
    // Static --------------------------------------------------------
 
    // Constructors --------------------------------------------------
 
    /**
     * Construct a new executor
     * 
-    * @param poolSize the minimum pool size
+    * @param poolSize the maximum pool size
     */
    public MinPooledExecutor(int poolSize)
    {
       super(poolSize);
    }
    
    /**
     * Construct a new executor
     * 
     * @param channel the queue for any requests
-    * @param poolSize the minimum pool size
+    * @param poolSize the maximum pool size
     */
    public MinPooledExecutor(Channel channel, int poolSize)
    {
       super(channel, poolSize);
    }
 
    // Public --------------------------------------------------------
 
Index: common/src/main/org/jboss/util/timeout/Timeout.java
===================================================================
RCS file: /cvsroot/jboss/jboss-common/src/main/org/jboss/util/timeout/Timeout.java,v
retrieving revision 1.1
diff -b -B -U8 -r1.1 Timeout.java
--- common/src/main/org/jboss/util/timeout/Timeout.java	16 Feb 2002 10:33:09 -0000	1.1
+++ common/src/main/org/jboss/util/timeout/Timeout.java	26 Aug 2005 23:18:40 -0000
@@ -22,11 +22,11 @@
     *  the timeout will not happen, or (in case of late cancel) the
     *  timeout has happened and the timeout callback function has returned.
     *
     *  On return from this method this instance should no longer be
     *  used. The reason for this is that an implementation may reuse
     *  cancelled timeouts, and at return the instance may already be
     *  in use for another timeout.
     */
-   public void cancel();
+   public boolean cancel();
 }
 
Index: common/src/main/org/jboss/util/timeout/TimeoutFactory.java
===================================================================
RCS file: /cvsroot/jboss/jboss-common/src/main/org/jboss/util/timeout/TimeoutFactory.java,v
retrieving revision 1.2.6.1
diff -b -B -U8 -r1.2.6.1 TimeoutFactory.java
--- common/src/main/org/jboss/util/timeout/TimeoutFactory.java	17 Oct 2004 16:54:21 -0000	1.2.6.1
+++ common/src/main/org/jboss/util/timeout/TimeoutFactory.java	26 Aug 2005 23:18:40 -0000
@@ -1,567 +1,135 @@
 /*
  * JBoss, the OpenSource J2EE webOS
  *
  * Distributable under LGPL license.
  * See terms of license at gnu.org.
  */
 package org.jboss.util.timeout;
 
+import java.util.Date;
+import java.util.Timer;
+import java.util.TimerTask;
 import org.jboss.util.ThrowableHandler;
 import org.jboss.util.threadpool.BasicThreadPool;
 import org.jboss.util.threadpool.ThreadPool;
 
 /**
  *  The timeout factory.
+ * Upon timeout, calls a thread pool.
+ * Implemented as a wrapper around java.util.Timer.
  *
- *  This is written with performance in mind. In case of <code>n</code>
- *  active timeouts, creating, cancelling and firing timeouts all operate
- *  in time <code>O(log(n))</code>.
- *
- *  If a timeout is cancelled, the timeout is not discarded. Instead the
- *  timeout is saved to be reused for another timeout. This means that if
- *  no timeouts are fired, this class will eventually operate without
- *  allocating anything on the heap.
- *
- *  @author <a href="osh@sparre.dk">Ole Husgaard</a>
  *  @version $Revision: 1.2.6.1 $
  */
 public class TimeoutFactory
 {
-   //  Code commented out with the mark "INV:" are runtime checks
-   //  of invariants that are not needed for a production system.
-   //  For problem solving, you can remove these comments.
-   //  Multithreading notes:
-   //
-   //  While a TimeoutImpl is enqueued, its index field contains the index
-   //  of the instance in the queue; that is, for 1 <= n <= size,
-   //  q[n].index = n.
-   //  Modifications of an enqueued TimeoutImpl instance may only happen
-   //  in code synchronized on the TimeoutFactory instance that has it
-   //  enqueued.
-   //  Modifications on the priority queue may only happen while running in
-   //  code synchronized on the TimeoutFactory instance that holds the queue.
-   //  When a TimeoutImpl instance is no longer enqueued, its index field
-   //  changes to one of the negative constants declared in the TimeoutImpl
-   //  class.
-   //  When a TimeoutImpl is not in use, its index field is TimeoutImpl.DONE
-   //  and it is on the freeList.
-   //
-   //  Cancellation may race with the timeout.
-   //  To avoid problems with this, the TimeoutImpl index field is set to
-   //  TimeoutImpl.TIMEOUT when the TimeoutImpl is taken out of the queue.
-   //  Finally the index field is set to TimeoutImpl.DONE, and
-   //  the TimeoutImpl instance is discarded.
-
-   /** The threadpool used to invoke timeouts */
-   private ThreadPool threadPool = new BasicThreadPool("Timeouts");
-
-   /**
-    *  Our private Timeout implementation.
-    */
-   private class TimeoutImpl implements Timeout
-   {
-      static final int DONE = -1; // done, may be finalized or reused
-
-      static final int TIMEOUT = -2; // target being called
 
-      int index; // index in queue, or one of constants above.
+   private static ThreadPool DEFAULT_TP = new BasicThreadPool("Timeouts");
+   private Timer timer;
+   private ThreadPool threadPool;
 
-      long time; // time to fire
+   static TimeoutFactory singleton;
+   static int count = 0;
 
-      TimeoutTarget target; // target to fire at
-
-      TimeoutImpl nextFree; // next on free list
-
-      public void cancel()
+   /** Construct a new factory with a specific thread pool. */
+   public TimeoutFactory(ThreadPool threadPool)
       {
-         TimeoutFactory.this.dropTimeout(this);
+      this.threadPool = threadPool;
+      this.timer = new Timer(true);
+      timer.schedule(new TimerTask() {
+         public void run() {
+            Thread.currentThread().setName("TimeoutFactory-" + count++);
       }
+      }, 0L);
    }
 
-   /**
-    *  A worker thread that fires the timeout.
-    */
-   private static class TimeoutWorker implements Runnable
+   /** Construct a new factory with a default thread pool. */
+   public TimeoutFactory()
    {
-      private TimeoutImpl work;
-
-      /**
-       *  Create a new instance.
-       *
-       *  @param work The timeout that should be fired.
-       */
-      TimeoutWorker(TimeoutImpl work)
-      {
-         this.work = work;
+      this(DEFAULT_TP);
       }
 
       /**
-       *  Override to fire the timeout.
+    * Schedules a new timeout.
        */
-      public void run()
-      {
-         try
-         {
-            work.target.timedOut(work);
-         }
-         catch (Throwable t)
-         {
-            ThrowableHandler.add(ThrowableHandler.Type.ERROR, t);
-         }
-         synchronized (work)
+   public TimerTask schedule(long time, Runnable run)
          {
-            work.index = TimeoutImpl.DONE;
+      TimerTask t = new PooledRunner(run);
+      timer.schedule(t, new Date(time));
+      return t;
          }
-      }
-   }
-
-   /** Linked list of free TimeoutImpl instances. */
-   private TimeoutImpl freeList;
-
-   /** The size of the timeout queue. */
-   private int size;
 
    /**
-    *  Our priority queue.
-    *
-    *  This is a balanced binary tree. If nonempty, the root is at index 1,
-    *  and all nodes are at indices 1..size. Nodes with index greater than
-    *  size are null. Index 0 is never used.
-    *  Children of the node at index <code>j</code> are at <code>j*2</code>
-    *  and <code>j*2+1</code>. The children of a node always fire the timeout
-    *  no earlier than the node.
-    *
-    *
-    *  Or, more formally:
-    *
-    *  Only indices <code>1</code>..<code>size</code> of this array are used.
-    *  All other indices contain the null reference.
-    *  This array represent a balanced binary tree.
-    *
-    *  If <code>size</code> is <code>0</code> the tree is empty, otherwise
-    *  the root of the tree is at index <code>1</code>.
-    *
-    *  Given an arbitrary node at index <code>n</code> that is not the root
-    *  node, the parent node of <code>n</code> is at index <code>n/2</code>.
-    *
-    *  Given an arbitrary node at index <code>n</code>; if
-    *  <code>2*n <= size</code> the node at <code>n</code> has its left child
-    *  at index <code>2*n</code>, otherwise the node at <code>n</code> has
-    *  no left child.
-    *
-    *  Given an arbitrary node at index <code>n</code>; if
-    *  <code>2*n+1 <= size</code> the node at <code>n</code> has its right child
-    *  at index <code>2*n+1</code>, otherwise the node at <code>n</code> has
-    *  no right child.
-    *
-    *  The priority function is called T. Given a node <code>n</code>,
-    *  <code>T(n)</code> denotes the absolute time (in milliseconds since
-    *  the epoch) that the timeout for node <code>n</code> should happen.
-    *  Smaller values of <code>T</code> means higher priority.
-    *
-    *  The tree satisfies the following invariant:
-    *  <i>
-    *  For any node <code>n</code> in the tree:
-    *  If node <code>n</code> has a left child <code>l</code>,
-    *  <code>T(n) <= T(l)</code>.
-    *  If node <code>n</code> has a right child <code>r</code>,
-    *  <code>T(n) <= T(r)</code>.
-    *  </i>
-    *
-    *
-    *  The invariant may be temporarily broken while executing synchronized
-    *  on <code>this</code> instance, but is always reestablished before
-    *  leaving the synchronized code.
-    *
-    *  The node at index <code>1</code> is always the first node to timeout,
-    *  as can be deduced from the invariant.
-    *
-    *  For the following algorithm pseudocode, the operation
-    *  <code>swap(n,m)</code> denotes the exchange of the nodes at indices
-    *  <code>n</code> and <code>m</code> in the tree.
-    *
-    *  Insertion of a new node happend as follows:
-    *  <pre>
-    *    IF size = q.length THEN
-    *      "expand q array to be larger";
-    *    ENDIF
-    *    size <- size + 1;
-    *    q[size] <- "new node";
-    *    n <- size;
-    *    WHILE n > 1 AND T(n/2) > T(n) DO
-    *      swap(n/2, n);
-    *      n <- n/2;
-    *    ENDWHILE
-    *  </pre>
-    *  Proof that this insertion algorithm respects the invariant is left to
-    *  the interested reader.
-    *
-    *  The removal algorithm is a bit more complicated. To remove the node
-    *  at index <code>n</code>:
-    *  <pre>
-    *    swap(n, size);
-    *    size <- size - 1;
-    *    IF n > 1 AND T(n/2) > T(n) THEN
-    *      WHILE n > 1 AND T(n/2) > T(n) DO
-    *        swap(n/2, n);
-    *        n <- n/2;
-    *      ENDWHILE
-    *    ELSE
-    *      WHILE 2*n <= size DO
-    *        IF 2*n+1 <= size THEN
-    *          // Both children present
-    *          IF T(2*n) <= T(2*n+1) THEN
-    *            IF T(n) <= T(2*n) THEN
-    *              EXIT;
-    *            ENDIF
-    *            swap(n, 2*n);
-    *            n <- 2*n;
-    *          ELSE
-    *            IF T(n) <= T(2*n+1) THEN
-    *              EXIT;
-    *            ENDIF
-    *            swap(n, 2*n+1);
-    *            n <- 2*n+1;
-    *          ENDIF
-    *        ELSE
-    *          // Only left child, right child not present.
-    *          IF T(n) <= T(2*n) THEN
-    *            EXIT;
-    *          ENDIF
-    *          swap(n, 2*n);
-    *          n <- 2*n;
-    *        ENDIF
-    *      ENDWHILE
-    *    ENDIF
-    *  </pre>
-    *  Proof that this removal algorithm respects the invariant is left to
-    *  the interested reader. Really, I am not going to prove it here.
-    *
-    *  If you are interested, you can find this data structure and its
-    *  associated operations in most textbooks on algorithmics.
-    *
-    *  @see checkTree
-    */
-   private TimeoutImpl[] q;
-
-   /**
-    *  Debugging helper.
+    * Schedule a new timeout.
     */
-   private void assertExpr(boolean expr)
-   {
-      if (!expr)
-      {
-         RuntimeException ex = new RuntimeException("***** assert failed *****");
-         try
+   public Timeout schedule(long time, TimeoutTarget target)
          {
-            Thread.sleep(30000);
-         }
-         catch (Exception e)
-         {
-         }
-      }
+      TimeoutImpl t = new TimeoutImpl(target);
+      timer.schedule(t, new Date(time));
+      return t;
    }
 
-   /**
-    *  Check invariants of the queue.
-    */
-   private void checkTree()
-   {
-      assertExpr(size >= 0);
-      assertExpr(size < q.length);
-      assertExpr(q[0] == null);
-      if (size > 0)
-      {
-         assertExpr(q[1] != null);
-         assertExpr(q[1].index == 1);
-         for (int i = 2; i <= size; ++i)
-         {
-            assertExpr(q[i] != null);
-            assertExpr(q[i].index == i);
-            assertExpr(q[i >> 1].time <= q[i].time); // parent fires first
-         }
-         for (int i = size + 1; i < q.length; ++i)
-            assertExpr(q[i] == null);
-      }
+   private synchronized static TimeoutFactory getSingleton() {
+      if (singleton != null)
+         return singleton;
+      singleton = new TimeoutFactory(DEFAULT_TP);
+      return singleton;
    }
 
    /**
-    *  Check invariants of the free list.
-    */
-   private void checkFreeList()
-   {
-      TimeoutImpl to = freeList;
-      while (to != null)
-      {
-         assertExpr(to.index == TimeoutImpl.DONE);
-         to = to.nextFree;
-      }
-   }
-
-   /**
-    *  Swap two nodes in the tree.
+    * Schedule a new timeout.
     */
-   private void swap(int a, int b)
+   public static Timeout createTimeout(long time, TimeoutTarget target)
    {
-      // INV: assertExpr(a > 0);
-      // INV: assertExpr(a <= size);
-      // INV: assertExpr(b > 0);
-      // INV: assertExpr(b <= size);
-      // INV: assertExpr(q[a] != null);
-      // INV: assertExpr(q[b] != null);
-      // INV: assertExpr(q[a].index == a);
-      // INV: assertExpr(q[b].index == b);
-      TimeoutImpl temp = q[a];
-      q[a] = q[b];
-      q[a].index = a;
-      q[b] = temp;
-      q[b].index = b;
+      return getSingleton().schedule(time, target);
    }
 
-   /**
-    *  A new node has been added at index <code>index</code>.
-    *  Normalize the tree by moving the new node up the tree.
-    *
-    *  @return True iff the tree was modified.
-    */
-   private boolean normalizeUp(int index)
+   public void cancel()
    {
-      // INV: assertExpr(index > 0);
-      // INV: assertExpr(index <= size);
-      // INV: assertExpr(q[index] != null);
-      if (index == 1)
-         return false; // at root
-      boolean ret = false;
-      long t = q[index].time;
-      int p = index >> 1;
-      while (q[p].time > t)
-      {
-         // INV: assertExpr(q[index].time == t);
-         swap(p, index);
-         ret = true;
-         if (p == 1)
-            break; // at root
-         index = p;
-         p >>= 1;
-      }
-      return ret;
+      timer.cancel();
    }
 
    /**
-    *  Remove a node from the tree and normalize.
-    *
-    *  @return The removed node.
+    *  Our private Timeout implementation.
     */
-   private TimeoutImpl removeNode(int index)
+   private class TimeoutImpl extends TimerTask implements Timeout
    {
-      // INV: assertExpr(index > 0);
-      // INV: assertExpr(index <= size);
-      TimeoutImpl res = q[index];
-      // INV: assertExpr(res != null);
-      // INV: assertExpr(res.index == index);
-      if (index == size)
-      {
-         --size;
-         q[index] = null;
-         return res;
-      }
-      swap(index, size); // Exchange removed node with last leaf node
-      --size;
-      // INV: assertExpr(res.index == size + 1);
-      q[res.index] = null;
-      if (normalizeUp(index))
-         return res; // Node moved up, so it shouldn't move down
-      long t = q[index].time;
-      int c = index << 1;
-      while (c <= size)
-      {
-         // INV: assertExpr(q[index].time == t);
-         TimeoutImpl l = q[c];
-         // INV: assertExpr(l != null);
-         // INV: assertExpr(l.index == c);
-         if (c + 1 <= size)
-         {
-            // two children, swap with smallest
-            TimeoutImpl r = q[c + 1];
-            // INV: assertExpr(r != null);
-            // INV: assertExpr(r.index == c+1);
-            if (l.time <= r.time)
-            {
-               if (t <= l.time)
-                  break; // done
-               swap(index, c);
-               index = c;
-            }
-            else
-            {
-               if (t <= r.time)
-                  break; // done
-               swap(index, c + 1);
-               index = c + 1;
-            }
-         }
-         else
-         { // one child
-            if (t <= l.time)
-               break; // done
-            swap(index, c);
-            index = c;
-         }
-         c = index << 1;
-      }
-      return res;
-   }
+      private TimeoutTarget target; // target to fire at
 
-   /**
-    *  Create a new timeout.
-    */
-   private synchronized Timeout newTimeout(long time, TimeoutTarget target)
+      public TimeoutImpl(TimeoutTarget target)
    {
-      // INV: checkTree();
-      // INV: assertExpr(size < q.length);
-      if (++size == q.length)
-      {
-         TimeoutImpl[] newQ = new TimeoutImpl[2 * q.length];
-         System.arraycopy(q, 0, newQ, 0, q.length);
-         q = newQ;
-      }
-      // INV: assertExpr(size < q.length);
-      // INV: assertExpr(q[size] == null);
-      TimeoutImpl timeout;
-      if (freeList != null)
-      {
-         timeout = q[size] = freeList;
-         freeList = timeout.nextFree;
-         // INV: checkFreeList();
-         // INV: assertExpr(timeout.index == TimeoutImpl.DONE);
-      }
-      else
-         timeout = q[size] = new TimeoutImpl();
-      timeout.index = size;
-      timeout.time = time;
-      timeout.target = target;
-      normalizeUp(size);
-      if (timeout.index == 1)
-         notify();
-      // INV: checkTree();
-      return timeout;
+         if (target == null)
+            throw new IllegalArgumentException("Null target");
+         this.target = target;
    }
 
-   /**
-    *  Cancel a timeout.
-    */
-   private void dropTimeout(TimeoutImpl timeout)
-   {
-      synchronized (this)
+      public void run()
       {
-         if (timeout.index > 0)
+         Runnable r = new Runnable() {
+            public void run()
          {
-            // Active timeout, remove it.
-            // INV: assertExpr(q[timeout.index] == timeout);
-            // INV: checkTree();
-            removeNode(timeout.index);
-            // INV: checkTree();
-            timeout.index = TimeoutImpl.DONE;
-            timeout.nextFree = freeList;
-            freeList = timeout;
-            // INV: checkFreeList();
-            return;
-         }
+               target.timedOut(TimeoutImpl.this);
       }
+         };
+         threadPool.run(r);
    }
 
-   /**
-    *  Timeout worker method.
-    *  This method never returns. Whenever it is time to do a timeout,
-    *  the callback method is called from here.
-    */
-   private void doWork()
-   {
-      while (true)
-      {
-         TimeoutImpl work = null;
-         // Look for work
-         synchronized (this)
-         {
-            if (size == 0)
-            {
-               try
-               {
-                  wait();
-               }
-               catch (InterruptedException ex)
-               {
-               }
-            }
-            else
-            {
-               long now = System.currentTimeMillis();
-               if (q[1].time > now)
-               {
-                  try
-                  {
-                     wait(q[1].time - now);
                   }
-                  catch (InterruptedException ex)
-                  {
-                  }
-               }
-               if (size > 0 && q[1].time <= System.currentTimeMillis())
-               {
-                  work = removeNode(1);
-                  work.index = TimeoutImpl.TIMEOUT;
-               }
-            }
-         }
-         // Do work, if any
-         if (work != null)
+
+   private class PooledRunner extends TimerTask
          {
-            // Create a new thread to do the callback.
-            TimeoutWorker worker = new TimeoutWorker(work);
-            threadPool.run(worker);
-         }
-      }
-   }
 
-   /** Our singleton instance. */
-   static private TimeoutFactory singleton;
+      Runnable run;
 
-   /** Our private constructor. */
-   private TimeoutFactory()
+      public PooledRunner(Runnable run)
    {
-      freeList = null;
-      size = 0;
-      q = new TimeoutImpl[16];
+         this.run = run;
    }
 
-   /**
-    *  Initialize class.
-    *  The will initialize the singleton and create a single
-    *  worker thread.
-    */
-   static
-   {
-      singleton = new TimeoutFactory();
-      Thread thread = new Thread("TimeoutFactory")
-      {
          public void run()
          {
-            singleton.doWork();
-         }
-      };
-      thread.setDaemon(true);
-      thread.start();
+         threadPool.run(this.run);
    }
 
-   /**
-    *  Schedule a new timeout.
-    */
-   static public Timeout createTimeout(long time, TimeoutTarget target)
-   {
-      if (time <= 0)
-         throw new IllegalArgumentException("Time not positive");
-      if (target == null)
-         throw new IllegalArgumentException("Null target");
-      return singleton.newTimeout(time, target);
    }
+
 }
\ No newline at end of file
Index: common/src/main/org/jboss/xb/binding/metadata/unmarshalling/impl/DocumentBindingFactoryImpl.java
===================================================================
RCS file: /cvsroot/jboss/jboss-common/src/main/org/jboss/xb/binding/metadata/unmarshalling/impl/DocumentBindingFactoryImpl.java,v
retrieving revision 1.1.2.2
diff -b -B -U8 -r1.1.2.2 DocumentBindingFactoryImpl.java
--- common/src/main/org/jboss/xb/binding/metadata/unmarshalling/impl/DocumentBindingFactoryImpl.java	25 Jul 2005 16:19:39 -0000	1.1.2.2
+++ common/src/main/org/jboss/xb/binding/metadata/unmarshalling/impl/DocumentBindingFactoryImpl.java	26 Aug 2005 23:18:41 -0000
@@ -111,17 +111,17 @@
       protected final DocumentBinding doc;
 
       protected AbstractDocumentBinding(DocumentBinding doc)
       {
          if(doc == null)
          {
             // todo
             doc = DocumentBindingFactory.newInstance().newDocumentBindingStack();
-            ((DocumentBindingStackImpl)doc).push(this);
+            ((DocumentBindingFactoryImpl.DocumentBindingStackImpl)doc).push(this);
          }
          this.doc = doc;
       }
 
       public NamespaceBinding getNamespace(String namespaceUri)
       {
          return doc.getNamespace(namespaceUri);
       }
Index: messaging/src/etc/server/default/deploy/hsqldb-jdbc2-service.xml
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/etc/server/default/deploy/hsqldb-jdbc2-service.xml,v
retrieving revision 1.7.2.7
diff -b -B -U8 -r1.7.2.7 hsqldb-jdbc2-service.xml
--- messaging/src/etc/server/default/deploy/hsqldb-jdbc2-service.xml	13 Aug 2005 20:47:52 -0000	1.7.2.7
+++ messaging/src/etc/server/default/deploy/hsqldb-jdbc2-service.xml	26 Aug 2005 23:18:41 -0000
@@ -11,16 +11,17 @@
 
   <!--
      | The destination manager is the core service within JBossMQ
   -->
   <mbean code="org.jboss.mq.server.jmx.DestinationManager" name="jboss.mq:service=DestinationManager">
     <depends optional-attribute-name="MessageCache">jboss.mq:service=MessageCache</depends>
     <depends optional-attribute-name="PersistenceManager">jboss.mq:service=PersistenceManager</depends>
     <depends optional-attribute-name="StateManager">jboss.mq:service=StateManager</depends>
+    <depends optional-attribute-name="ThreadPool">jboss.mq:service=ThreadPool</depends>
     <depends>jboss:service=Naming</depends>
   </mbean>
 
   <!--
      | The MessageCache decides where to put JBossMQ message that
      | are sitting around waiting to be consumed by a client.
      |
      | The memory marks are in Megabytes.  Once the JVM memory usage hits
Index: messaging/src/etc/server/default/deploy/jbossmq-service.xml
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/etc/server/default/deploy/jbossmq-service.xml,v
retrieving revision 1.24.2.1
diff -b -B -U8 -r1.24.2.1 jbossmq-service.xml
--- messaging/src/etc/server/default/deploy/jbossmq-service.xml	1 Jun 2005 05:56:44 -0000	1.24.2.1
+++ messaging/src/etc/server/default/deploy/jbossmq-service.xml	26 Aug 2005 23:18:41 -0000
@@ -29,16 +29,36 @@
       <security>
         <role name="guest" read="true" write="true" create="true"/>
       </security>
     </attribute>
     <attribute name="SecurityDomain">java:/jaas/jbossmq</attribute>
     <depends optional-attribute-name="NextInterceptor">jboss.mq:service=DestinationManager</depends>
   </mbean>
 
+  <!-- A Thread pool service -->
+  <mbean code="org.jboss.util.threadpool.BasicThreadPool"
+     name="jboss.mq:service=ThreadPool">
+     <attribute name="Name">JMSThread</attribute>
+     <attribute name="ThreadGroupName">JBossMQ Server Threads</attribute>
+     <!-- The max number of threads in the pool -->
+     <attribute name="MaximumPoolSize">10</attribute>
+     <!-- The max number of tasks before the queue is full -->
+     <attribute name="MaximumQueueSize">1000</attribute>
+     <!-- The behavior of the pool when a task is added and the queue is full.
+     abort - a RuntimeException is thrown
+     run - the calling thread executes the task
+     wait - the calling thread blocks until the queue has room
+     discard - the task is silently discarded without being run
+     discardOldest - check to see if a task is about to complete and enque
+        the new task if possible, else run the task in the calling thread
+     -->
+     <attribute name="BlockingMode">run</attribute>
+  </mbean>
+
   <!--
      | The ClientMonitorInterceptor disconnects clients that have been idle for to long.
      | This interceptor is not enabled by default since the server might disconnect clients
      | when the it is under high load.
    -->
   <!--
   <mbean code="org.jboss.mq.server.jmx.ClientMonitorInterceptor" name="jboss.mq:service=ClientMonitorInterceptor">
     <attribute name="ClientTimeout">80000</attribute>
Index: messaging/src/main/org/jboss/mq/AcknowledgementRequest.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/AcknowledgementRequest.java,v
retrieving revision 1.6
diff -b -B -U8 -r1.6 AcknowledgementRequest.java
--- messaging/src/main/org/jboss/mq/AcknowledgementRequest.java	8 Dec 2003 15:02:52 -0000	1.6
+++ messaging/src/main/org/jboss/mq/AcknowledgementRequest.java	26 Aug 2005 23:18:41 -0000
@@ -24,31 +24,65 @@
 {
    // Constants -----------------------------------------------------
 
    /** The serialVersionUID */
    private static final long serialVersionUID = -2227528634302168874L;
    
    // Attributes ----------------------------------------------------
 
-   /** Is it an acknowledgement */
-   public boolean isAck;
+   /** Message ack */
+   public static byte ACK  = 1;
+
+   /** Message nacked */
+   public static byte NACK = 0;
+
+   /** Message expired */
+   public static byte EACK = 2;
+
+   /** ACK state */
+   public byte ack = ACK;
    /** The destination */
    public Destination destination = null;
    /** The messageID */
    public String messageID = null;
    /** The subscriberId */
    public int subscriberId;
    
    // Static --------------------------------------------------------
    
    // Constructors --------------------------------------------------
    
+   public AcknowledgementRequest()
+   {
+      this(true);
+   }
+
+   public AcknowledgementRequest(boolean ack)
+   {
+      this(ack ? ACK : NACK);
+   }
+   
+   public AcknowledgementRequest(byte ack)
+   {
+      this.ack = ack;
+   }
+   
    // Public --------------------------------------------------------
    
+   public boolean isAck()
+   {
+      return ack == ACK;
+   }
+   
+   public boolean isExpiredAck()
+   {
+      return ack == EACK;
+   }
+   
    // Object overrides ----------------------------------------------
    
    public boolean equals(Object o)
    {
 
       if (!(o instanceof AcknowledgementRequest))
       {
          return false;
@@ -61,32 +95,33 @@
 
    public int hashCode()
    {
       return messageID.hashCode();
    }
 
    public String toString()
    {
-      return "AcknowledgementRequest:" + (isAck ? "ACK" : "NACK") + "," + destination + "," + messageID;
+      String s = (ack == ACK) ? "ACK" : (ack == EACK) ? "EACK" : "NACK";
+      return "AcknowledgementRequest:" + s + "," + destination + "," + messageID;
    }
 
    public void readExternal(java.io.ObjectInput in) throws java.io.IOException
    {
-      isAck = in.readBoolean();
+      ack = in.readByte();
       destination = SpyDestination.readDest(in);
       messageID = in.readUTF();
       subscriberId = in.readInt();
    }
    
    // Externalizable implementation ---------------------------------
    
    public void writeExternal(java.io.ObjectOutput out) throws java.io.IOException
    {
-      out.writeBoolean(isAck);
+      out.writeByte(ack);
       SpyDestination.writeDest(out, destination);
       out.writeUTF(messageID);
       out.writeInt(subscriberId);
    }
    
    // Package protected ---------------------------------------------
    
    // Protected -----------------------------------------------------
Index: messaging/src/main/org/jboss/mq/Connection.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/Connection.java,v
retrieving revision 1.44.2.4
diff -b -B -U8 -r1.44.2.4 Connection.java
--- messaging/src/main/org/jboss/mq/Connection.java	14 Apr 2005 00:13:40 -0000	1.44.2.4
+++ messaging/src/main/org/jboss/mq/Connection.java	26 Aug 2005 23:18:41 -0000
@@ -21,16 +21,17 @@
 import javax.jms.Queue;
 import javax.jms.TemporaryQueue;
 import javax.jms.TemporaryTopic;
 
 import org.jboss.logging.Logger;
 import org.jboss.mq.il.ClientILService;
 import org.jboss.mq.il.ServerIL;
 import org.jboss.util.UnreachableStatementException;
+import org.jboss.util.threadpool.BasicThreadPool;
 
 import EDU.oswego.cs.dl.util.concurrent.ClockDaemon;
 import EDU.oswego.cs.dl.util.concurrent.Semaphore;
 import EDU.oswego.cs.dl.util.concurrent.ThreadFactory;
 
 /**
  * This class implements javax.jms.Connection.
  * 
@@ -85,16 +86,17 @@
 {
    // Constants -----------------------------------------------------
 
    /** The serialVersionUID */
    private static final long serialVersionUID = 87938199839407082L;
 
    /** The threadGroup */
    private static ThreadGroup threadGroup = new ThreadGroup("JBossMQ Client Threads");
+   private static BasicThreadPool threadPool = new BasicThreadPool("Client", threadGroup);
 
    /** The log */
    static Logger log = Logger.getLogger(Connection.class);
 
    /** Whether trace is enabled */
    static boolean trace = log.isTraceEnabled();
 
    /**
@@ -219,19 +221,22 @@
             t.setDaemon(true);
             return t;
          }
       });
    }
 
    public static ThreadGroup getThreadGroup()
    {
-      if (threadGroup.isDestroyed())
-         threadGroup = new ThreadGroup("JBossMQ Client Threads");
-      return threadGroup;
+      return threadPool.getThreadGroup();
+   }
+   
+   public static BasicThreadPool getThreadPool()
+   {
+      return threadPool;
    }
    
    // Constructors --------------------------------------------------
 
    /**
 	 * Create a new Connection
 	 * 
 	 * @param userName the username
@@ -390,33 +395,34 @@
 
       if (trace)
          log.trace("Async deliver requests=" + Arrays.asList(requests) + " " + this);
       
       try
       {
          for (int i = 0; i < requests.length; i++)
          {
+            ReceiveRequest r = requests[i];
             if (trace)
-               log.trace("Processing request=" + requests[i] + " " + this);
+               log.trace("Processing request=" + r + " " + this);
             
-            SpyConsumer consumer = (SpyConsumer) subscriptions.get(requests[i].subscriptionId);
-            requests[i].message.createAcknowledgementRequest(requests[i].subscriptionId.intValue());
+            SpyConsumer consumer = (SpyConsumer) subscriptions.get(r.subscriptionId);
+            r.message.createAcknowledgementRequest(r.subscriptionId.intValue());
 
             if (consumer == null)
             {
-               send(requests[i].message.getAcknowledgementRequest(false));
-               log.debug("WARNING: NACK issued due to non existent subscription " + requests[i].message.header.messageId);
+               send(r.message.getAcknowledgementRequest(AcknowledgementRequest.NACK));
+               log.debug("WARNING: NACK issued due to non existent subscription " + r.message.header.messageId);
                continue;
             }
 
             if (trace)
-               log.trace("Delivering messageid=" + requests[i].message.header.messageId + " to consumer=" + consumer);
+               log.trace("Delivering messageid=" + r.message.header.messageId + " to consumer=" + consumer);
             
-            consumer.addMessage(requests[i].message);
+            consumer.addMessage(r.message);
          }
       }
       catch (Throwable t)
       {
          asynchFailure("Error during async delivery", t);
       }
    }
    /**
Index: messaging/src/main/org/jboss/mq/SpyMessage.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/SpyMessage.java,v
retrieving revision 1.27.6.2
diff -b -B -U8 -r1.27.6.2 SpyMessage.java
--- messaging/src/main/org/jboss/mq/SpyMessage.java	19 Oct 2004 01:11:52 -0000	1.27.6.2
+++ messaging/src/main/org/jboss/mq/SpyMessage.java	26 Aug 2005 23:18:42 -0000
@@ -64,16 +64,31 @@
 
    /**
 	 * JBoss-vendor specific property specifying the limit of redelivery
 	 * attempts of a message. The message will be redelivered a given number of
 	 * times. If not set, the container default is used.
 	 */
    public static final String PROPERTY_REDELIVERY_LIMIT = "JMS_JBOSS_REDELIVERY_LIMIT";
 
+   /** 
+   * JBoss-vendor property name holding original destination.
+   */
+   public static final String PROPERTY_ORIG_DESTINATION = "JBOSS_ORIG_DESTINATION";
+
+   /** 
+   * JBoss-vendor property name holding original expiration value.
+   */
+   public static final String PROPERTY_ORIG_EXPIRATION = "JBOSS_ORIG_EXPIRATION";
+
+   /** 
+   * JBoss-vendor property name holding original message ID value.
+   */
+   public static final String PROPERTY_ORIG_MESSAGEID = "JBOSS_ORIG_MESSAGEID";
+
    /** An object message */
    protected static final byte OBJECT_MESS = 1;
    /** An bytes message */
    protected static final byte BYTES_MESS = 2;
    /** A map message */
    protected static final byte MAP_MESS = 3;
    /** A text message */
    protected static final byte TEXT_MESS = 4;
@@ -295,48 +310,42 @@
                original.header.durableSubscriberID.subscriptionName, original.header.durableSubscriberID.selector);
    }
 
    /**
 	 * Test whether a message has expired
 	 * 
 	 * @return true when expired false otherwise
 	 */
-   public boolean isOutdated()
+   public boolean isExpired()
    {
       if (header.jmsExpiration == 0)
          return false;
       long ts = System.currentTimeMillis();
       return header.jmsExpiration < ts;
    }
 
    /**
     * Actually acknowledge a message
     *
     * @throws JMSException for any error
     */
    public void doAcknowledge() throws JMSException
    {
-      if (session.closed)
-         throw new IllegalStateException("Session is closed.");
       session.doAcknowledge(this, getAcknowledgementRequest(true));
    }
 
    /**
-    * Actually nack a message
+    * Nack an expired message
     *
     * @throws JMSException for any error
     */
-   public void doNegAcknowledge() throws JMSException
+   public void doExpiredAcknowledge() throws JMSException
    {
-      if (session.closed)
-      {
-         throw new IllegalStateException("Session is closed.");
-      }
-      session.doAcknowledge(this, getAcknowledgementRequest(false));
+      session.doAcknowledge(this, getAcknowledgementRequest(AcknowledgementRequest.EACK));
    }
 
    /**
     * Create an acknowledgement request for the message
     */
    public void createAcknowledgementRequest(int subscriptionId)
    {
       ack = new AcknowledgementRequest();
@@ -348,22 +357,33 @@
    /**
     * Get an acknowledgement request for the message
     * 
     * @param isAck true for an ack, false for a nack
     * @throws JMSException for any error
     */
    public AcknowledgementRequest getAcknowledgementRequest(boolean isAck) throws JMSException
    {
+      byte ack = isAck ? AcknowledgementRequest.ACK : AcknowledgementRequest.NACK;
+      return getAcknowledgementRequest(ack);
+   }
+
+   /**
+    * Get an acknowledgement request for the message
+    * 
+    * @param ack AcknowledgementRequest state
+    * @throws JMSException for any error
+    */
+   public AcknowledgementRequest getAcknowledgementRequest(byte ackState) throws JMSException
+   {
       //don't know if we have to copy but to be on safe side...
-      AcknowledgementRequest item = new AcknowledgementRequest();
+      AcknowledgementRequest item = new AcknowledgementRequest(ackState);
       item.destination = ack.destination;
       item.messageID = ack.messageID;
       item.subscriberId = ack.subscriberId;
-      item.isAck = isAck;
       return item;
    }
 
    // Comparable implementation -------------------------------------
 
    public int compareTo(Object o)
    {
       SpyMessage sm = (SpyMessage) o;
@@ -930,19 +950,16 @@
    void checkProperty(String name, Object value) throws JMSException
    {
       if (name == null)
          throw new IllegalArgumentException("The name of a property must not be null.");
 
       if (name.equals(""))
          throw new IllegalArgumentException("The name of a property must not be an empty String.");
 
-      if (Strings.isValidJavaIdentifier(name) == false)
-         throw new IllegalArgumentException("The property name '" + name + "' is not a valid java identifier.");
-
       if (reservedIdentifiers.contains(name))
          throw new IllegalArgumentException("The property name '" + name + "' is reserved due to selector syntax.");
 
       if (name.regionMatches(false, 0, "JMS_", 0, 4))
       {
          if (name.equals(PROPERTY_SCHEDULED_DELIVERY))
          {
             if (!(value instanceof Long))
@@ -958,30 +975,47 @@
             if (!(value instanceof Number))
                throw new JMSException(name + " must be Number: " + value);
          }
          else if (name.equals(PROPERTY_REDELIVERY_LIMIT))
          {
             if (!(value instanceof Number))
                throw new JMSException(name + " must be Number: " + value);
          }
+         else if (name.equals(PROPERTY_ORIG_EXPIRATION))
+         {
+            if (!(value instanceof Long))
+               throw new JMSException(name + " must be Long: " + value);
+         }
+         else if (name.equals(PROPERTY_ORIG_DESTINATION))
+         {
+            // no validation
+         }
+         else if (name.equals(PROPERTY_ORIG_MESSAGEID))
+         {
+            // no validation
+         }
          else
          {
             throw new JMSException("Illegal property name: " + name);
          }
       }
 
       if (name.regionMatches(false, 0, "JMSX", 0, 4))
       {
          if (name.equals("JMSXGroupID"))
             return;
          if (name.equals("JMSXGroupSeq"))
             return;
          throw new JMSException("Illegal property name: " + name);
       }
+
+      if (Strings.isValidJavaIdentifier(name) == false)
+         throw new IllegalArgumentException("The property name '" + name + "' is not a valid java identifier.");
+
    }
 
    /**
 	 * Clear a message
 	 * 
 	 * @throws JMSException for any error
 	 */
    void clearMessage() throws JMSException
Index: messaging/src/main/org/jboss/mq/SpyMessageConsumer.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/SpyMessageConsumer.java,v
retrieving revision 1.33.2.3
diff -b -B -U8 -r1.33.2.3 SpyMessageConsumer.java
--- messaging/src/main/org/jboss/mq/SpyMessageConsumer.java	18 Feb 2005 20:39:17 -0000	1.33.2.3
+++ messaging/src/main/org/jboss/mq/SpyMessageConsumer.java	26 Aug 2005 23:18:42 -0000
@@ -623,21 +623,21 @@
                   }
                   else
                   {
                      if (trace)
                         log.trace("run() message from server mes=" + mes.getJMSMessageID() + " " + this); 
                   }
                }
                mes.session = session;
-               if (mes.isOutdated())
+               if (mes.isExpired())
                {
                   if (trace)
                      log.trace("run() acking expired message mes=" + mes.getJMSMessageID() + " " + this); 
-                  mes.doAcknowledge();
+                  mes.doExpiredAcknowledge();
                   mes = null;
                }
             }
 
             MessageListener thisListener;
             synchronized (stateLock)
             {
                if (!isListening())
@@ -801,21 +801,21 @@
    }
 
    Message preProcessMessage(SpyMessage message) throws JMSException
    {
       message.session = session;
       session.addUnacknowlegedMessage(message);
 
       // Has the message expired?
-      if (message.isOutdated())
+      if (message.isExpired())
       {
          if (trace)
             log.trace("preprocess() acking expired message=" + message.getJMSMessageID() + " " + this);
-         message.doAcknowledge();
+         message.doExpiredAcknowledge();
          return null;
       }
 
       // Should we try to ack before the message is processed?
       if (!isListening())
       {
          if (session.transacted)
          {
@@ -854,21 +854,21 @@
       {
          return listening;
       }
    }
 
    protected void sessionConsumerProcessMessage(SpyMessage message) throws JMSException
    {
       message.session = session;
-      if (message.isOutdated())
+      if (message.isExpired())
       {
          if (trace)
-            log.trace("consumer() acking message in tx message=" + message.getJMSMessageID() + " " + this);
-         message.doAcknowledge();
+            log.trace("consumer() expiring message in tx message=" + message.getJMSMessageID() + " " + this);
+         message.doExpiredAcknowledge();
          return;
       }
       //simply pass on to messageListener (if there is one)
       MessageListener thisListener;
       synchronized (stateLock)
       {
          thisListener = messageListener;
       }
Index: messaging/src/main/org/jboss/mq/SpySession.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/SpySession.java,v
retrieving revision 1.32.2.5
diff -b -B -U8 -r1.32.2.5 SpySession.java
--- messaging/src/main/org/jboss/mq/SpySession.java	14 Apr 2005 00:13:40 -0000	1.32.2.5
+++ messaging/src/main/org/jboss/mq/SpySession.java	26 Aug 2005 23:18:42 -0000
@@ -158,18 +158,20 @@
 	 * API documentation to define Message.acknowledge in the same manner.
 	 * 
 	 * @param message the message to acknowledge
 	 * @param ack the acknowledgement request
 	 * @throws JMSException for any error
 	 */
    public void doAcknowledge(Message message, AcknowledgementRequest ack) throws JMSException
    {
+      if (closed)
+         throw new IllegalStateException("The session is closed");
       //if we are acking, ack all messages consumed by this session
-      if (ack.isAck)
+      if (ack.isAck())
       {
          synchronized (unacknowledgedMessages)
          {
             if (trace)
                log.trace("Acknowledging message " + ack);
 
             //ack the current message
             connection.send(((SpyMessage) message).getAcknowledgementRequest(true));
Index: messaging/src/main/org/jboss/mq/SpyXAResourceManager.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/SpyXAResourceManager.java,v
retrieving revision 1.15.6.1
diff -b -B -U8 -r1.15.6.1 SpyXAResourceManager.java
--- messaging/src/main/org/jboss/mq/SpyXAResourceManager.java	13 Apr 2005 18:43:21 -0000	1.15.6.1
+++ messaging/src/main/org/jboss/mq/SpyXAResourceManager.java	26 Aug 2005 23:18:42 -0000
@@ -235,17 +235,17 @@
          if (state.ackedMessages.size() != 0)
          {
             AcknowledgementRequest job[] = new AcknowledgementRequest[state.ackedMessages.size()];
             job = (AcknowledgementRequest[]) state.ackedMessages.toArray(job);
             transaction.acks = job;
             //Neg Acknowlege all consumed messages
             for (int i = 0; i < transaction.acks.length; i++)
               {
-               transaction.acks[i].isAck = false;
+               transaction.acks[i].ack = AcknowledgementRequest.NACK;
             }
          }
          connection.send(transaction);
       }
       else
       {
          TransactionRequest transaction = new TransactionRequest();
          transaction.xid = xid;
Index: messaging/src/main/org/jboss/mq/il/Invoker.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/il/Invoker.java,v
retrieving revision 1.3
diff -b -B -U8 -r1.3 Invoker.java
--- messaging/src/main/org/jboss/mq/il/Invoker.java	27 Aug 2003 04:31:29 -0000	1.3
+++ messaging/src/main/org/jboss/mq/il/Invoker.java	26 Aug 2005 23:18:43 -0000
@@ -7,16 +7,17 @@
 package org.jboss.mq.il;
 import javax.jms.Destination;
 
 import javax.jms.JMSException;
 import javax.jms.Queue;
 import javax.jms.TemporaryQueue;
 import javax.jms.TemporaryTopic;
 import javax.jms.Topic;
+import org.jboss.util.threadpool.ThreadPool;
 import org.jboss.mq.AcknowledgementRequest;
 import org.jboss.mq.ConnectionToken;
 import org.jboss.mq.DurableSubscriptionID;
 import org.jboss.mq.SpyTopic;
 import org.jboss.mq.Subscription;
 
 import org.jboss.mq.SpyDestination;
 import org.jboss.mq.SpyMessage;
@@ -30,19 +31,25 @@
  * @author     <a href="pra@tim.se">Peter Antman</a>
  * @version   $Revision: 1.3 $
  * @created   August 16, 2001
  */
 public interface Invoker
 {
 
    /**
+    * Get the thread pool of the server.
+    */ 
+   public ThreadPool getThreadPool();
+
+   /**
     * Get the thread group of the server.
     */ 
    public ThreadGroup getThreadGroup();
+
    /**
     * Gets a clientID from server.
     *
     * @return               The ID value
     * @exception JMSException  Description of Exception
     */
    public String getID()
           throws JMSException;
Index: messaging/src/main/org/jboss/mq/il/ServerILJMXService.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/il/ServerILJMXService.java,v
retrieving revision 1.18
diff -b -B -U8 -r1.18 ServerILJMXService.java
--- messaging/src/main/org/jboss/mq/il/ServerILJMXService.java	5 Mar 2004 12:35:22 -0000	1.18
+++ messaging/src/main/org/jboss/mq/il/ServerILJMXService.java	26 Aug 2005 23:18:43 -0000
@@ -201,17 +201,17 @@
       ctx.unbind(xaConnectionFactoryJNDIRef);
    }
 
    /**
     * @return Description of the Returned Value
     * @exception Exception Description of Exception
     * @throws javax.naming.NamingException if the server is not found
     */
-   public Invoker lookupJMSServer()
+   public Invoker getJMSServer()
    {
       return jmsServer;
    }
 
    /**
     * @return long the period of time in ms to wait between connection pings
     *         factory should be bound to
     * 
Index: messaging/src/main/org/jboss/mq/il/jvm/JVMServerILService.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/il/jvm/JVMServerILService.java,v
retrieving revision 1.14
diff -b -B -U8 -r1.14 JVMServerILService.java
--- messaging/src/main/org/jboss/mq/il/jvm/JVMServerILService.java	29 Aug 2003 02:04:37 -0000	1.14
+++ messaging/src/main/org/jboss/mq/il/jvm/JVMServerILService.java	26 Aug 2005 23:18:43 -0000
@@ -38,17 +38,17 @@
    /**
     *  Used to construct the GenericConnectionFactory (bindJNDIReferences()
     *  builds it)
     *
     * @return     The ServerIL value
     * @returns    ServerIL the instance of this IL
     */
    public ServerIL getServerIL() {
-      return new JVMServerIL( lookupJMSServer() );
+      return new JVMServerIL( getJMSServer() );
       //return null;
    }
 
    /**
     *  Used to construct the GenericConnectionFactory (bindJNDIReferences()
     *  builds it) Sets up the connection properties need by a client to use this
     *  IL
     *
Index: messaging/src/main/org/jboss/mq/il/oil/OILServerILService.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/il/oil/OILServerILService.java,v
retrieving revision 1.30
diff -b -B -U8 -r1.30 OILServerILService.java
--- messaging/src/main/org/jboss/mq/il/oil/OILServerILService.java	22 Apr 2004 15:21:26 -0000	1.30
+++ messaging/src/main/org/jboss/mq/il/oil/OILServerILService.java	26 Aug 2005 23:18:43 -0000
@@ -538,17 +538,17 @@
     *
     * @exception Exception  Description of Exception
     */
    public void startService() throws Exception
    {
       super.startService();
 
       running = true;
-      this.server = lookupJMSServer();
+      this.server = getJMSServer();
 
       // Use the default javax.net.ServerSocketFactory if none was set
       if (serverSocketFactory == null)
          serverSocketFactory = ServerSocketFactory.getDefault();
 
       /* See if the server socket supports setSecurityDomain(SecurityDomain)
       if an securityDomain was specified
       */
Index: messaging/src/main/org/jboss/mq/il/oil2/OIL2ServerILService.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/il/oil2/OIL2ServerILService.java,v
retrieving revision 1.13
diff -b -B -U8 -r1.13 OIL2ServerILService.java
--- messaging/src/main/org/jboss/mq/il/oil2/OIL2ServerILService.java	22 Apr 2004 15:21:25 -0000	1.13
+++ messaging/src/main/org/jboss/mq/il/oil2/OIL2ServerILService.java	26 Aug 2005 23:18:43 -0000
@@ -452,17 +452,17 @@
     *
     * @exception Exception  Description of Exception
     */
    public void startService() throws Exception
    {
       super.startService();
 
       running = true;
-      this.server = lookupJMSServer();
+      this.server = getJMSServer();
 
       // Use the default javax.net.ServerSocketFactory if none was set
       if (serverSocketFactory == null)
          serverSocketFactory = ServerSocketFactory.getDefault();
 
       /* See if the server socket supports setSecurityDomain(SecurityDomain)
       if an securityDomain was specified
       */
Index: messaging/src/main/org/jboss/mq/il/rmi/RMIServerILService.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/il/rmi/RMIServerILService.java,v
retrieving revision 1.16
diff -b -B -U8 -r1.16 RMIServerILService.java
--- messaging/src/main/org/jboss/mq/il/rmi/RMIServerILService.java	29 Aug 2003 02:04:40 -0000	1.16
+++ messaging/src/main/org/jboss/mq/il/rmi/RMIServerILService.java	26 Aug 2005 23:18:43 -0000
@@ -63,17 +63,17 @@
    /**
     *  Starts this IL, and binds it to JNDI
     *
     * @exception  Exception  Description of Exception
     */
    public void startService() throws Exception
    {
       super.startService();
-      serverIL = new RMIServerIL(lookupJMSServer());
+      serverIL = new RMIServerIL(getJMSServer());
       bindJNDIReferences();
 
    }
 
    /**
     *  Stops this IL, and unbinds it from JNDI
     */
    public void stopService() throws Exception
Index: messaging/src/main/org/jboss/mq/il/uil2/SocketManager.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/il/uil2/SocketManager.java,v
retrieving revision 1.14.2.4
diff -b -B -U8 -r1.14.2.4 SocketManager.java
--- messaging/src/main/org/jboss/mq/il/uil2/SocketManager.java	9 Mar 2005 21:03:14 -0000	1.14.2.4
+++ messaging/src/main/org/jboss/mq/il/uil2/SocketManager.java	26 Aug 2005 23:18:43 -0000
@@ -14,56 +14,58 @@
 import java.util.Iterator;
 
 import javax.jms.JMSException;
 
 import org.jboss.logging.Logger;
 import org.jboss.mq.il.uil2.msgs.BaseMsg;
 import org.jboss.util.stream.NotifyingBufferedInputStream;
 import org.jboss.util.stream.NotifyingBufferedOutputStream;
+import org.jboss.util.threadpool.ThreadPool;
 
 import EDU.oswego.cs.dl.util.concurrent.ConcurrentHashMap;
 import EDU.oswego.cs.dl.util.concurrent.LinkedQueue;
 import EDU.oswego.cs.dl.util.concurrent.PooledExecutor;
 import EDU.oswego.cs.dl.util.concurrent.SynchronizedBoolean;
 import EDU.oswego.cs.dl.util.concurrent.ThreadFactory;
+import EDU.oswego.cs.dl.util.concurrent.SynchronizedInt;
 
 /** Used to manage the client/server and server/client communication in an
  * asynchrounous manner.
  *
  * @todo verify the pooled executor config
  *
  * @author  Scott.Stark@jboss.org
  * @version $Revision: 1.14.2.4 $
  */
 public class SocketManager
 {
    private static Logger log = Logger.getLogger(SocketManager.class);
 
    private static final int STOPPED = 0;
    private static final int STARTED = 1;
    private static final int STOPPING = 2;
-   private static int taskID = 0;
+   private static SynchronizedInt taskID = new SynchronizedInt(0);
 
    /** The socket created by the IL layer */
    private Socket socket;
    /** The input stream used by the read task */
    private ObjectInputStream in;
    /** The buffering for output */
    NotifyingBufferedInputStream bufferedInput;
    /** The output stream used by the write task */
    private ObjectOutputStream out;
    /** The buffering for output */
    NotifyingBufferedOutputStream bufferedOutput;
    /** The write task thread */
    private Thread writeThread;
    /** The read task thread */
    private Thread readThread;
    /** The thread pool used to service incoming requests */
-   PooledExecutor pool;
+   private ThreadPool pool;
    /** The flag used to control the read loop */
    private int readState = STOPPED;
    /** The flag used to control the write loop */
    private int writeState = STOPPED;
    /** Used for constrolling the state */
    private SynchronizedBoolean running = new SynchronizedBoolean(false);
    /** The queue of messages to be processed by the write task */
    private LinkedQueue sendQueue;
@@ -85,45 +87,34 @@
       replyMap = new ConcurrentHashMap();
       trace = log.isTraceEnabled();
    }
 
    /** Start the read and write threads using the given thread group and
     * names of "UIL2.SocketManager.ReadTask" and "UIL2.SocketManager.WriteTask".
     * @param tg the thread group to use for the read and write threads.
     */
-   public void start(ThreadGroup tg)
+   public void start(ThreadPool pool, ThreadGroup tg)
    {
       if (trace)
          log.trace("start called", new Exception("Start stack trace"));
 
+      this.pool = pool;
+
       InetAddress inetAddr = socket.getInetAddress();
       String ipAddress = (inetAddr != null) ? inetAddr.getHostAddress() : "<unknown>";
       ipAddress += ":" + socket.getPort();
-      if (pool == null)
-      {
-         // TODO: Check the validity of this config
-         pool = new PooledExecutor(5);
-         pool.setMinimumPoolSize(1);
-         pool.setKeepAliveTime(1000 * 60);
-         pool.runWhenBlocked();
-         String id = "SocketManager.MsgPool@"+
-            Integer.toHexString(System.identityHashCode(this))
-            + " client=" +  ipAddress;
-         pool.setThreadFactory(new UILThreadFactory(id));
-      }
 
-      taskID++;
+      int id = taskID.increment();
       ReadTask readTask = new ReadTask();
-      readThread = new Thread(tg, readTask, "UIL2.SocketManager.ReadTask#" + taskID + " client=" +  ipAddress);
+      readThread = new Thread(tg, readTask, "UIL2.SocketManager.ReadTask#" + id + " client=" +  ipAddress);
       readThread.setDaemon(true);
 
-      taskID++;
       WriteTask writeTask = new WriteTask();
-      writeThread = new Thread(tg, writeTask, "UIL2.SocketManager.WriteTask#" + taskID + " client=" + ipAddress);
+      writeThread = new Thread(tg, writeTask, "UIL2.SocketManager.WriteTask#" + id + " client=" + ipAddress);
       writeThread.setDaemon(true);
       
       synchronized (running)
       {
          readState = STARTED;
          writeState = STARTED;
          running.set(true);
       }
@@ -144,21 +135,16 @@
             readThread.interrupt();
          }
          if (writeState == STARTED)
          {
             writeState = STOPPING;
             writeThread.interrupt();
          }
          running.set(false);
-         if (pool != null)
-         {
-            pool.shutdownNow();
-            pool = null;
-         }
       }
    }
 
    /** Set the callback handler for msgs that were not originated by the
     * socket manager. This is any msgs read that was not sent via the
     * sendMessage method.
     *
     * @param handler
@@ -298,20 +284,18 @@
                {
                   msg = BaseMsg.createMsg(msgType);
                   msg.setMsgID(msgID);
                   msg.read(in);
                   if (trace)
                      log.trace("Read new msg: " + msg);
 
                   // Handle the message
-                  if (pool == null)
-                     break;
                   msg.setHandler(this);
-                  pool.execute(msg);
+                  pool.run(msg);
                }
                else
                {
                   if (trace)
                      log.trace("Found replyMap msg: " + msg);
                   msg.setMsgID(msgID);
                   try
                   {
Index: messaging/src/main/org/jboss/mq/il/uil2/UILServerIL.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/il/uil2/UILServerIL.java,v
retrieving revision 1.7.2.3
diff -b -B -U8 -r1.7.2.3 UILServerIL.java
--- messaging/src/main/org/jboss/mq/il/uil2/UILServerIL.java	15 Jul 2005 20:40:00 -0000	1.7.2.3
+++ messaging/src/main/org/jboss/mq/il/uil2/UILServerIL.java	26 Aug 2005 23:18:44 -0000
@@ -670,17 +670,17 @@
       }
 
       socket.setTcpNoDelay(enableTcpNoDelay);
       if (soTimeout != 0)
          socket.setSoTimeout(soTimeout);
       socketMgr = new SocketManager(socket);
       socketMgr.setBufferSize(bufferSize);
       socketMgr.setChunkSize(chunkSize);
-      socketMgr.start(Connection.getThreadGroup());
+      socketMgr.start(Connection.getThreadPool(), Connection.getThreadGroup());
    }
 
    /**
     * Used to close the current connection with the server
     *
     */
    protected void destroyConnection()
    {
Index: messaging/src/main/org/jboss/mq/il/uil2/UILServerILService.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/il/uil2/UILServerILService.java,v
retrieving revision 1.5.6.5
diff -b -B -U8 -r1.5.6.5 UILServerILService.java
--- messaging/src/main/org/jboss/mq/il/uil2/UILServerILService.java	15 Jul 2005 20:40:00 -0000	1.5.6.5
+++ messaging/src/main/org/jboss/mq/il/uil2/UILServerILService.java	26 Aug 2005 23:18:44 -0000
@@ -135,21 +135,22 @@
          try
          {
             socket = serverSocket.accept();
             if( trace )
                log.trace("Accepted connection: "+socket);
             socket.setSoTimeout(readTimeout);
             socket.setTcpNoDelay(enableTcpNoDelay);
             socketMgr = new SocketManager(socket);
-            ServerSocketManagerHandler handler = new ServerSocketManagerHandler(server, socketMgr);
+            ServerSocketManagerHandler handler = new ServerSocketManagerHandler(getJMSServer(), socketMgr);
             socketMgr.setHandler(handler);
             socketMgr.setBufferSize(bufferSize);
             socketMgr.setChunkSize(chunkSize);
-            socketMgr.start(server.getThreadGroup());
+            Invoker s = getJMSServer();
+            socketMgr.start(s.getThreadPool(), s.getThreadGroup());
          }
          catch (IOException e)
          {
             if (running)
                log.warn("Failed to setup client connection", e);
          }
          catch(Throwable e)
          {
@@ -163,17 +164,16 @@
     * Starts this IL, and binds it to JNDI
     *
     * @exception Exception  Description of Exception
     */
    public void startService() throws Exception
    {
       super.startService();
       running = true;
-      this.server = lookupJMSServer();
 
       // Use the default javax.net.ServerSocketFactory if none was set
       if (serverSocketFactory == null)
          serverSocketFactory = ServerSocketFactory.getDefault();
 
       /* See if the server socket supports setSecurityDomain(SecurityDomain)
       if an securityDomain was specified
       */
@@ -199,17 +199,17 @@
          }
       }
 
       // Create the server socket using the socket factory
       serverSocket = serverSocketFactory.createServerSocket(serverBindPort, 50, bindAddress);
 
       InetAddress socketAddress = serverSocket.getInetAddress();
       log.info("JBossMQ UIL service available at : " + socketAddress + ":" + serverSocket.getLocalPort());
-      acceptThread = new Thread(server.getThreadGroup(), this, "UILServerILService Accept Thread");
+      acceptThread = new Thread(getJMSServer().getThreadGroup(), this, "UILServerILService Accept Thread");
       acceptThread.start();
 
       /* We need to check the socketAddress against "0.0.0.0/0.0.0.0"
          because this is not a valid address on Win32 while it is for
          *NIX. See BugParade bug #4343286.
       */
       socketAddress = ServerConfigUtil.fixRemoteAddress(socketAddress);
       // If an explicit client bind address has been specified use it
Index: messaging/src/main/org/jboss/mq/server/BasicQueue.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/BasicQueue.java,v
retrieving revision 1.42.2.4
diff -b -B -U8 -r1.42.2.4 BasicQueue.java
--- messaging/src/main/org/jboss/mq/server/BasicQueue.java	27 Jul 2005 21:40:15 -0000	1.42.2.4
+++ messaging/src/main/org/jboss/mq/server/BasicQueue.java	26 Aug 2005 23:18:44 -0000
@@ -18,22 +18,23 @@
 
 import javax.jms.IllegalStateException;
 import javax.jms.JMSException;
 
 import org.jboss.logging.Logger;
 import org.jboss.mq.AcknowledgementRequest;
 import org.jboss.mq.DestinationFullException;
 import org.jboss.mq.SpyJMSException;
+import org.jboss.mq.SpyDestination;
 import org.jboss.mq.SpyMessage;
 import org.jboss.mq.Subscription;
 import org.jboss.mq.pm.Tx;
+import org.jboss.mq.pm.TxManager;
 import org.jboss.mq.selectors.Selector;
 import org.jboss.util.timeout.Timeout;
-import org.jboss.util.timeout.TimeoutFactory;
 import org.jboss.util.timeout.TimeoutTarget;
 
 import EDU.oswego.cs.dl.util.concurrent.ConcurrentHashMap;
 import EDU.oswego.cs.dl.util.concurrent.SynchronizedInt;
 
 /**
  *  This class represents a queue which provides it's messages exclusively to one
  *  consumer at a time.<p>
@@ -207,19 +208,17 @@
             for (Iterator it = messages.iterator(); it.hasNext();)
             {
                MessageReference message = (MessageReference) it.next();
                try
                {
                   if (message.isExpired())
                   {
                      it.remove();
-                     if (trace)
-                        log.trace("message expired: " + message);
-                     dropMessage(message);
+                     expireMessageAsync(message);
                   }
                   else if (sub.accepts(message.getHeaders()))
                   {
                      //queue message for sending to this sub
                      it.remove();
                      found = message;
                      break;
                   }
@@ -465,19 +464,17 @@
                // find a non-expired message
                while (messages.size() != 0)
                {
                   messageRef = (MessageReference) messages.first();
                   messages.remove(messageRef);
 
                   if (messageRef.isExpired())
                   {
-                     if (trace)
-                        log.trace("message expired: " + messageRef);
-                     dropMessage(messageRef);
+                     expireMessageAsync(messageRef);
                      messageRef = null;
                   }
                   else
                      break;
                }
             }
          }
          else
@@ -487,19 +484,17 @@
             {
                Iterator i = messages.iterator();
                while (i.hasNext())
                {
                   MessageReference mr = (MessageReference) i.next();
                   if (mr.isExpired())
                   {
                      i.remove();
-                     if (trace)
-                        log.trace("message expired: " + mr);
-                     dropMessage(mr);
+                     expireMessageAsync(mr);
                   }
                   else if (sub.accepts(mr.getHeaders()))
                   {
                      messageRef = mr;
                      i.remove();
                      break;
                   }
                }
@@ -547,21 +542,24 @@
             map.remove(unacked.messageRef);
          if (map == null || map.isEmpty())
            unackedBySubscription.remove(unacked.sub);
       }
 
       MessageReference m = unacked.messageRef;
 
       // Was it a negative acknowledge??
-      if (!item.isAck)
+      if (!item.isAck())
       {
-         Runnable task = new RestoreMessageTask(m);
+         Runnable task;
+         if (item.isExpiredAck())
+           task = new ExpireMessageTask(m);
+         else
+           task = new RestoreMessageTask(m);
          server.getPersistenceManager().getTxManager().addPostCommitTask(txId, task);
-
       }
       else
       {
          if (m.isPersistent())
             server.getPersistenceManager().remove(m, txId);
 
          Runnable task = new RestoreMessageTask(m);
          server.getPersistenceManager().getTxManager().addPostRollbackTask(txId, task);
@@ -815,47 +813,49 @@
          log.trace("removeReceiver " + " " + sub + " " + this);
 
       synchronized (receivers)
       {
          receivers.remove(sub);
       }
    }
 
+   private void addTimeout(MessageReference message, TimeoutTarget t, long ts)
+   {
+      Timeout timeout = server.getTimeoutFactory().schedule(ts, t);
+      events.put(message, timeout);
+   }
+
    /**
     * Add a message
     *
     * @param message the message to add
     */
    private void internalAddMessage(MessageReference message)
    {
       boolean trace = log.isTraceEnabled();
       if (trace)
          log.trace("internalAddMessage " + " " + message + " " + this);
 
       // If scheduled, put in timer queue
       long ts = message.messageScheduledDelivery;
       if (ts > 0 && ts > System.currentTimeMillis())
       {
-         EnqueueMessageTask t = new EnqueueMessageTask(message);
-         Timeout timeout = TimeoutFactory.createTimeout(ts, t);
-         events.put(message, timeout);
+         addTimeout(message, new EnqueueMessageTask(message), ts);
          scheduledMessageCount.increment();
          if (trace)
             log.trace("scheduled message at " + new Date(ts) + ": " + message);
          // Can't deliver now
          return;
       }
 
       // Don't bother with expired messages
       if (message.isExpired())
       {
-         if (trace)
-            log.trace("message expired: " + message);
-         dropMessage(message);
+         expireMessageAsync(message);
          return;
       }
 
       try
       {
          Subscription found = null;
          synchronized (receivers)
          {
@@ -878,20 +878,17 @@
                synchronized (messages)
                {
                   messages.add(message);
 
                   // If a message is set to expire, and nobody wants it, put its reaper in
                   // the timer queue 
                   if (message.messageExpiration > 0)
                   {
-                     // This could be a waste of memory for large numbers of expirations
-                     ExpireMessageTask t = new ExpireMessageTask(message);
-                     Timeout timeout = TimeoutFactory.createTimeout(message.messageExpiration, t);
-                     events.put(message, timeout);
+                     addTimeout(message, new ExpireMessageTask(message), message.messageExpiration);
                   }
                }
             }
          }
          
          // Queue to the receiver
          if (found != null)
             queueMessageForSending(found, message);
@@ -936,34 +933,33 @@
     *
     * @param sub the subscription receiving the message
     * @param messageRef the message to be acknowledged
     * @throws JMSException for any error
     */
    protected void setupMessageAcknowledgement(Subscription sub, MessageReference messageRef) throws JMSException
    {
       SpyMessage message = messageRef.getMessage();
-      AcknowledgementRequest ack = new AcknowledgementRequest();
-      ack.destination = message.getJMSDestination();
-      ack.messageID = message.getJMSMessageID();
-      ack.subscriberId = sub.subscriptionId;
-      ack.isAck = false;
+      AcknowledgementRequest nack = new AcknowledgementRequest(false);
+      nack.destination = message.getJMSDestination();
+      nack.messageID = message.getJMSMessageID();
+      nack.subscriberId = sub.subscriptionId;
 
       synchronized (messages)
       {
          UnackedMessageInfo unacked = new UnackedMessageInfo(messageRef, sub);
-         unacknowledgedMessages.put(ack, unacked);
-         unackedByMessageRef.put(messageRef, ack);
+         unacknowledgedMessages.put(nack, unacked);
+         unackedByMessageRef.put(messageRef, nack);
          HashMap map = (HashMap) unackedBySubscription.get(sub);
          if (map == null)
          {
             map = new HashMap();
             unackedBySubscription.put(sub, map);
          }
-         map.put(messageRef, ack);
+         map.put(messageRef, nack);
       }
    }
    /**
     * Remove a message
     *
     * @param message the message to remove
     */
    protected void dropMessage(MessageReference message)
@@ -1007,16 +1003,68 @@
          server.getMessageCache().remove(message);
       }
       catch (JMSException e)
       {
          log.warn("Error dropping message " + message, e);
       }
    }
 
+   /**
+    * Expire a message asynchronously.
+    *
+    * @param messageRef the message to remove
+    */
+   protected void expireMessageAsync(MessageReference messageRef)
+   {
+      server.getThreadPool().run(new ExpireMessageTask(messageRef));
+   }
+   
+   /**
+    * Expire a message
+    *
+    * @param messageRef the message to remove
+    */
+   protected void expireMessage(MessageReference messageRef)
+   {
+      boolean trace = log.isTraceEnabled();
+      if (trace)
+         log.trace("message expired: " + messageRef);
+
+      SpyDestination ed = parameters.expiryDestination;
+      if (ed == null)
+      {
+         dropMessage(messageRef);
+         return;
+      }
+
+      if (trace)
+         log.trace("sending to: " + ed);
+
+      try
+      {
+         SpyMessage orig = messageRef.getMessage();
+         SpyMessage copy = orig.myClone();
+         copy.header.jmsPropertiesReadWrite = true;
+         copy.setJMSExpiration(0);
+         copy.setJMSDestination(ed);
+         copy.setLongProperty(SpyMessage.PROPERTY_ORIG_EXPIRATION, orig.getJMSExpiration());
+         copy.setStringProperty(SpyMessage.PROPERTY_ORIG_DESTINATION, orig.getJMSDestination().toString());
+         TxManager tm = server.getPersistenceManager().getTxManager();
+         Tx tx = tm.createTx();
+         server.addMessage(null, copy, tx);
+         dropMessage(messageRef, tx);
+         tm.commitTx(tx);
+      }
+      catch (JMSException e)
+      {
+         log.error("Could not move expired message: " + messageRef, e);
+      }
+   }
+
    // Package Private -----------------------------------------------
 
    // Private -------------------------------------------------------
 
    /**
     * Check whether a removed subscription can be permenantly removed.
     * This method is private because it assumes external synchronization
     *
@@ -1162,17 +1210,17 @@
          internalAddMessage(messageRef);
          scheduledMessageCount.decrement();
       }
    }
 
    /**
     * Drop a message when it expires
     */
-   private class ExpireMessageTask implements TimeoutTarget
+   private class ExpireMessageTask implements TimeoutTarget, Runnable
    {
       private MessageReference messageRef;
 
       public ExpireMessageTask(MessageReference messageRef)
       {
          this.messageRef = messageRef;
       }
 
@@ -1181,19 +1229,22 @@
          events.remove(messageRef);
          synchronized (messages)
          {
             // If the message was already sent, then do nothing
             // (This probably happens more than not)
             if (messages.remove(messageRef) == false)
                return;
          }
-         if (log.isTraceEnabled())
-            log.trace("message expired: " + messageRef);
-         dropMessage(messageRef);
+         expireMessage(messageRef);
+      }
+
+      public void run()
+      {
+         expireMessage(messageRef);
       }
    }
 
    /**
     * Information about unacknowledged messages
     */
    private static class UnackedMessageInfo
    {
Index: messaging/src/main/org/jboss/mq/server/BasicQueueParameters.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/BasicQueueParameters.java,v
retrieving revision 1.4.4.2
diff -b -B -U8 -r1.4.4.2 BasicQueueParameters.java
--- messaging/src/main/org/jboss/mq/server/BasicQueueParameters.java	12 Jul 2005 17:12:02 -0000	1.4.4.2
+++ messaging/src/main/org/jboss/mq/server/BasicQueueParameters.java	26 Aug 2005 23:18:44 -0000
@@ -1,16 +1,18 @@
 /*
  * JBossMQ, the OpenSource JMS implementation
  *
  * Distributable under LGPL license.
  * See terms of license at gnu.org.
  */
 package org.jboss.mq.server;
 
+import org.jboss.mq.SpyDestination;
+
 /**
  * Parameters controlling a basic queue
  *
  * @author <a href="mailto:adrian@jboss.org">Adrian Brock</a>
  * @version $Revision: 1.4.4.2 $
  */
 public class BasicQueueParameters
 {
@@ -35,9 +37,12 @@
    /** The receivers implementation class */
    public Class receiversImpl;
    
    /** The message counter history */
    public int messageCounterHistoryDayLimit = -1;
    
    /** The number of retries when recovering a queue/durable subscription */
    public int recoveryRetries = 0;
+   
+   /** The destination that receives expired messages */
+   public SpyDestination expiryDestination = null;
 }
Index: messaging/src/main/org/jboss/mq/server/ClientConsumer.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/ClientConsumer.java,v
retrieving revision 1.21.6.2
diff -b -B -U8 -r1.21.6.2 ClientConsumer.java
--- messaging/src/main/org/jboss/mq/server/ClientConsumer.java	13 Oct 2004 15:18:55 -0000	1.21.6.2
+++ messaging/src/main/org/jboss/mq/server/ClientConsumer.java	26 Aug 2005 23:18:44 -0000
@@ -9,34 +9,34 @@
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.LinkedList;
 
 import javax.jms.InvalidDestinationException;
 import javax.jms.JMSException;
 
 import org.jboss.logging.Logger;
+import org.jboss.util.threadpool.ThreadPool;
 import org.jboss.mq.AcknowledgementRequest;
 import org.jboss.mq.ConnectionToken;
 import org.jboss.mq.ReceiveRequest;
 import org.jboss.mq.SpyMessage;
 import org.jboss.mq.Subscription;
-import org.jboss.mq.threadpool.ThreadPool;
 import org.jboss.mq.threadpool.Work;
 
 /**
  *  This represent the clients queue which consumes messages from the
  *  destinations on the provider.
  *
  * @author     Hiram Chirino (Cojonudo14@hotmail.com)
  * @author <a href="mailto:pra@tim.se">Peter Antman</a>
  * @created    August 16, 2001
  * @version    $Revision: 1.21.6.2 $
  */
-public class ClientConsumer implements Work
+public class ClientConsumer implements Runnable
 {
    private static Logger log = Logger.getLogger(ClientConsumer.class);
    //The JMSServer object
    JMSDestinationManager server;
    //The connection this queue will send messages over
    ConnectionToken connectionToken;
    //Is this connection enabled (Can we transmit to the receiver)
    boolean enabled;
@@ -58,30 +58,25 @@
    private boolean enqueued = false;
 
    // Static ---------------------------------------------------
 
    /**
     *  The {@link org.jboss.mq.threadpool.ThreadPool ThreadPool} that
     *  does the actual message pushing for us.
     */
-   private static ThreadPool threadPool = null;
+   private ThreadPool threadPool = null;
 
    // Constructor ---------------------------------------------------
 
    public ClientConsumer(JMSDestinationManager server, ConnectionToken connectionToken) throws JMSException
    {
       this.server = server;
       this.connectionToken = connectionToken;
-      // Create thread pool
-      synchronized (ClientConsumer.class)
-      {
-         if (threadPool == null)
-            threadPool = new ThreadPool("Message Pushers", server.threadGroup, 10, true);
-      }
+      this.threadPool = server.getThreadPool();
    }
 
    public void setEnabled(boolean enabled) throws JMSException
    {
       if (log.isTraceEnabled())
          log.trace("" + this +"->setEnabled(enabled=" + enabled + ")");
 
       // queues might be waiting for messages.
@@ -108,17 +103,17 @@
       synchronized (messages)
       {
          if (closed)
             return; // Wouldn't be delivered anyway
 
          messages.add(r);
          if (!enqueued)
          {
-            threadPool.enqueueWork(this);
+            threadPool.run(this);
             enqueued = true;
          }
       }
    }
 
    public void addSubscription(Subscription req) throws JMSException
    {
       if (log.isTraceEnabled())
@@ -144,19 +139,16 @@
       if (trace)
          log.trace("" + this +"->close()");
 
       synchronized (messages)
       {
          closed = true;
          if (enqueued)
          {
-            if (trace)
-               log.trace("" + this +"->close(): Cancelling work in progress.");
-            threadPool.cancelWork(this);
             enqueued = false;
          }
          messages.clear();
       }
 
       // Remove all the subscriptions for this client
       HashMap subscriptionsClone = null;
       synchronized (subscriptions)
@@ -247,17 +239,17 @@
 
       queue.removeSubscriber(req);
 
    }
 
    /**
     *  Push some messages.
     */
-   public void doWork()
+   public void run()
    {
       try
       {
 
          ReceiveRequest[] job;
 
          synchronized (messages)
          {
Index: messaging/src/main/org/jboss/mq/server/JMSDestinationManager.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/JMSDestinationManager.java,v
retrieving revision 1.17.6.3
diff -b -B -U8 -r1.17.6.3 JMSDestinationManager.java
--- messaging/src/main/org/jboss/mq/server/JMSDestinationManager.java	13 Apr 2005 18:43:24 -0000	1.17.6.3
+++ messaging/src/main/org/jboss/mq/server/JMSDestinationManager.java	26 Aug 2005 23:18:44 -0000
@@ -15,16 +15,18 @@
 import javax.jms.Destination;
 import javax.jms.InvalidDestinationException;
 import javax.jms.JMSException;
 import javax.jms.Queue;
 import javax.jms.TemporaryQueue;
 import javax.jms.TemporaryTopic;
 import javax.jms.Topic;
 
+import org.jboss.util.threadpool.ThreadPool;
+import org.jboss.util.timeout.TimeoutFactory;
 import org.jboss.logging.Logger;
 import org.jboss.mq.AcknowledgementRequest;
 import org.jboss.mq.ConnectionToken;
 import org.jboss.mq.DurableSubscriptionID;
 import org.jboss.mq.SpyDestination;
 import org.jboss.mq.SpyJMSException;
 import org.jboss.mq.SpyMessage;
 import org.jboss.mq.SpyQueue;
@@ -62,18 +64,22 @@
    // Attributes
    /////////////////////////////////////////////////////////////////////
 
    /** Destinations SpyDestination -> JMSDestination */ 
    public Map destinations = new ConcurrentReaderHashMap();
    /** Destinations being closed SpyDestination -> JMSDestination */ 
    public Map closingDestinations = new ConcurrentReaderHashMap();
    //Thread group for server side threads.
+   /** Thread pool */
+   public ThreadPool threadPool;
    /** Thread group */
-   public ThreadGroup threadGroup = new ThreadGroup("JBossMQ Server Threads");
+   public ThreadGroup threadGroup;
+   /** Timeout factory */
+   public TimeoutFactory timeoutFactory;
 
    //The list of ClientConsumers hased by ConnectionTokens
    Map clientConsumers = new ConcurrentReaderHashMap();
 
    //last id given to a client
    private int lastID = 1;
    //last id given to a temporary topic
    private int lastTemporaryTopic = 1;
@@ -176,21 +182,41 @@
    /** Obtain a copy of the current clients
     * @return a HashMap<ConnectionToken, ClientConsumer> of current clients
     */
    public HashMap getClients()
    {
       return new HashMap(clientConsumers);
    }
 
+   public void setThreadPool(ThreadPool threadPool)
+   {
+      this.threadPool = threadPool;
+   }
+
+   public ThreadPool getThreadPool()
+   {
+      return threadPool;
+   }
+
+   public void setThreadGroup(ThreadGroup threadGroup)
+   {
+      this.threadGroup = threadGroup;
+   }
+
    public ThreadGroup getThreadGroup()
    {
       return threadGroup;
    }
 
+   public TimeoutFactory getTimeoutFactory()
+   {
+      return timeoutFactory;
+   }
+
    //Get a new ClientID for a connection
    /**
     * Gets the ID attribute of the JMSServer object
     *
     * @return   The ID value
     */
    public String getID()
    {
@@ -330,27 +356,29 @@
    /**
     * #Description of the Method
     */
    public void startServer()
    {
       synchronized (stateLock)
       {
          this.stopped = false;
+         this.timeoutFactory = new TimeoutFactory(this.threadPool);
       }
    }
 
    /**
     * #Description of the Method
     */
    public void stopServer()
    {
       synchronized (stateLock)
       {
          this.stopped = true;
+         this.timeoutFactory.cancel();
       }
    }
 
    /**
     * Check a clienID set by the client.
     *
     * @param ID                Description of Parameter
     * @exception JMSException  Description of Exception
Index: messaging/src/main/org/jboss/mq/server/JMSServerInterceptor.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/JMSServerInterceptor.java,v
retrieving revision 1.5
diff -b -B -U8 -r1.5 JMSServerInterceptor.java
--- messaging/src/main/org/jboss/mq/server/JMSServerInterceptor.java	5 Mar 2004 12:35:20 -0000	1.5
+++ messaging/src/main/org/jboss/mq/server/JMSServerInterceptor.java	26 Aug 2005 23:18:44 -0000
@@ -8,16 +8,17 @@
 
 import javax.jms.Destination;
 import javax.jms.JMSException;
 import javax.jms.Queue;
 import javax.jms.TemporaryQueue;
 import javax.jms.TemporaryTopic;
 import javax.jms.Topic;
 
+import org.jboss.util.threadpool.ThreadPool;
 import org.jboss.mq.AcknowledgementRequest;
 import org.jboss.mq.ConnectionToken;
 import org.jboss.mq.DurableSubscriptionID;
 import org.jboss.mq.SpyDestination;
 import org.jboss.mq.SpyMessage;
 import org.jboss.mq.SpyTopic;
 import org.jboss.mq.Subscription;
 import org.jboss.mq.TransactionRequest;
@@ -46,16 +47,22 @@
     * Get next invoker in chain to be called. Is mot often the real JMSServer
     */
    public JMSServerInterceptor getNext();
 
    /**
     * Get the thread group of the server.
     */
    public ThreadGroup getThreadGroup();
+
+   /**
+    * Get the thread pool of the server.
+    */
+   public ThreadPool getThreadPool();
+
    /**
     * Gets a clientID from server.
     *
     * @return               The ID value
     * @exception JMSException  Description of Exception
     */
    public String getID() throws JMSException;
 
Index: messaging/src/main/org/jboss/mq/server/JMSServerInterceptorSupport.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/JMSServerInterceptorSupport.java,v
retrieving revision 1.5
diff -b -B -U8 -r1.5 JMSServerInterceptorSupport.java
--- messaging/src/main/org/jboss/mq/server/JMSServerInterceptorSupport.java	5 Mar 2004 12:35:20 -0000	1.5
+++ messaging/src/main/org/jboss/mq/server/JMSServerInterceptorSupport.java	26 Aug 2005 23:18:45 -0000
@@ -9,16 +9,17 @@
 import javax.jms.Destination;
 import javax.jms.JMSException;
 import javax.jms.Queue;
 import javax.jms.TemporaryQueue;
 import javax.jms.TemporaryTopic;
 import javax.jms.Topic;
 
 import org.jboss.logging.Logger;
+import org.jboss.util.threadpool.ThreadPool;
 import org.jboss.mq.AcknowledgementRequest;
 import org.jboss.mq.ConnectionToken;
 import org.jboss.mq.DurableSubscriptionID;
 import org.jboss.mq.SpyDestination;
 import org.jboss.mq.SpyMessage;
 import org.jboss.mq.SpyTopic;
 import org.jboss.mq.Subscription;
 import org.jboss.mq.TransactionRequest;
@@ -61,16 +62,25 @@
 
    /**
     * Get the thread group of the server.
     */
    public ThreadGroup getThreadGroup()
    {
       return nextInterceptor.getThreadGroup();
    }
+
+   /**
+    * Returns the thread pool of the server.
+    */
+   public ThreadPool getThreadPool()
+   {
+      return nextInterceptor.getThreadPool();
+   }
+
    /**
     * Gets a clientID from server.
     *
     * @return               The ID value
     * @exception JMSException  Description of Exception
     */
    public String getID() throws JMSException
    {
Index: messaging/src/main/org/jboss/mq/server/JMSServerInvoker.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/JMSServerInvoker.java,v
retrieving revision 1.6
diff -b -B -U8 -r1.6 JMSServerInvoker.java
--- messaging/src/main/org/jboss/mq/server/JMSServerInvoker.java	5 Mar 2004 12:35:20 -0000	1.6
+++ messaging/src/main/org/jboss/mq/server/JMSServerInvoker.java	26 Aug 2005 23:18:45 -0000
@@ -9,16 +9,17 @@
 import javax.jms.Destination;
 import javax.jms.JMSException;
 import javax.jms.Queue;
 import javax.jms.TemporaryQueue;
 import javax.jms.TemporaryTopic;
 import javax.jms.Topic;
 
 import org.jboss.logging.Logger;
+import org.jboss.util.threadpool.ThreadPool;
 import org.jboss.mq.AcknowledgementRequest;
 import org.jboss.mq.ConnectionToken;
 import org.jboss.mq.DurableSubscriptionID;
 import org.jboss.mq.SpyDestination;
 import org.jboss.mq.SpyMessage;
 import org.jboss.mq.SpyTopic;
 import org.jboss.mq.Subscription;
 import org.jboss.mq.TransactionRequest;
@@ -62,16 +63,25 @@
 
    /**
     * Get the thread group of the server.
     */
    public ThreadGroup getThreadGroup()
    {
       return nextInterceptor.getThreadGroup();
    }
+
+   /**
+    * Get the thread pool of the server.
+    */
+   public ThreadPool getThreadPool()
+   {
+      return nextInterceptor.getThreadPool();
+   }
+
    /**
     * Gets a clientID from server.
     *
     * @return               The ID value
     * @exception JMSException  Description of Exception
     */
    public String getID() throws JMSException
    {
Index: messaging/src/main/org/jboss/mq/server/MessageCache.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/MessageCache.java,v
retrieving revision 1.31.2.1
diff -b -B -U8 -r1.31.2.1 MessageCache.java
--- messaging/src/main/org/jboss/mq/server/MessageCache.java	13 Apr 2005 17:58:26 -0000	1.31.2.1
+++ messaging/src/main/org/jboss/mq/server/MessageCache.java	26 Aug 2005 23:18:45 -0000
@@ -884,15 +884,15 @@
          return mostRecent;
       }
       public Node getLeastRecent()
       {
          return leastRecent;
       }
    }
 
-   class Node
+   static class Node
    {
       Node moreRecent = null;
       Node lessRecent = null;
       Object data = null;
    }
 }
Index: messaging/src/main/org/jboss/mq/server/MessageReference.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/MessageReference.java,v
retrieving revision 1.20.2.2
diff -b -B -U8 -r1.20.2.2 MessageReference.java
--- messaging/src/main/org/jboss/mq/server/MessageReference.java	12 Jul 2005 00:42:50 -0000	1.20.2.2
+++ messaging/src/main/org/jboss/mq/server/MessageReference.java	26 Aug 2005 23:18:45 -0000
@@ -172,17 +172,17 @@
       if (redeliveryDelay != 0)
       {
          log.trace("message has redelivery delay");
          messageScheduledDelivery = System.currentTimeMillis() + redeliveryDelay;
       }
 
       ++redeliveryCount;
       
-      if (queue.parameters.lateClone == false)
+      if (isLateClone() == false)
       {
          SpyMessage message = getMessage();
          message.setJMSRedelivered(redelivered);
 
          message.header.jmsProperties.put(SpyMessage.PROPERTY_REDELIVERY_COUNT, new Integer(redeliveryCount));
       }
    }
 
Index: messaging/src/main/org/jboss/mq/server/jmx/DestinationMBean.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/jmx/DestinationMBean.java,v
retrieving revision 1.7.4.2
diff -b -B -U8 -r1.7.4.2 DestinationMBean.java
--- messaging/src/main/org/jboss/mq/server/jmx/DestinationMBean.java	12 Jul 2005 17:12:02 -0000	1.7.4.2
+++ messaging/src/main/org/jboss/mq/server/jmx/DestinationMBean.java	26 Aug 2005 23:18:45 -0000
@@ -168,9 +168,20 @@
     * Returns the recovery retries
     */
    public int getRecoveryRetries();
 
    /**
     * Sets the class implementating the receivers
     */
    public void setRecoveryRetries(int retries);
+
+   /**
+    * Returns the expiry destination.
+    */
+   public ObjectName getExpiryDestination();
+
+   /**
+    * Sets the expiry destination.
+    */
+   public void setExpiryDestination(ObjectName destination);
+
 }
Index: messaging/src/main/org/jboss/mq/server/jmx/DestinationMBeanSupport.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/jmx/DestinationMBeanSupport.java,v
retrieving revision 1.10.4.5
diff -b -B -U8 -r1.10.4.5 DestinationMBeanSupport.java
--- messaging/src/main/org/jboss/mq/server/jmx/DestinationMBeanSupport.java	1 Aug 2005 16:42:17 -0000	1.10.4.5
+++ messaging/src/main/org/jboss/mq/server/jmx/DestinationMBeanSupport.java	26 Aug 2005 23:18:45 -0000
@@ -41,16 +41,18 @@
    String destinationName;
    String jndiName;
    boolean jndiBound;
    ObjectName jbossMQService;
    //String securityConf;
    Element securityConf;
    /** The basic parameters */
    BasicQueueParameters parameters = new BasicQueueParameters();
+   /** Destination JNDI name */
+   ObjectName expiryDestination;
 
    /**
     * A optional security manager. Must be set to use security conf.
     */
    ObjectName securityManager;
 
    /**
     * Get the value of JBossMQService.
@@ -100,27 +102,43 @@
    
    public void createService() throws Exception
    {
       // Copy default values from the server when null or zero
       if (parameters.receiversImpl == null)
          parameters.receiversImpl = (Class) server.getAttribute(jbossMQService, "ReceiversImpl");
       if (parameters.recoveryRetries == 0)
          parameters.recoveryRetries = ((Integer) server.getAttribute(jbossMQService, "RecoveryRetries")).intValue();
+      if (expiryDestination == null)
+         expiryDestination = (ObjectName) server.getAttribute(jbossMQService, "ExpiryDestination");
    }
    
    public void startService() throws Exception
    {
 
       if (securityManager != null)
       {
          // Set securityConf at manager
          getServer().invoke(securityManager, "addDestination", new Object[]{spyDest.getName(), securityConf}, new String[]{"java.lang.String", "org.w3c.dom.Element"});
       }
 
+      if (expiryDestination != null)
+      {
+         String jndi = (String)getServer().getAttribute(expiryDestination, "JNDIName");
+         InitialContext ctx = new InitialContext();
+         try
+         {
+            parameters.expiryDestination = (SpyDestination)ctx.lookup(jndi);
+         }
+         finally
+         {
+            ctx.close();
+         }
+      }
+
    }
 
    public void stopService() throws Exception
    {
       // unbind from JNDI
       if (jndiBound)
       {
          InitialContext ctx = new InitialContext();
@@ -501,9 +519,20 @@
    {
       return parameters.recoveryRetries;
    }
 
    public void setRecoveryRetries(int retries)
    {
       parameters.recoveryRetries = retries;
    }
+
+   public ObjectName getExpiryDestination() 
+   {
+      return expiryDestination;
+   }
+
+   public void setExpiryDestination(ObjectName expiryDestination)
+   {
+      this.expiryDestination = expiryDestination;
+   }
+
 }
Index: messaging/src/main/org/jboss/mq/server/jmx/DestinationManager.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/jmx/DestinationManager.java,v
retrieving revision 1.13.4.3
diff -b -B -U8 -r1.13.4.3 DestinationManager.java
--- messaging/src/main/org/jboss/mq/server/jmx/DestinationManager.java	12 Jul 2005 18:04:36 -0000	1.13.4.3
+++ messaging/src/main/org/jboss/mq/server/jmx/DestinationManager.java	26 Aug 2005 23:18:45 -0000
@@ -20,16 +20,18 @@
 import org.jboss.mq.pm.PersistenceManager;
 import org.jboss.mq.server.BasicQueueParameters;
 import org.jboss.mq.server.JMSDestinationManager;
 import org.jboss.mq.server.JMSServerInterceptor;
 import org.jboss.mq.server.MessageCache;
 import org.jboss.mq.server.MessageCounter;
 import org.jboss.mq.server.Receivers;
 import org.jboss.mq.sm.StateManager;
+import org.jboss.util.threadpool.ThreadPool;
+import org.jboss.util.threadpool.BasicThreadPool;
 import org.jboss.mx.util.MBeanProxyExt;
 import org.jboss.system.ServiceControllerMBean;
 
 /**
  * JMX MBean implementation for JBossMQ.
  *
  * @jmx:mbean extends="org.jboss.mq.server.jmx.InterceptorMBean"
  * @author     Vincent Sheffer (vsheffer@telkel.com)
@@ -52,16 +54,20 @@
    /** The JMX ObjectName of the configured persistence manager */
    private ObjectName persistenceManager;
    /** The JMX ObjectName of the configured state manager */
    private ObjectName stateManager;
    /** The JMX ObjectName of the message cache */
    private ObjectName messageCache;
    /** The temporary topic/queue parameters */
    private BasicQueueParameters tempParameters = new BasicQueueParameters();
+   /** Default thread pool */
+   private ObjectName threadPool;
+   /** Default expiry destination */
+   private ObjectName expiryDestination;
 
    /**
     * @jmx:managed-attribute
     * @return the count of active clients
     */
    public int getClientCount()
    {
       if (jmsServer == null)
@@ -214,16 +220,36 @@
       return tempParameters.recoveryRetries;
    }
 
    public void setRecoveryRetries(int retries)
    {
       tempParameters.recoveryRetries = retries;
    }
    
+   public ObjectName getThreadPool()
+   {
+      return this.threadPool;
+   }
+
+   public void setThreadPool(ObjectName threadPool)
+   {
+      this.threadPool = threadPool;
+   }
+   
+   public ObjectName getExpiryDestination() 
+   {
+      return expiryDestination;
+   }
+
+   public void setExpiryDestination(ObjectName expiryDestination)
+   {
+      this.expiryDestination = expiryDestination;
+   }
+
    /**
     * @jmx:managed-operation
     */
    public void createQueue(String name) throws Exception
    {
       createDestination("org.jboss.mq.server.jmx.Queue", getQueueObjectName(name), null);
    }
 
@@ -354,23 +380,41 @@
       PersistenceManager pm = (PersistenceManager) server.getAttribute(persistenceManager, "Instance");
       jmsServer.setPersistenceManager(pm);
 
       StateManager sm = (StateManager) server.getAttribute(stateManager, "Instance");
       jmsServer.setStateManager(sm);
 
       // We were either told the message cache or we get it from the
       // persistence manager
-      MessageCache mc = null;
+      MessageCache mc;
       if (messageCache != null)
          mc = (MessageCache) server.getAttribute(messageCache, "Instance");
       else
          mc = pm.getMessageCacheInstance();
       jmsServer.setMessageCache(mc);
 
+      ThreadPool tp;
+      ThreadGroup tg;
+      if (threadPool == null)
+      {
+         tg = new ThreadGroup("JBossMQ Server Threads");
+         tp = new BasicThreadPool("JMSThread", tg);
+      }
+      else
+      {
+         tp = (ThreadPool) server.getAttribute(threadPool, "Instance");
+         if (tp instanceof BasicThreadPool)
+           tg = ((BasicThreadPool)tp).getThreadGroup();
+         else
+           tg = new ThreadGroup("JBossMQ Server Threads");
+      }
+      jmsServer.setThreadPool(tp);
+      jmsServer.setThreadGroup(tg);
+
       jmsServer.startServer();
 
       super.startService();
    }
 
    protected void stopService()
    {
       jmsServer.stopServer();
Index: messaging/src/main/org/jboss/mq/server/jmx/DestinationManagerMBean.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/jmx/DestinationManagerMBean.java,v
retrieving revision 1.3.6.3
diff -b -B -U8 -r1.3.6.3 DestinationManagerMBean.java
--- messaging/src/main/org/jboss/mq/server/jmx/DestinationManagerMBean.java	12 Jul 2005 17:12:02 -0000	1.3.6.3
+++ messaging/src/main/org/jboss/mq/server/jmx/DestinationManagerMBean.java	26 Aug 2005 23:18:45 -0000
@@ -1,16 +1,18 @@
 /*
  * JBoss, the OpenSource J2EE webOS
  *
  * Distributable under LGPL license.
  * See terms of license at gnu.org.
  */
 package org.jboss.mq.server.jmx;
 
+import javax.management.ObjectName;
+
 /**
  * MBean interface.
  */
 public interface DestinationManagerMBean extends org.jboss.mq.server.jmx.InterceptorMBean
 {
 
    int getClientCount();
 
@@ -87,16 +89,36 @@
     */
    public int getRecoveryRetries();
 
    /**
     * Sets the class implementating the receivers
     */
    public void setRecoveryRetries(int retries);
 
+   /**
+    * Returns ThreadPool.
+    */
+   public ObjectName getThreadPool();
+
+   /**
+    * Sets ThreadPool.
+    */
+   public void setThreadPool(ObjectName threadPool);
+
+   /**
+    * Returns the expiry destination.
+    */
+   public ObjectName getExpiryDestination();
+
+   /**
+    * Sets the expiry destination.
+    */
+   public void setExpiryDestination(ObjectName destination);
+
    void createQueue(java.lang.String name) throws java.lang.Exception;
 
    void createTopic(java.lang.String name) throws java.lang.Exception;
 
    void createQueue(java.lang.String name, java.lang.String jndiLocation) throws java.lang.Exception;
 
    void createTopic(java.lang.String name, java.lang.String jndiLocation) throws java.lang.Exception;
 
Index: messaging/src/main/org/jboss/mq/threadpool/ThreadPool.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/threadpool/ThreadPool.java,v
retrieving revision 1.1
diff -b -B -U8 -r1.1 ThreadPool.java
--- messaging/src/main/org/jboss/mq/threadpool/ThreadPool.java	5 Oct 2001 19:34:14 -0000	1.1
+++ messaging/src/main/org/jboss/mq/threadpool/ThreadPool.java	26 Aug 2005 23:18:45 -0000
@@ -17,223 +17,9 @@
  *  This is an implementation of a simple thread pool with
  *  an embedded work queue.
  *
  * @author    Ole Husgaard (osh@sparre.dk)
  * @version   $Revision: 1.1 $
  */
 public class ThreadPool
 {
-   /**
-    *  The name of this thread pool
-    */
-   private String name;
-
-   /**
-    *  Flags that worker threads should be created as daemon threads.
-    */
-   private boolean daemon;
-
-   /**
-    *  The ThreadGroup of threads in this pool.
-    */
-   private ThreadGroup threadGroup;
-
-   /**
-    *  The worker threads.
-    */
-   private ArrayList workers;
-
-   /**
-    *  Maximum number of worker threads.
-    */
-   private int maxWorkers;
-
-   /**
-    *  Count of idle worker threads.
-    *  Synchronized on the [@link #workers} field.
-    */
-   private int idleWorkers;
-
-   /**
-    *  Flags that we are shutting down the pool.
-    *  Synchronized on the [@link #workers} field.
-    */
-   private volatile boolean stopping;
-
-   /**
-    *  The work queue.
-    */
-   private LinkedList queue;
-
-
-   /**
-    *  Create a new thread pool instance.
-    *
-    *  @param Name The name of this thread pool. This is used for naming its
-    *         worker threads.
-    *  @param threadGroup The <code>ThreadGroup</code> that worker threads
-    *         in this pool should belong to.
-    *  @param maxWorkers The maximum number of worker threads in this pool.
-    *  @param daemon If <code>true</code>, worker threads will be created as
-    *         daemon threads.
-    */
-   public ThreadPool(String name, ThreadGroup threadGroup, int maxWorkers,
-                     boolean daemon)
-   {
-      if (name == null || threadGroup == null || maxWorkers <= 0)
-         throw new IllegalArgumentException();
-
-      this.name = name;
-      this.daemon = daemon;
-      this.threadGroup = threadGroup;
-      workers = new ArrayList();
-      this.maxWorkers = maxWorkers;
-      idleWorkers = 0;
-      stopping = false;
-      queue = new LinkedList();
-   }
-
-   /**
-    *  Shutdown this thread pool.
-    *  This will not return until all enqueued work has been cancelled,
-    *  and all worker threads have done any work they started and have
-    *  died.
-    */
-    public void shutdown()
-    {
-       stopping = true;
-
-       synchronized (queue) {
-          // Remove all queued work
-          queue.clear();
-          // Notify all waiting threads
-          queue.notifyAll();
-       }
-
-       // wait for all worker threads to die.
-       synchronized (workers) {
-          while (workers.size() > 0) {
-             try {
-                // wait for some worker threads to die.
-                workers.wait();
-             } catch (InterruptedException ex) {
-                // ignore
-             }
-          }
-       }
-    }
-
-
-   /**
-    *  Enqueue a piece of work for this thread to handle.
-    *  As soon as a thread becomes available, it will call
-    *  {@link Work#doWork} of the argument.
-    *  If the pool is shutting down, this method will not enqueue the
-    *  work, but instead simply return.
-    *
-    *  @param work The piece of work to be enqueued.
-    */
-   public void enqueueWork(Work work)
-   {
-//System.err.println("ThreadPool("+name+"): enqueueWork() entered.");
-      // We may want to start a worker thread
-      synchronized (workers) {
-//System.err.println("ThreadPool("+name+"): enqueueWork(): idleWorkers="+idleWorkers+" stopping="+stopping+".");
-//System.err.println("ThreadPool("+name+"): enqueueWork(): workers.size()="+workers.size()+" maxWorkers="+maxWorkers+".");
-         if (idleWorkers == 0 && !stopping && workers.size() < maxWorkers)
-{
-            new WorkerThread(name + "-" + (workers.size() + 1)).start();
-//System.err.println("ThreadPool("+name+"): started new WorkerThread.");
-}
-      }
-
-      synchronized (queue) {
-         if (stopping)
-            return; // we are shutting down, cannot take new work.
-
-         queue.addLast(work);
-//System.err.println("ThreadPool("+name+"): enqueueWork(): enqueued work..");
-         queue.notify();
-      }
-   }
-
-   /**
-    *  Cancel a piece of enqueued work.
-    *
-    *  @param work The piece of work to be cancel.
-    */
-   public void cancelWork(Work work)
-   {
-      synchronized (queue) {
-         // It may be enqueued several times.
-         while (queue.remove(work))
-            ;
-      }
-   }
-
-
-   /**
-    *  The threads that do the actual work.
-    */
-   private class WorkerThread
-      extends Thread
-   {
-      /**
-       *  Create a new WorkerThread.
-       *  This must be called when holding the workers monitor.
-       */
-      WorkerThread(String name)
-      {
-         super(threadGroup, name);
-         setDaemon(daemon);
-         workers.add(this);
-//System.err.println("ThreadPool("+name+"): " + getName() + " created.");
-      }
-
-      /**
-       *  Wait for work do to.
-       *  This must be called when holding the queue monitor.
-       *  This will temporarily increment the count of idle workers
-       *  while waiting.
-       */
-      private void idle()
-      {
-         try {
-            synchronized (workers) {
-              ++idleWorkers;
-            }
-            //System.err.println("ThreadPool("+name+"): " + getName() + " starting to wait.");
-            queue.wait();
-         } catch (InterruptedException ex) {
-            // ignore
-         } finally {
-            //System.err.println("ThreadPool("+name+"): " + getName() + " done waiting.");
-            synchronized (workers) {
-              --idleWorkers;
-            }
-         }
-      }
-
-      public void run()
-      {
-//System.err.println("ThreadPool("+name+"): " + getName() + " started to run.");
-         while (!stopping) {
-            Work work = null;
-
-            synchronized (queue) {
-               if (queue.size() == 0)
-                  idle();
-               if (!stopping && queue.size() > 0)
-                  work = (Work)queue.removeFirst();
-            }
-
-            if (work != null)
-               work.doWork();
-         }
-         synchronized (workers) {
-            workers.remove(this);
-            // Notify the shutdown thread.
-            workers.notify();
-         }
-      }
-   }
 }
Index: messaging/src/main/org/jboss/mq/threadpool/Work.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/threadpool/Work.java,v
retrieving revision 1.1
diff -b -B -U8 -r1.1 Work.java
--- messaging/src/main/org/jboss/mq/threadpool/Work.java	5 Oct 2001 19:34:14 -0000	1.1
+++ messaging/src/main/org/jboss/mq/threadpool/Work.java	26 Aug 2005 23:18:45 -0000
@@ -13,13 +13,9 @@
  *  interface to have one of the threads in the thread pool call
  *  back the method declared here.
  *
  * @author    Ole Husgaard (osh@sparre.dk)
  * @version   $Revision: 1.1 $
  */
 public interface Work
 {
-   /**
-    *  Callback to do the actual work.
-    */
-   public void doWork();
 }
