[jboss-cvs] JBossRemoting/src/main/org/jboss/remoting/callback ...

[Ron Sigal]

  User: rsigal  
  Date: 06/10/28 15:15:20

  Modified:    src/main/org/jboss/remoting/callback  CallbackPoller.java
  Log:
  JBREM-605, JBREM-610, JBREM-617:
  
  (1) [JBREM-605, JBREM-610] Changed call to Client.acknowledgeCallback() to Client.acknowledgeCallbacks().  Also, added a continue to AcknowledgeThread.run() to prevent a possible NullPointerException.
  (2) [JBREM-617] Changed to use separate Timer thread.
  
  Revision  Changes    Path
  1.5       +25 -6     JBossRemoting/src/main/org/jboss/remoting/callback/CallbackPoller.java
  
  (In the diff below, changes in quantity of whitespace are not shown.)
  
  Index: CallbackPoller.java
  ===================================================================
  RCS file: /cvsroot/jboss/JBossRemoting/src/main/org/jboss/remoting/callback/CallbackPoller.java,v
  retrieving revision 1.4
  retrieving revision 1.5
  diff -u -b -r1.4 -r1.5
  --- CallbackPoller.java	12 Oct 2006 05:04:50 -0000	1.4
  +++ CallbackPoller.java	28 Oct 2006 19:15:20 -0000	1.5
  @@ -24,6 +24,7 @@
   import java.util.ArrayList;
   import java.util.List;
   import java.util.Map;
  +import java.util.Timer;
   import java.util.TimerTask;
   
   import org.jboss.logging.Logger;
  @@ -52,6 +53,7 @@
      private Map metadata = null;
      private Object callbackHandlerObject = null;
      private long pollPeriod = DEFAULT_POLL_PERIOD;
  +   private Timer timer;
      
      private ArrayList toHandleList = new ArrayList();
      private ArrayList toAcknowledgeList = new ArrayList();
  @@ -111,7 +113,12 @@
         handleThread = new HandleThread();
         handleThread.start();
   
  -      TimerUtil.schedule(this, pollPeriod);
  +      timer = new Timer(true);
  +      
  +      // If CallbackPoller is getting behind, it's probably because it's getting a burst
  +      // of callbacks.  Calling scheduleAtFixedRate() will cause the task to be run
  +      // extra times to accomodate the increase load.  See JBREM-618.
  +      timer.scheduleAtFixedRate(this, pollPeriod, pollPeriod);
   
      }
   
  @@ -143,13 +150,21 @@
         cancel();
         client = null;
         callbackHandler = null;
  +      if (handleThread != null)
  +      {
         handleThread.shutdown();
         handleThread = null;
  +      }
         if (acknowledgeThread != null)
         {
            acknowledgeThread.shutdown();
            acknowledgeThread = null;
         }
  +      if (timer != null)
  +      {
  +         timer.cancel();
  +         timer = null;
  +      }
      }
      
      class HandleThread extends Thread
  @@ -172,6 +187,7 @@
                     catch (InterruptedException e)
                     {
                        log.warn("unexpected interrupt");
  +                     continue;
                     }
                  }
                  callback = (Callback) toHandleList.remove(0);
  @@ -201,6 +217,7 @@
      {
         boolean running = true;
         Callback callback;
  +      ArrayList callbacksCopy = new ArrayList();
         
         public void run()
         {
  @@ -220,12 +237,14 @@
                        continue;
                     }
                  }
  -               callback = (Callback) toAcknowledgeList.remove(0);
  +               callbacksCopy.addAll(toAcknowledgeList);
  +               toAcknowledgeList.clear();
               }
               
               try
               {
  -               client.acknowledgeCallback(callbackHandler, callback);
  +               client.acknowledgeCallbacks(callbackHandler, callbacksCopy);
  +               callbacksCopy.clear();
               }
               catch (Throwable t)
               {