← Back to branch summary

~ubuntu-branches/ubuntu/raring/libjboss-remoting-java/raring

« back to all changes in this revision

Viewing changes to src/main/org/jboss/remoting/callback/CallbackPoller.java

  • Committer: Package Import Robot
  • Author(s): Torsten Werner
  • Date: 2011-09-09 14:01:03 UTC
  • mfrom: (1.1.6 upstream)
  • Revision ID: package-import@ubuntu.com-20110909140103-hqokx61534tas9rg
Tags: 2.5.3.SP1-1
http://bugs.debian.org/628303
* Newer but not newest upstream release. Do not build samples.
* Change debian/watch to upstream's svn repo.
* Add patch to fix compile error caused by tomcat update.
  (Closes: #628303)
* Switch to source format 3.0.
* Switch to debhelper level 7.
* Remove useless Depends.
* Update Standards-Version: 3.9.2.
* Update README.source.

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/main/org/jboss/remoting/callback/CallbackPoller.java
 
1
/*
 
2
* JBoss, Home of Professional Open Source
 
3
* Copyright 2005, JBoss Inc., and individual contributors as indicated
 
4
* by the @authors tag. See the copyright.txt in the distribution for a
 
5
* full listing of individual contributors.
 
6
*
 
7
* This is free software; you can redistribute it and/or modify it
 
8
* under the terms of the GNU Lesser General Public License as
 
9
* published by the Free Software Foundation; either version 2.1 of
 
10
* the License, or (at your option) any later version.
 
11
*
 
12
* This software is distributed in the hope that it will be useful,
 
13
* but WITHOUT ANY WARRANTY; without even the implied warranty of
 
14
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 
15
* Lesser General Public License for more details.
 
16
*
 
17
* You should have received a copy of the GNU Lesser General Public
 
18
* License along with this software; if not, write to the Free
 
19
* Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
 
20
* 02110-1301 USA, or see the FSF site: http://www.fsf.org.
 
21
*/
 
22
package org.jboss.remoting.callback;
 
23
 
 
24
import java.util.ArrayList;
 
25
import java.util.HashMap;
 
26
import java.util.Iterator;
 
27
import java.util.List;
 
28
import java.util.Map;
 
29
import java.util.Timer;
 
30
import java.util.TimerTask;
 
31
 
 
32
import org.jboss.logging.Logger;
 
33
import org.jboss.remoting.Client;
 
34
import org.jboss.remoting.ServerInvoker;
 
35
import org.jboss.remoting.transport.ClientInvoker;
 
36
 
 
37
/**
 
38
 * CallbackPoller is used to simulate push callbacks on transports that don't support
 
39
 * bidirectional connections.  It will periodically pull callbacks from the server
 
40
 * and pass them to the InvokerCallbackHandler.
 
41
 *
 
42
 * @author <a href="mailto:tom.elrod@jboss.com">Tom Elrod</a>
 
43
 * @author <a href="mailto:ron.sigal@jboss.com">Ron Sigal</a>
 
44
 */
 
45
public class CallbackPoller extends TimerTask implements Runnable
 
46
{
 
47
   /*
 
48
    * Implementation note.
 
49
    *
 
50
    * CallbackPoller uses two, or possibly three, threads.  The first thread is the
 
51
    * Timer thread, which periodically pulls callbacks from the server and adds them
 
52
    * to toHandleList.  The second thread takes callbacks from toHandleList, passes
 
53
    * them to the CallbackHandler, and, if an acknowledgement is requested for a
 
54
    * callback, it adds the callback to toAcknowledgeList.  The third thread, which is
 
55
    * created in response to the first callback for which an acknowledgement is requested,
 
56
    * takes the contents of toAcknowledgeList and acknowledges them in a batch.
 
57
    *
 
58
    * CallbackPoller will not shut down until all received callbacks have been processed
 
59
    * by the CallbackHandler and acknowledgements have been sent for all callbacks for
 
60
    * which acknowledgements have been requested.
 
61
    */
 
62
 
 
63
   /**
 
64
    * Default polling period for getting callbacks from the server.
 
65
    * Default is 5000 milliseconds.
 
66
    */
 
67
   public static final long DEFAULT_POLL_PERIOD = 5000;
 
68
   
 
69
   /**
 
70
    * Default timeout for getting callbacks in blocking mode.
 
71
    * Default is 5000 milliseconds.
 
72
    */
 
73
   public static final int DEFAULT_BLOCKING_TIMEOUT = 5000;
 
74
   
 
75
   /**
 
76
    * Default number of exceptions before callback polling wil be shut down.
 
77
    * Default is 5.
 
78
    */
 
79
   public static final int DEFAULT_MAX_ERROR_COUNT = 5;
 
80
   
 
81
   /**
 
82
    * The key value to use to specify if stop() should wait for the call to
 
83
    * org.jboss.remoting.Client.getCallbacks() should return.  The default 
 
84
    * behavior is do a synchronized shutdown for nonblocking callbacks and
 
85
    * a nonsynchronized shutdown for blocking callbacks.
 
86
    */
 
87
   public static final String SYNCHRONIZED_SHUTDOWN = "doSynchronizedShutdown";
 
88
   
 
89
   /**
 
90
    * The key value to use to specify the desired poll period
 
91
    * within the metadata Map.
 
92
    */
 
93
   public static final String CALLBACK_POLL_PERIOD = "callbackPollPeriod";
 
94
 
 
95
   /** Use java.util.timer.schedule(). */
 
96
   public static final String SCHEDULE_FIXED_RATE = "scheduleFixedRate";
 
97
 
 
98
   /** Use java.util.timer.scheduleAtFixedRate(). */
 
99
   public static final String SCHEDULE_FIXED_DELAY = "scheduleFixedDelay";
 
100
   
 
101
   /**
 
102
    * The key to use to specify the number of errors before callback polling
 
103
    * will be shut down.
 
104
    */
 
105
   public static final String MAX_ERROR_COUNT = "maxErrorCount";
 
106
 
 
107
   /** The key to use in metadata Map to request statistics.  The associated
 
108
    *  is ignored. */
 
109
   public static final String REPORT_STATISTICS = "reportStatistics";
 
110
 
 
111
   private Client client = null;
 
112
   private InvokerCallbackHandler callbackHandler = null;
 
113
   private Map metadata = null;
 
114
   private Object callbackHandlerObject = null;
 
115
   private boolean blocking = false;
 
116
   private boolean synchronizedShutdown = false;
 
117
   private long pollPeriod = DEFAULT_POLL_PERIOD;
 
118
   private Timer timer;
 
119
   private String scheduleMode = SCHEDULE_FIXED_RATE;
 
120
   private boolean reportStatistics;
 
121
   private boolean running;
 
122
   private int maxErrorCount = -1;
 
123
   private int errorCount;
 
124
   private boolean useAllParams;
 
125
 
 
126
   private ArrayList toHandleList = new ArrayList();
 
127
   private ArrayList toAcknowledgeList = new ArrayList();
 
128
   private HandleThread handleThread;
 
129
   private AcknowledgeThread acknowledgeThread;
 
130
   private BlockingPollerThread blockingPollerThread;
 
131
 
 
132
   private static final Logger log = Logger.getLogger(CallbackPoller.class);
 
133
 
 
134
 
 
135
   public CallbackPoller(Client client, InvokerCallbackHandler callbackhandler, Map metadata, Object callbackHandlerObject)
 
136
   {
 
137
      this.client = client;
 
138
      this.callbackHandler = callbackhandler;
 
139
      this.metadata = new HashMap(metadata);
 
140
      this.callbackHandlerObject = callbackHandlerObject;
 
141
   }
 
142
 
 
143
   public void start() throws Exception
 
144
   {
 
145
      if (callbackHandler == null)
 
146
      {
 
147
         throw new NullPointerException("Can not poll for callbacks when InvokerCallbackHandler is null.");
 
148
      }
 
149
      if (client != null)
 
150
      {
 
151
         client.connect();
 
152
      }
 
153
      else
 
154
      {
 
155
         throw new NullPointerException("Can not poll for callbacks when Client is null.");
 
156
      }
 
157
 
 
158
      configureParameters();
 
159
 
 
160
      handleThread = new HandleThread("HandleThread");
 
161
      handleThread.start();
 
162
      if (log.isTraceEnabled()) log.trace("blocking: " + blocking);
 
163
      if (blocking)
 
164
      {
 
165
         if (maxErrorCount == -1)
 
166
            maxErrorCount = DEFAULT_MAX_ERROR_COUNT;
 
167
         
 
168
         running = true;
 
169
         metadata.put(Client.THROW_CALLBACK_EXCEPTION, "true");
 
170
         blockingPollerThread = new BlockingPollerThread();
 
171
         blockingPollerThread.start();
 
172
      }
 
173
      else
 
174
      {
 
175
         timer = new Timer(true);
 
176
         if (SCHEDULE_FIXED_DELAY.equals(scheduleMode))
 
177
            timer.schedule(this, pollPeriod, pollPeriod);
 
178
         else
 
179
            timer.scheduleAtFixedRate(this, pollPeriod, pollPeriod);
 
180
      }
 
181
   }
 
182
 
 
183
   public synchronized void run()
 
184
   {
 
185
      // need to pull callbacks from server and give them to callback handler
 
186
      try
 
187
      {
 
188
         if (log.isTraceEnabled()) log.trace(this + " getting callbacks for " + callbackHandler);
 
189
         List callbacks = client.getCallbacks(callbackHandler, metadata);
 
190
         if (log.isTraceEnabled()) log.trace(this + " callback count: " + (callbacks == null ? 0 : callbacks.size()));
 
191
 
 
192
         if (callbacks != null && callbacks.size() > 0)
 
193
         {
 
194
            synchronized (toHandleList)
 
195
            {
 
196
               toHandleList.addAll(callbacks);
 
197
               if (toHandleList.size() == callbacks.size())
 
198
                  toHandleList.notify();
 
199
            }
 
200
         }
 
201
 
 
202
         if (reportStatistics)
 
203
            reportStatistics(callbacks);
 
204
      }
 
205
      catch (Throwable throwable)
 
206
      {
 
207
         if (!running)
 
208
         {
 
209
            stop();
 
210
            return;
 
211
         }
 
212
         
 
213
         log.info(this + " Error getting callbacks from server.");
 
214
         log.debug(this + " Error getting callbacks from server.", throwable);
 
215
         String errorMessage = throwable.getMessage();
 
216
         if (errorMessage != null)
 
217
         {
 
218
            if (errorMessage.startsWith("Could not find listener id"))
 
219
            {
 
220
               log.error("Client no longer has InvokerCallbackHandler (" + 
 
221
                          callbackHandler +
 
222
                         ") registered.  Shutting down callback polling");
 
223
               stop();
 
224
               return;
 
225
            }
 
226
            if (errorMessage.startsWith("Can not make remoting client invocation " +
 
227
                                        "due to not being connected to server."))
 
228
            {
 
229
               log.error("Client no longer connected.  Shutting down callback polling");
 
230
               stop();
 
231
               return;
 
232
            }
 
233
         }
 
234
         if (maxErrorCount >= 0)
 
235
         {
 
236
            if (++errorCount > maxErrorCount)
 
237
            {
 
238
               log.error("Error limit of " + maxErrorCount + 
 
239
                          " exceeded.  Shutting down callback polling");
 
240
               stop();
 
241
               return;
 
242
            }
 
243
         }
 
244
      }
 
245
   }
 
246
   
 
247
   public void stop()
 
248
   {
 
249
      stop(-1);
 
250
   }
 
251
 
 
252
   /**
 
253
    * stop() will not return until all received callbacks have been processed
 
254
    * by the CallbackHandler and acknowledgements have been sent for all callbacks for
 
255
    * which acknowledgements have been requested.
 
256
    */
 
257
   public void stop(int timeout)
 
258
   {
 
259
      log.debug(this + " is shutting down");
 
260
      running = false;
 
261
      
 
262
      if (!blocking)
 
263
      {
 
264
         cancel();
 
265
         
 
266
         if (timer != null)
 
267
         {
 
268
            timer.cancel();
 
269
            timer = null;
 
270
         }
 
271
      }
 
272
      
 
273
      if (timeout == 0)
 
274
         return;
 
275
      
 
276
      if (synchronizedShutdown)
 
277
      {
 
278
         // run() and stop() are synchronized so that stop() will wait until run() has finished
 
279
         // adding any callbacks it has received to toHandleList.  Therefore, once cancel()
 
280
         // returns, no more callbacks will arrive from the server.
 
281
         synchronized (this)
 
282
         {
 
283
            shutdown();
 
284
         }
 
285
      }
 
286
      else
 
287
      {
 
288
         shutdown();
 
289
      }
 
290
 
 
291
      log.debug(this + " has shut down");
 
292
   }
 
293
 
 
294
   
 
295
   private void shutdown()
 
296
   {
 
297
      // HandleThread.shutdown() will not return until all received callbacks have been
 
298
      // processed and, if necessary, added to toAcknowledgeList.
 
299
      if (handleThread != null)
 
300
      {
 
301
         handleThread.shutdown();
 
302
         handleThread = null;
 
303
      }
 
304
 
 
305
      // AcknowledgeThread.shutdown() will not return until acknowledgements have been sent
 
306
      // for all callbacks for which acknowledgements have been requested.
 
307
      if (acknowledgeThread != null)
 
308
      {
 
309
         acknowledgeThread.shutdown();
 
310
         acknowledgeThread = null;
 
311
      }
 
312
   }
 
313
   
 
314
   
 
315
   class BlockingPollerThread extends Thread
 
316
   {
 
317
      public BlockingPollerThread()
 
318
      {
 
319
         String threadName = getName();
 
320
         int i = threadName.indexOf('-');
 
321
         String threadNumber = null;
 
322
         if (i >= 0)
 
323
            threadNumber = threadName.substring(i+1);
 
324
         else
 
325
            threadNumber = Long.toString(System.currentTimeMillis());
 
326
         String pollerString = CallbackPoller.this.toString();
 
327
         String address = pollerString.substring(pollerString.indexOf('@'));
 
328
         setName("CallbackPoller:" + threadNumber + "[" + address + "]");
 
329
         setDaemon(true);
 
330
      }
 
331
 
 
332
      public void run()
 
333
      {
 
334
         while (running)
 
335
         {
 
336
            CallbackPoller.this.run();
 
337
         }
 
338
      }
 
339
   }  
 
340
 
 
341
   
 
342
   class HandleThread extends Thread
 
343
   {
 
344
      boolean running = true;
 
345
      boolean done;
 
346
      ArrayList toHandleListCopy = new ArrayList();
 
347
      Callback callback;
 
348
 
 
349
      HandleThread(String name)
 
350
      {
 
351
         super(name);
 
352
      }
 
353
      public void run()
 
354
      {
 
355
         while (true)
 
356
         {
 
357
            synchronized (toHandleList)
 
358
            {
 
359
               if (toHandleList.isEmpty() && running)
 
360
               {
 
361
                  try
 
362
                  {
 
363
                     toHandleList.wait();
 
364
                  }
 
365
                  catch (InterruptedException e)
 
366
                  {
 
367
                     log.debug("unexpected interrupt");
 
368
                     continue;
 
369
                  }
 
370
               }
 
371
 
 
372
               // If toHandleList is empty, then running must be false.  We return
 
373
               // only when both conditions are true.
 
374
               if (toHandleList.isEmpty())
 
375
               {
 
376
                  done = true;
 
377
                  toHandleList.notify();
 
378
                  return;
 
379
               }
 
380
 
 
381
               toHandleListCopy.addAll(toHandleList);
 
382
               toHandleList.clear();
 
383
            }
 
384
 
 
385
            while (!toHandleListCopy.isEmpty())
 
386
            {
 
387
               try
 
388
               {
 
389
                  callback = (Callback) toHandleListCopy.remove(0);
 
390
                  callback.setCallbackHandleObject(callbackHandlerObject);
 
391
                  callbackHandler.handleCallback(callback);
 
392
               }
 
393
               catch (HandleCallbackException e)
 
394
               {
 
395
                  log.error("Error delivering callback to callback handler (" + callbackHandler + ").", e);
 
396
               }
 
397
 
 
398
               checkForAcknowledgeRequest(callback);
 
399
            }
 
400
         }
 
401
      }
 
402
 
 
403
      /**
 
404
       *  Once CallbackPoller.stop() has called HandleThread.shutdown(), CallbackPoller.run()
 
405
       *  has terminated and no additional callbacks will be received.  shutdown() will
 
406
       *  not return until HandleThread has processed all received callbacks.
 
407
       *
 
408
       *  Either run() or shutdown() will enter its own synchronized block first.
 
409
       *
 
410
       *  case 1): run() enters its synchronized block first:
 
411
       *     If toHandleList is empty, then run() will reach toHandleList.wait(), shutdown()
 
412
       *     will wake up run(), and run() will exit.  If toHandleList is not empty, then run()
 
413
       *     will process all outstanding callbacks and return to its synchronized block.  At
 
414
       *     this point, either case 1) (with toHandleList empty) or case 2) applies.
 
415
       *
 
416
       *  case 2): shutdown() enters its synchronized block first:
 
417
       *     run() will process all outstanding callbacks and return to its synchronized block.
 
418
       *     After shutdown() reaches toHandleList.wait(), run() will enter its synchronized
 
419
       *     block, find running == false and toHandleList empty, and it will exit.
 
420
       */
 
421
      protected void shutdown()
 
422
      {
 
423
         log.debug(this + " is shutting down");
 
424
         synchronized (toHandleList)
 
425
         {
 
426
            running = false;
 
427
            toHandleList.notify();
 
428
            while (!done)
 
429
            {
 
430
               try
 
431
               {
 
432
                  toHandleList.wait();
 
433
               }
 
434
               catch (InterruptedException ignored) {}
 
435
            }
 
436
         }
 
437
         log.debug(this + " has shut down");
 
438
         return;
 
439
      }
 
440
   }
 
441
 
 
442
 
 
443
   class AcknowledgeThread extends Thread
 
444
   {
 
445
      boolean running = true;
 
446
      boolean done;
 
447
      ArrayList toAcknowledgeListCopy = new ArrayList();
 
448
 
 
449
      AcknowledgeThread(String name)
 
450
      {
 
451
         super(name);
 
452
      }
 
453
      public void run()
 
454
      {
 
455
         while (true)
 
456
         {
 
457
            synchronized (toAcknowledgeList)
 
458
            {
 
459
               while (toAcknowledgeList.isEmpty() && running)
 
460
               {
 
461
                  try
 
462
                  {
 
463
                     toAcknowledgeList.wait();
 
464
                  }
 
465
                  catch (InterruptedException e)
 
466
                  {
 
467
                     log.debug("unexpected interrupt");
 
468
                     continue;
 
469
                  }
 
470
               }
 
471
 
 
472
               // If toAcknowledgeList is empty, then running must be false.  We return
 
473
               // only when both conditions are true.
 
474
               if (toAcknowledgeList.isEmpty())
 
475
               {
 
476
                  done = true;
 
477
                  toAcknowledgeList.notify();
 
478
                  return;
 
479
               }
 
480
 
 
481
               toAcknowledgeListCopy.addAll(toAcknowledgeList);
 
482
               toAcknowledgeList.clear();
 
483
            }
 
484
 
 
485
            try
 
486
            {
 
487
               if (log.isTraceEnabled())
 
488
               {
 
489
                  Iterator it = toAcknowledgeListCopy.iterator();
 
490
                  while (it.hasNext())
 
491
                  {
 
492
                     Callback cb = (Callback) it.next();
 
493
                     Map map = cb.getReturnPayload();
 
494
                     log.trace("acknowledging: " + map.get(ServerInvokerCallbackHandler.CALLBACK_ID));
 
495
                  }
 
496
               }
 
497
               client.acknowledgeCallbacks(callbackHandler, toAcknowledgeListCopy);
 
498
               toAcknowledgeListCopy.clear();
 
499
            }
 
500
            catch (Throwable t)
 
501
            {
 
502
               log.error("Error acknowledging callback for callback handler (" + callbackHandler + ").", t);
 
503
            }
 
504
         }
 
505
      }
 
506
 
 
507
      /**
 
508
       *  Once CallbackPoller.stop() has called AcknowledgeThread.shutdown(), HandleThread
 
509
       *  has terminated and no additional callbacks will be added to toAcknowledgeList.
 
510
       *  shutdown() will not return until AcknowledgeThread has acknowledged all callbacks
 
511
       *  in toAcknowledgeList.
 
512
       *
 
513
       *  Either run() or shutdown() will enter its own synchronized block first.
 
514
       *
 
515
       *  case 1): run() enters its synchronized block first:
 
516
       *     If toAcknowledgeList is empty, then run() will reach toAcknowledgeList.wait(),
 
517
       *     shutdown() will wake up run(), and run() will exit.  If toAcknowledgeList is not
 
518
       *     empty, then run() will process all callbacks in toAcknowledgeList and return to
 
519
       *     its synchronized block.  At this point, either case 1) (with toAcknowledgeList
 
520
       *     empty) or case 2) applies.
 
521
       *
 
522
       *  case 2): shutdown() enters its synchronized block first:
 
523
       *     run() will process all callbacks in toAcknowledgeList and return to its
 
524
       *     synchronized block.  After shutdown() reaches toAcknowledgeList.wait(), run()
 
525
       *     will enter its synchronized block, find running == false and toAcknowledgeList
 
526
       *     empty, and it will exit.
 
527
       */
 
528
      public void shutdown()
 
529
      {
 
530
         log.debug(this + " is shutting down");      
 
531
         synchronized (toAcknowledgeList)
 
532
         {
 
533
            running = false;
 
534
            toAcknowledgeList.notify();
 
535
            while (!done)
 
536
            {
 
537
               try
 
538
               {
 
539
                  toAcknowledgeList.wait();
 
540
               }
 
541
               catch (InterruptedException ignored) {}
 
542
            }
 
543
         }
 
544
         log.debug(this + " has shut down");
 
545
         return;
 
546
      }
 
547
   }
 
548
 
 
549
 
 
550
   private void checkForAcknowledgeRequest(Callback callback)
 
551
   {
 
552
      Map returnPayload = callback.getReturnPayload();
 
553
      if (returnPayload != null)
 
554
      {
 
555
         Object callbackId = returnPayload.get(ServerInvokerCallbackHandler.CALLBACK_ID);
 
556
         if (callbackId != null)
 
557
         {
 
558
            Object o = returnPayload.get(ServerInvokerCallbackHandler.REMOTING_ACKNOWLEDGES_PUSH_CALLBACKS);
 
559
            if (o instanceof String  && Boolean.valueOf((String)o).booleanValue() ||
 
560
                o instanceof Boolean && ((Boolean)o).booleanValue())
 
561
            {
 
562
               synchronized (toAcknowledgeList)
 
563
               {
 
564
                  toAcknowledgeList.add(callback);
 
565
                  if (toAcknowledgeList.size() == 1)
 
566
                  {
 
567
                     if (acknowledgeThread == null)
 
568
                     {
 
569
                        acknowledgeThread = new AcknowledgeThread("AcknowledgeThread");
 
570
                        acknowledgeThread.start();
 
571
                     }
 
572
                     else
 
573
                     {
 
574
                        toAcknowledgeList.notify();
 
575
                     }
 
576
                  }
 
577
               }
 
578
            }
 
579
         }
 
580
      }
 
581
   }
 
582
   
 
583
   
 
584
   private void configureParameters()
 
585
   {
 
586
      Map config = new HashMap();
 
587
      ClientInvoker invoker = client.getInvoker();
 
588
      if (invoker != null)
 
589
      {
 
590
         config.putAll(invoker.getLocator().getParameters());
 
591
      }
 
592
      config.putAll(client.getConfiguration());
 
593
      config.putAll(metadata);
 
594
      
 
595
      Object val = config.get(Client.USE_ALL_PARAMS);
 
596
      if (val != null)
 
597
      {
 
598
         if (val instanceof String)
 
599
         {
 
600
            useAllParams = Boolean.valueOf((String) val).booleanValue();
 
601
         }
 
602
         else
 
603
         {
 
604
            log.warn("Value for " + Client.USE_ALL_PARAMS + " must be of type " +
 
605
                     String.class.getName() + " and is " + val.getClass().getName());
 
606
         }
 
607
      }
 
608
      log.debug(this + ": useAllParams: " + useAllParams);
 
609
      if (!useAllParams)
 
610
      {
 
611
         config = metadata;
 
612
      }
 
613
      
 
614
      val = config.get(ServerInvoker.BLOCKING_MODE);
 
615
      if (val != null)
 
616
      {
 
617
         if (val instanceof String)
 
618
         {
 
619
            if (ServerInvoker.BLOCKING.equals(val))
 
620
            {
 
621
               blocking = true;
 
622
               synchronizedShutdown = false;
 
623
            }
 
624
            else if (ServerInvoker.NONBLOCKING.equals(val))
 
625
            {
 
626
               blocking = false;
 
627
               synchronizedShutdown = true;
 
628
            }
 
629
            else
 
630
            {
 
631
               log.warn("Value for " + ServerInvoker.BLOCKING_MODE + 
 
632
                     " configuration is " + val + ". Must be either " +
 
633
                     ServerInvoker.BLOCKING + " or " + ServerInvoker.NONBLOCKING +
 
634
                     ". Using " + ServerInvoker.BLOCKING + ".");
 
635
            }
 
636
         }
 
637
         else
 
638
         {
 
639
            log.warn("Value for " + ServerInvoker.BLOCKING_MODE + 
 
640
                  " configuration must be of type " + String.class.getName() +
 
641
                  " and is of type " + val.getClass().getName());
 
642
         }
 
643
      }
 
644
 
 
645
      // Default blocking mode on server is nonblocking.
 
646
      if (blocking)
 
647
         metadata.put(ServerInvoker.BLOCKING_MODE, ServerInvoker.BLOCKING);
 
648
 
 
649
      val = config.get(ServerInvoker.BLOCKING_TIMEOUT);
 
650
      if (val != null)
 
651
      {
 
652
         if (val instanceof String)
 
653
         {
 
654
            try
 
655
            {
 
656
               int blockingTimeout = Integer.parseInt((String) val);
 
657
               metadata.put(ServerInvoker.TIMEOUT, Integer.toString(blockingTimeout));
 
658
            }
 
659
            catch (NumberFormatException e)
 
660
            {
 
661
               log.warn("Error converting " + ServerInvoker.BLOCKING_TIMEOUT + " to type long.  " + e.getMessage());
 
662
            }
 
663
         }
 
664
         else
 
665
         {
 
666
            log.warn("Value for " + ServerInvoker.BLOCKING_TIMEOUT + " configuration must be of type " + String.class.getName() +
 
667
                  " and is " + val.getClass().getName());
 
668
         }
 
669
      }
 
670
 
 
671
      val = config.get(SYNCHRONIZED_SHUTDOWN);
 
672
      if (val != null)
 
673
      {
 
674
         if (val instanceof String)
 
675
         {
 
676
            synchronizedShutdown = Boolean.valueOf((String) val).booleanValue();
 
677
         }
 
678
         else
 
679
         {
 
680
            log.warn("Value for " + SYNCHRONIZED_SHUTDOWN + " must be of type " + String.class.getName() +
 
681
                  " and is " + val.getClass().getName());
 
682
         }
 
683
      }
 
684
 
 
685
      val = config.get(CALLBACK_POLL_PERIOD);
 
686
      if (val != null)
 
687
      {
 
688
         if (val instanceof String)
 
689
         {
 
690
            try
 
691
            {
 
692
               pollPeriod = Long.parseLong((String) val);
 
693
            }
 
694
            catch (NumberFormatException e)
 
695
            {
 
696
               log.warn("Error converting " + CALLBACK_POLL_PERIOD + " to type long.  " + e.getMessage());
 
697
            }
 
698
         }
 
699
         else
 
700
         {
 
701
            log.warn("Value for " + CALLBACK_POLL_PERIOD + " configuration must be of type " + String.class.getName() +
 
702
                  " and is " + val.getClass().getName());
 
703
         }
 
704
      }
 
705
      val = config.get(CALLBACK_SCHEDULE_MODE);
 
706
      if (val != null)
 
707
      {
 
708
         if (val instanceof String)
 
709
         {
 
710
            if (SCHEDULE_FIXED_DELAY.equals(val) || SCHEDULE_FIXED_RATE.equals(val))
 
711
            {
 
712
               scheduleMode = (String) val;
 
713
            }
 
714
            else
 
715
            {
 
716
               log.warn("Unrecognized value for " + CALLBACK_SCHEDULE_MODE + ": " + val);
 
717
               log.warn("Using " + scheduleMode);
 
718
            }
 
719
         }
 
720
         else
 
721
         {
 
722
            log.warn("Value for " + CALLBACK_SCHEDULE_MODE + " must be of type " + String.class.getName() +
 
723
                  " and is " + val.getClass().getName());
 
724
         }
 
725
      }
 
726
      val = config.get(MAX_ERROR_COUNT);
 
727
      if (val != null)
 
728
      {
 
729
         if (val instanceof String)
 
730
         {
 
731
            try
 
732
            {
 
733
               maxErrorCount = Integer.parseInt((String) val);
 
734
            }
 
735
            catch (NumberFormatException e)
 
736
            {
 
737
               log.warn("Error converting " + MAX_ERROR_COUNT + " to type int.  " + e.getMessage());
 
738
            }
 
739
         }
 
740
         else
 
741
         {
 
742
            log.warn("Value for " + MAX_ERROR_COUNT + " configuration must be of type " + String.class.getName() +
 
743
                  " and is " + val.getClass().getName());
 
744
         }
 
745
      }
 
746
      if (config.get(REPORT_STATISTICS) != null)
 
747
      {
 
748
         reportStatistics = true;
 
749
      }
 
750
   }
 
751
 
 
752
 
 
753
   private void reportStatistics(List callbacks)
 
754
   {
 
755
      int toHandle;
 
756
      int toAcknowledge = 0;
 
757
 
 
758
      synchronized (toHandleList)
 
759
      {
 
760
         toHandle = toHandleList.size() + handleThread.toHandleListCopy.size();
 
761
      }
 
762
 
 
763
      synchronized (toAcknowledgeList)
 
764
      {
 
765
         if (acknowledgeThread != null)
 
766
            toAcknowledge = toAcknowledgeList.size() + acknowledgeThread.toAcknowledgeListCopy.size();
 
767
      }
 
768
 
 
769
      StringBuffer message = new StringBuffer("\n");
 
770
      message.append("================================\n")
 
771
             .append("  retrieved " + callbacks.size() + " callbacks\n")
 
772
             .append("  callbacks waiting to be processed: " + toHandle + "\n")
 
773
             .append("  callbacks waiting to be acknowledged: " + toAcknowledge + "\n")
 
774
             .append("================================");
 
775
      log.info(message);
 
776
   }
 
777
 
 
778
 
 
779
   /**
 
780
    * The key value to use in metadata Map to specify the desired scheduling mode. 
 
781
    */
 
782
   public static final String CALLBACK_SCHEDULE_MODE = "scheduleMode";
 
783
}
 
 
b'\\ No newline at end of file'