← 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/transport/bisocket/BisocketServerInvoker.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/transport/bisocket/BisocketServerInvoker.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
 
 
23
package org.jboss.remoting.transport.bisocket;
 
24
 
 
25
import java.io.DataInputStream;
 
26
import java.io.DataOutputStream;
 
27
import java.io.IOException;
 
28
import java.lang.reflect.Method;
 
29
import java.net.InetAddress;
 
30
import java.net.ServerSocket;
 
31
import java.net.Socket;
 
32
import java.security.AccessController;
 
33
import java.security.PrivilegedActionException;
 
34
import java.security.PrivilegedExceptionAction;
 
35
import java.util.ArrayList;
 
36
import java.util.Collection;
 
37
import java.util.Collections;
 
38
import java.util.HashMap;
 
39
import java.util.HashSet;
 
40
import java.util.Iterator;
 
41
import java.util.LinkedList;
 
42
import java.util.List;
 
43
import java.util.Map;
 
44
import java.util.Set;
 
45
import java.util.StringTokenizer;
 
46
import java.util.Timer;
 
47
import java.util.TimerTask;
 
48
 
 
49
import org.jboss.logging.Logger;
 
50
import org.jboss.remoting.Client;
 
51
import org.jboss.remoting.Home;
 
52
import org.jboss.remoting.InvocationRequest;
 
53
import org.jboss.remoting.InvokerLocator;
 
54
import org.jboss.remoting.Remoting;
 
55
import org.jboss.remoting.ServerInvocationHandler;
 
56
import org.jboss.remoting.invocation.InternalInvocation;
 
57
import org.jboss.remoting.socketfactory.CreationListenerServerSocket;
 
58
import org.jboss.remoting.socketfactory.CreationListenerSocketFactory;
 
59
import org.jboss.remoting.socketfactory.SocketCreationListener;
 
60
import org.jboss.remoting.transport.PortUtil;
 
61
import org.jboss.remoting.transport.socket.LRUPool;
 
62
import org.jboss.remoting.transport.socket.SocketServerInvoker;
 
63
import org.jboss.remoting.util.SecurityUtility;
 
64
 
 
65
 
 
66
/**
 
67
 *
 
68
 * @author <a href="ron.sigal@jboss.com">Ron Sigal</a>
 
69
 * @version $Revision: 5605 $
 
70
 * <p>
 
71
 * Copyright Nov 23, 2006
 
72
 * </p>
 
73
 */
 
74
public class BisocketServerInvoker extends SocketServerInvoker
 
75
{
 
76
   private static final Logger log = Logger.getLogger(BisocketServerInvoker.class);
 
77
 
 
78
   private static Map listenerIdToServerInvokerMap = Collections.synchronizedMap(new HashMap());
 
79
   private static Timer timer;
 
80
   private static Object timerLock = new Object();
 
81
 
 
82
   private Map listenerIdToInvokerLocatorMap = Collections.synchronizedMap(new HashMap());
 
83
   private Set secondaryServerSockets = new HashSet();
 
84
   private InvokerLocator secondaryLocator;
 
85
   private Set secondaryServerSocketThreads = new HashSet();
 
86
   private Map controlConnectionThreadMap = new HashMap();
 
87
   private Map controlConnectionRestartsMap = Collections.synchronizedMap(new HashMap());
 
88
   private int pingFrequency = Bisocket.PING_FREQUENCY_DEFAULT;
 
89
   private int pingWindowFactor = Bisocket.PING_WINDOW_FACTOR_DEFAULT;
 
90
   private int pingWindow = pingWindowFactor * pingFrequency;
 
91
   private int socketCreationRetries = Bisocket.MAX_RETRIES_DEFAULT;
 
92
   private int controlConnectionRestarts = Bisocket.MAX_CONTROL_CONNECTION_RESTARTS_DEFAULT;
 
93
   private ControlMonitorTimerTask controlMonitorTimerTask;
 
94
   protected boolean isCallbackServer = false;
 
95
   protected List secondaryBindPorts = new ArrayList();
 
96
   protected List secondaryConnectPorts = new ArrayList();
 
97
 
 
98
 
 
99
   public static BisocketServerInvoker getBisocketServerInvoker(String listenerId)
 
100
   {
 
101
      return (BisocketServerInvoker) listenerIdToServerInvokerMap.get(listenerId);
 
102
   }
 
103
 
 
104
 
 
105
   public BisocketServerInvoker(InvokerLocator locator)
 
106
   {
 
107
      super(locator);
 
108
   }
 
109
 
 
110
 
 
111
   public BisocketServerInvoker(InvokerLocator locator, Map configuration)
 
112
   {
 
113
      super(locator, configuration);
 
114
   }
 
115
 
 
116
 
 
117
   public void start() throws IOException
 
118
   {
 
119
      if (isCallbackServer)
 
120
      {
 
121
         Object val = configuration.get(Bisocket.MAX_RETRIES);
 
122
         if (val != null)
 
123
         {
 
124
            try
 
125
            {
 
126
               int nVal = Integer.valueOf((String) val).intValue();
 
127
               socketCreationRetries = nVal;
 
128
               log.debug("Setting socket creation retry limit: " + socketCreationRetries);
 
129
            }
 
130
            catch (Exception e)
 
131
            {
 
132
               log.warn("Could not convert " + Bisocket.MAX_RETRIES +
 
133
                     " value of " + val + " to an int value.");
 
134
            }
 
135
         }
 
136
         
 
137
         val = configuration.get(Bisocket.MAX_CONTROL_CONNECTION_RESTARTS);
 
138
         if (val != null)
 
139
         {
 
140
            try
 
141
            {
 
142
               int nVal = Integer.valueOf((String) val).intValue();
 
143
               controlConnectionRestarts = nVal;
 
144
               log.debug("Setting control connection restart limit: " + controlConnectionRestarts);
 
145
            }
 
146
            catch (Exception e)
 
147
            {
 
148
               log.warn("Could not convert " + Bisocket.MAX_CONTROL_CONNECTION_RESTARTS +
 
149
                     " value of " + val + " to an int value.");
 
150
            }
 
151
         }
 
152
         
 
153
         if(maxPoolSize <= 0)
 
154
         {
 
155
            maxPoolSize = MAX_POOL_SIZE_DEFAULT;
 
156
         }
 
157
         clientpool = new LRUPool(2, maxPoolSize);
 
158
         clientpool.create();
 
159
         threadpool = new LinkedList();
 
160
         checkSocketFactoryWrapper();
 
161
 
 
162
         if (pingFrequency > 0)
 
163
         {
 
164
            controlMonitorTimerTask = new ControlMonitorTimerTask(this);
 
165
            synchronized (timerLock)
 
166
            {
 
167
               if (timer == null)
 
168
               {
 
169
                  timer = new Timer(true);
 
170
               }
 
171
               try
 
172
               {
 
173
                  timer.schedule(controlMonitorTimerTask, pingFrequency, pingFrequency);
 
174
               }
 
175
               catch (IllegalStateException e)
 
176
               {
 
177
                  log.debug("Unable to schedule TimerTask on existing Timer", e);
 
178
                  timer = new Timer(true);
 
179
                  timer.schedule(controlMonitorTimerTask, pingFrequency, pingFrequency);
 
180
               }
 
181
            }
 
182
         }
 
183
         
 
184
         running = true;
 
185
         started = true;
 
186
      }
 
187
      else
 
188
      {
 
189
         super.start();
 
190
         
 
191
         if (serverSockets.size() < secondaryBindPorts.size())
 
192
            log.warn(this + " extra secondary bind ports will be ignored");
 
193
         else if (serverSockets.size() > secondaryBindPorts.size())
 
194
            log.warn(this + " not enough secondary bind ports: will use anonymous ports as necessary");
 
195
        
 
196
         if (secondaryConnectPorts.size() == 0)
 
197
         {
 
198
            secondaryConnectPorts = secondaryBindPorts;
 
199
         }
 
200
         else if(secondaryConnectPorts.size() != secondaryBindPorts.size())
 
201
         {
 
202
            log.warn(this + " number of secondary connect ports != number of secondary bind ports");
 
203
            log.warn(this + " will ignore secondary connect ports");
 
204
            secondaryConnectPorts = secondaryBindPorts;
 
205
         }         
 
206
 
 
207
         int i = 0;
 
208
         Iterator it = serverSockets.iterator();
 
209
         while (it.hasNext())
 
210
         {
 
211
            ServerSocket ss = (ServerSocket) it.next();
 
212
            final InetAddress host = ss.getInetAddress();
 
213
            int secondaryBindPort = -1;
 
214
            if (secondaryBindPorts.size() > i)
 
215
            {
 
216
               secondaryBindPort = ((Integer) secondaryBindPorts.get(i)).intValue();
 
217
            }
 
218
            else
 
219
            {
 
220
               secondaryBindPorts.add(new Integer(-1));
 
221
            }
 
222
            if (secondaryBindPort < 0)
 
223
            {
 
224
               secondaryBindPort = PortUtil.findFreePort(host.getHostAddress());
 
225
               secondaryBindPorts.set(i, new Integer(secondaryBindPort));
 
226
            }
 
227
            
 
228
            ServerSocket secondaryServerSocket = null;
 
229
            final int finalBindPort = secondaryBindPort;
 
230
            
 
231
            try
 
232
            {
 
233
               secondaryServerSocket = (ServerSocket) AccessController.doPrivileged( new PrivilegedExceptionAction()
 
234
               {
 
235
                  public Object run() throws Exception
 
236
                  {
 
237
                     ServerSocket ss = null;
 
238
                     if (serverSocketFactory != null)
 
239
                     {
 
240
                        ss = serverSocketFactory.createServerSocket(finalBindPort, 0, host);
 
241
                     }
 
242
                     else
 
243
                     {
 
244
                        ss = new ServerSocket(finalBindPort, 0, host);
 
245
                     }
 
246
                     return ss;
 
247
                  }
 
248
               });
 
249
            }
 
250
            catch (PrivilegedActionException e)
 
251
            {
 
252
               throw (IOException) e.getCause();
 
253
            }
 
254
            
 
255
            ss = checkSecondaryServerSocketWrapper(secondaryServerSocket);
 
256
            secondaryServerSockets.add(ss);
 
257
            log.debug(this + " created secondary " + ss);
 
258
            i++;
 
259
         }
 
260
         
 
261
         i = 0;
 
262
         it = secondaryServerSockets.iterator();
 
263
         while (it.hasNext())
 
264
         {
 
265
            ServerSocket secondaryServerSocket = (ServerSocket) it.next();
 
266
            Thread t = new SecondaryServerSocketThread(secondaryServerSocket);
 
267
            t.setName("secondaryServerSocketThread[" + i++ + "]");
 
268
            t.setDaemon(true);
 
269
            t.start();
 
270
            secondaryServerSocketThreads.add(t);
 
271
            log.debug(this + " created " + t);
 
272
         }
 
273
         
 
274
         if (getLocator().isMultihome())
 
275
         {
 
276
            int j = 0;
 
277
            String host = ((Home) connectHomes.get(j)).host;
 
278
            int port = ((Integer) secondaryConnectPorts.get(j)).intValue();
 
279
            if (port < 0)
 
280
               port = ((Integer) secondaryBindPorts.get(j)).intValue();
 
281
            StringBuffer sb = new StringBuffer(host).append(':').append(port);
 
282
            for (j = 1; j < connectHomes.size(); j++)
 
283
            {
 
284
               host = ((Home) connectHomes.get(j)).host;
 
285
               port = ((Integer) secondaryConnectPorts.get(j)).intValue();
 
286
               if (port < 0)
 
287
                  port = ((Integer) secondaryBindPorts.get(j)).intValue();
 
288
               sb.append('!').append(host).append(':').append(port);
 
289
            }
 
290
            
 
291
            Map params = new HashMap();
 
292
            params.put(InvokerLocator.HOMES_KEY, sb.toString());
 
293
            secondaryLocator = new InvokerLocator(null, InvokerLocator.MULTIHOME, -1, null, params);
 
294
         }
 
295
         else
 
296
         {
 
297
            String connectAddress = getLocator().getHost();
 
298
            int connectPort = ((Integer) secondaryConnectPorts.get(0)).intValue();
 
299
            if (connectPort < 0)
 
300
               connectPort = ((Integer) secondaryBindPorts.get(0)).intValue();
 
301
            secondaryLocator = new InvokerLocator(null, connectAddress, connectPort, null, null);
 
302
         }
 
303
 
 
304
         log.debug(this + " created secondary InvokerLocator: " + secondaryLocator);
 
305
      }
 
306
   }
 
307
 
 
308
 
 
309
   public boolean isTransportBiDirectional()
 
310
   {
 
311
      return true;
 
312
   }
 
313
 
 
314
 
 
315
   public void createControlConnection(String listenerId, boolean firstConnection)
 
316
   throws IOException
 
317
   {
 
318
      BisocketClientInvoker clientInvoker = BisocketClientInvoker.getBisocketClientInvoker(listenerId);
 
319
      
 
320
      if (clientInvoker == null)
 
321
      {
 
322
         log.debug("Unable to retrieve client invoker: must have disconnected");
 
323
         throw new ClientUnavailableException();
 
324
      }
 
325
      
 
326
      InvokerLocator oldLocator = (InvokerLocator) listenerIdToInvokerLocatorMap.get(listenerId);
 
327
      InvokerLocator newLocator = null;
 
328
      
 
329
      try
 
330
      {
 
331
         newLocator = clientInvoker.getSecondaryLocator();
 
332
      }
 
333
      catch (Throwable t)
 
334
      {
 
335
         log.debug("unable to get secondary locator", t);
 
336
         throw new IOException("unable to get secondary locator: " + t.getMessage());
 
337
      }
 
338
      
 
339
 
 
340
      // If a server restarts, it is likely that it creates a new secondary server socket on
 
341
      // a different port.  It will possible to recreate the control connection, but if
 
342
      // there is no PingTimerTask running in the new server to keep it alive, it will just
 
343
      // die again.  Once a new secondary server socket address is detected, a count is kept
 
344
      // of the number of times the control connection is restarted, and when it hits a
 
345
      // configured maximum, it is allowed to die.  See JBREM-731.
 
346
      
 
347
      boolean locatorChanged = !newLocator.equals(oldLocator);
 
348
      listenerIdToInvokerLocatorMap.put(listenerId, newLocator);
 
349
      
 
350
      String host = newLocator.getHost();
 
351
      int port = newLocator.getPort();
 
352
      if (newLocator.isMultihome())
 
353
      {
 
354
         host = clientInvoker.getHomeInUse().host;
 
355
         port = -1;
 
356
         Iterator it = null;
 
357
         if (newLocator.getConnectHomeList().isEmpty())
 
358
            it =  newLocator.getHomeList().iterator();
 
359
         else
 
360
            it = newLocator.getConnectHomeList().iterator();
 
361
         
 
362
         while (it.hasNext())
 
363
         {
 
364
            Home h = (Home) it.next();
 
365
            if (host.equals(h.host))
 
366
            {
 
367
               port = h.port;
 
368
               newLocator.setHomeInUse(h);
 
369
               break;
 
370
            }
 
371
         }
 
372
      }
 
373
      
 
374
      if (port == -1)
 
375
      {
 
376
         throw new IOException("Cannot find matching home for control connection");
 
377
      }
 
378
      
 
379
      log.debug("creating control connection: " + newLocator);
 
380
 
 
381
      Socket socket = null;
 
382
      IOException savedException = null;
 
383
      final String finalHost = host;
 
384
      final int finalPort = port;
 
385
      
 
386
      for (int i = 0; i < socketCreationRetries; i++)
 
387
      {
 
388
         try
 
389
         {
 
390
            socket = (Socket) AccessController.doPrivileged( new PrivilegedExceptionAction()
 
391
            {
 
392
               public Object run() throws Exception
 
393
               {
 
394
                  Socket s = null;
 
395
                  if (socketFactory != null)
 
396
                     s = socketFactory.createSocket(finalHost, finalPort);
 
397
                  else
 
398
                     s = new Socket(finalHost, finalPort);
 
399
                  return s;
 
400
               }
 
401
            });
 
402
         }
 
403
         catch (PrivilegedActionException e)
 
404
         {
 
405
            IOException ioe = (IOException) e.getCause();
 
406
            log.debug("Error creating a control socket", ioe);
 
407
            savedException = ioe;
 
408
         }
 
409
         
 
410
         if (socket != null)
 
411
            break;
 
412
         
 
413
         try
 
414
         {
 
415
            Thread.sleep(1000);
 
416
         }
 
417
         catch (InterruptedException e)
 
418
         {
 
419
            log.debug("received interrupt");
 
420
         }
 
421
      }
 
422
 
 
423
      if (socket == null)
 
424
      {
 
425
         log.debug("unable to create control connection after "
 
426
                   + socketCreationRetries + " retries", savedException);
 
427
         throw savedException;
 
428
      }
 
429
      
 
430
      DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
 
431
      if (firstConnection)
 
432
      {
 
433
         dos.write(Bisocket.CREATE_CONTROL_SOCKET);
 
434
      }
 
435
      else
 
436
      {
 
437
         dos.write(Bisocket.RECREATE_CONTROL_SOCKET);
 
438
      }
 
439
      dos.writeUTF(listenerId);
 
440
      
 
441
      Thread thread = new ControlConnectionThread(socket, listenerId);
 
442
      thread.setName("control: " + socket.toString());
 
443
      thread.setDaemon(true);
 
444
 
 
445
      synchronized (controlConnectionThreadMap)
 
446
      {
 
447
         controlConnectionThreadMap.put(listenerId, thread);
 
448
      }
 
449
      
 
450
      Object o = controlConnectionRestartsMap.get(listenerId);
 
451
      if (o != null)
 
452
      {
 
453
         int restarts = ((Integer) o).intValue();
 
454
         if (locatorChanged || restarts > 0)
 
455
         controlConnectionRestartsMap.put(listenerId, new Integer(++restarts));
 
456
      }
 
457
      else
 
458
      {
 
459
         controlConnectionRestartsMap.put(listenerId, new Integer(0));
 
460
      }
 
461
 
 
462
      thread.start();
 
463
      log.debug(this + " created control connection (" + listenerId + "): " + socket.toString());
 
464
   }
 
465
 
 
466
 
 
467
   public void destroyControlConnection(String listenerId)
 
468
   {
 
469
      Thread t = null;
 
470
      
 
471
      synchronized (controlConnectionThreadMap)
 
472
      {
 
473
         t = (Thread) controlConnectionThreadMap.remove(listenerId);
 
474
      }
 
475
      
 
476
      if (t != null)
 
477
      {
 
478
         ((ControlConnectionThread)t).shutdown();
 
479
         log.debug(this + " shutting down control connection: " + listenerId);
 
480
      }
 
481
      else
 
482
      {
 
483
         log.debug("unrecognized listener ID: " + listenerId);
 
484
      }
 
485
      
 
486
      listenerIdToInvokerLocatorMap.remove(listenerId);
 
487
      controlConnectionRestartsMap.remove(listenerId);
 
488
   }
 
489
   
 
490
   
 
491
   public int getControlConnectionRestarts()
 
492
   {
 
493
      return controlConnectionRestarts;
 
494
   }
 
495
   
 
496
   
 
497
   public void setControlConnectionRestarts(int controlConnectionRestarts)
 
498
   {
 
499
      this.controlConnectionRestarts = controlConnectionRestarts;
 
500
   }
 
501
   
 
502
   
 
503
   public int getPingFrequency()
 
504
   {
 
505
      return pingFrequency;
 
506
   }
 
507
 
 
508
 
 
509
   public void setPingFrequency(int pingFrequency)
 
510
   {
 
511
      this.pingFrequency = pingFrequency;
 
512
      pingWindow = pingWindowFactor * pingFrequency;
 
513
   }
 
514
   
 
515
   
 
516
   public int getPingWindowFactor()
 
517
   {
 
518
      return pingWindowFactor;
 
519
   }
 
520
   
 
521
   
 
522
   public void setPingWindowFactor(int pingWindowFactor)
 
523
   {
 
524
      this.pingWindowFactor = pingWindowFactor;
 
525
      pingWindow = pingWindowFactor * pingFrequency;
 
526
   }
 
527
   
 
528
   
 
529
   public int getSecondaryBindPort()
 
530
   {
 
531
      if (secondaryBindPorts.size() == 0 || secondaryBindPorts.size() > 1)
 
532
         return -1;
 
533
      
 
534
      return ((Integer) secondaryBindPorts.get(0)).intValue();
 
535
   }
 
536
   
 
537
   
 
538
   public void setSecondaryBindPort(int secondaryPort)
 
539
   {
 
540
      secondaryBindPorts.clear();
 
541
      secondaryBindPorts.add(new Integer(secondaryPort));
 
542
   }
 
543
   
 
544
   
 
545
   public List getSecondaryBindPorts()
 
546
   {
 
547
      return new ArrayList(secondaryBindPorts);
 
548
   }
 
549
 
 
550
 
 
551
   public void setSecondaryBindPorts(List secondaryBindPorts)
 
552
   {
 
553
      this.secondaryBindPorts = secondaryBindPorts;
 
554
   }
 
555
   
 
556
   
 
557
   public void setSecondaryBindPorts(String secondaryBindPortString)
 
558
   {
 
559
      StringTokenizer tok = new StringTokenizer(secondaryBindPortString, "!");
 
560
      String token = null;
 
561
      while (tok.hasMoreTokens())
 
562
      {
 
563
         try
 
564
         {
 
565
            token = tok.nextToken();
 
566
            secondaryBindPorts.add(Integer.valueOf(token));
 
567
         }
 
568
         catch (NumberFormatException e)
 
569
         {
 
570
            log.warn("Invalid format for " + "\"" + Bisocket.SECONDARY_BIND_PORT + "\": " + token);
 
571
            secondaryBindPorts.add(new Integer(-1));
 
572
         }
 
573
      }
 
574
   }
 
575
 
 
576
 
 
577
   public int getSecondaryConnectPort()
 
578
   {
 
579
      if (secondaryConnectPorts.size() ==  0 || secondaryConnectPorts.size() > 1)
 
580
         return -1;
 
581
      
 
582
      return ((Integer) secondaryConnectPorts.get(0)).intValue();
 
583
   }
 
584
   
 
585
   
 
586
   public void setSecondaryConnectPort(int secondaryConnectPort)
 
587
   {
 
588
      secondaryConnectPorts.clear();
 
589
      secondaryConnectPorts.add(new Integer(secondaryConnectPort));
 
590
   }
 
591
   
 
592
   
 
593
   public List getSecondaryConnectPorts()
 
594
   {
 
595
      return new ArrayList(secondaryConnectPorts);
 
596
   }
 
597
 
 
598
 
 
599
   public void setSecondaryConnectPorts(List secondaryConnectPorts)
 
600
   {
 
601
      this.secondaryConnectPorts = secondaryConnectPorts;
 
602
   }
 
603
   
 
604
   
 
605
   public void setSecondaryConnectPorts(String secondaryConnectPortString)
 
606
   {
 
607
      StringTokenizer tok = new StringTokenizer(secondaryConnectPortString, "!");
 
608
      String token = null;
 
609
      while (tok.hasMoreTokens())
 
610
      {
 
611
         try
 
612
         {
 
613
            token = tok.nextToken();
 
614
            secondaryConnectPorts.add(Integer.valueOf(token));
 
615
         }
 
616
         catch (NumberFormatException e)
 
617
         {
 
618
            log.warn("Invalid format for " + "\"" + Bisocket.SECONDARY_CONNECT_PORT + "\": " + token);
 
619
            secondaryConnectPorts.add(new Integer(-1));
 
620
         }
 
621
      }
 
622
   }
 
623
   
 
624
   
 
625
   public int getSocketCreationRetries()
 
626
   {
 
627
      return socketCreationRetries;
 
628
   }
 
629
   
 
630
   
 
631
   public void setSocketCreationRetries(int socketCreationRetries)
 
632
   {
 
633
      this.socketCreationRetries = socketCreationRetries;
 
634
   }
 
635
   
 
636
   
 
637
   protected void setup() throws Exception
 
638
   {  
 
639
      Object o = configuration.get(Bisocket.IS_CALLBACK_SERVER);
 
640
      if (o != null)
 
641
      {
 
642
         if (o instanceof String)
 
643
            isCallbackServer = Boolean.valueOf((String) o).booleanValue();
 
644
         else if (o instanceof Boolean)
 
645
            isCallbackServer = ((Boolean) o).booleanValue();
 
646
         else
 
647
            log.error("unrecognized value for configuration key \"" +
 
648
                  Bisocket.IS_CALLBACK_SERVER + "\": " + o);
 
649
      }
 
650
      
 
651
      super.setup();
 
652
      
 
653
      o = configuration.get(Bisocket.PING_FREQUENCY);
 
654
      if (o instanceof String && ((String) o).length() > 0)
 
655
      {
 
656
            try
 
657
            {
 
658
               pingFrequency = Integer.valueOf(((String) o)).intValue();
 
659
               log.debug(this + " setting pingFrequency to " + pingFrequency);
 
660
            }
 
661
            catch (NumberFormatException e)
 
662
            {
 
663
               log.warn("Invalid format for " + "\"" + Bisocket.PING_FREQUENCY + "\": " + o);
 
664
            }
 
665
      }
 
666
      else if (o != null)
 
667
      {
 
668
         log.warn("\"" + Bisocket.PING_FREQUENCY + "\" must be specified as a String");
 
669
      }
 
670
      
 
671
      o = configuration.get(Bisocket.PING_WINDOW_FACTOR);
 
672
      if (o instanceof String && ((String) o).length() > 0)
 
673
      {
 
674
            try
 
675
            {
 
676
               pingWindowFactor = Integer.valueOf(((String) o)).intValue();
 
677
               log.debug(this + " setting pingWindowFactor to " + pingWindowFactor);
 
678
            }
 
679
            catch (NumberFormatException e)
 
680
            {
 
681
               log.warn("Invalid format for " + "\"" + Bisocket.PING_WINDOW_FACTOR + "\": " + o);
 
682
            }
 
683
      }
 
684
      else if (o != null)
 
685
      {
 
686
         log.warn("\"" + Bisocket.PING_WINDOW_FACTOR + "\" must be specified as a String");
 
687
      }
 
688
 
 
689
      pingWindow = pingWindowFactor * pingFrequency;
 
690
 
 
691
      o = configuration.get(Bisocket.SECONDARY_BIND_PORTS);
 
692
      if (o instanceof String && ((String) o).length() > 0)
 
693
      {
 
694
         setSecondaryBindPorts((String) o);
 
695
      }
 
696
      else if (o instanceof List)
 
697
      {
 
698
         setSecondaryBindPorts((List) o);
 
699
      }
 
700
      else if (o != null)
 
701
      {
 
702
         log.warn("\"" + Bisocket.SECONDARY_BIND_PORTS + "\" must be specified as a String or a List");
 
703
      }
 
704
      
 
705
      o = configuration.get(Bisocket.SECONDARY_CONNECT_PORTS);
 
706
      if (o instanceof String && ((String) o).length() > 0)
 
707
      {
 
708
         setSecondaryConnectPorts((String) o);
 
709
      }
 
710
      else if (o instanceof List)
 
711
      {
 
712
         setSecondaryConnectPorts((List) o);
 
713
      }
 
714
      else if (o != null)
 
715
      {
 
716
         log.warn("\"" + Bisocket.SECONDARY_CONNECT_PORTS + "\" must be specified as a String or a List");
 
717
      }
 
718
 
 
719
      if (secondaryBindPorts.isEmpty())
 
720
      {
 
721
         for (int i = 0; i < homes.size(); i++)
 
722
            secondaryBindPorts.add(new Integer(-1));
 
723
      }
 
724
      
 
725
      if (secondaryConnectPorts.isEmpty())
 
726
      {
 
727
         secondaryConnectPorts = new ArrayList(secondaryBindPorts);
 
728
      }
 
729
      
 
730
      if (isCallbackServer)
 
731
      {
 
732
         socketFactory = createSocketFactory(configuration);
 
733
      }
 
734
   }
 
735
 
 
736
 
 
737
   protected void cleanup()
 
738
   {
 
739
      synchronized (controlConnectionThreadMap)
 
740
      {
 
741
         Iterator it = controlConnectionThreadMap.values().iterator();
 
742
         while (it.hasNext())
 
743
         {
 
744
            ControlConnectionThread t = (ControlConnectionThread) it.next();
 
745
            it.remove();
 
746
            t.shutdown();
 
747
         }
 
748
      }
 
749
 
 
750
      super.cleanup();
 
751
 
 
752
      if (controlMonitorTimerTask != null)
 
753
         controlMonitorTimerTask.shutdown();
 
754
      
 
755
      Iterator it = secondaryServerSocketThreads.iterator();
 
756
      while (it.hasNext())
 
757
      {
 
758
         SecondaryServerSocketThread t = (SecondaryServerSocketThread) it.next();
 
759
         t.shutdown();
 
760
      }
 
761
      
 
762
      it = secondaryServerSockets.iterator();
 
763
      while (it.hasNext())
 
764
      {
 
765
         try
 
766
         {
 
767
            ServerSocket ss = (ServerSocket) it.next();
 
768
            ss.close();
 
769
         }
 
770
         catch (IOException e)
 
771
         {
 
772
            log.info("Error closing secondary server socket: " + e.getMessage());
 
773
         }
 
774
      }
 
775
      
 
776
      secondaryBindPorts.clear();
 
777
      secondaryConnectPorts.clear();
 
778
   }
 
779
 
 
780
 
 
781
   protected InvokerLocator getSecondaryLocator()
 
782
   {
 
783
      return secondaryLocator;
 
784
   }
 
785
 
 
786
 
 
787
   protected void checkSocketFactoryWrapper() throws IOException
 
788
   {
 
789
 
 
790
      Object o = configuration.get(Remoting.SOCKET_CREATION_SERVER_LISTENER);
 
791
      if (o != null)
 
792
      {
 
793
         if (o instanceof SocketCreationListener)
 
794
         {
 
795
            SocketCreationListener listener = (SocketCreationListener) o;
 
796
            if (socketFactory instanceof CreationListenerSocketFactory)
 
797
            {
 
798
               CreationListenerSocketFactory clsf = (CreationListenerSocketFactory) socketFactory;
 
799
               clsf.setListener(listener);
 
800
            }
 
801
            else
 
802
            {
 
803
               socketFactory = new CreationListenerSocketFactory(socketFactory, listener);
 
804
            }
 
805
         }
 
806
         else
 
807
         {
 
808
            log.error("socket creation listener of invalid type: " + o);
 
809
         }
 
810
      }
 
811
      else
 
812
      {
 
813
         if (socketFactory instanceof CreationListenerSocketFactory)
 
814
         {
 
815
            CreationListenerSocketFactory clsf = (CreationListenerSocketFactory) socketFactory;
 
816
            socketFactory = clsf.getFactory();
 
817
         }
 
818
      }
 
819
   }
 
820
 
 
821
 
 
822
   protected ServerSocket checkSecondaryServerSocketWrapper(ServerSocket secondaryServerSocket) throws IOException
 
823
   {
 
824
      Object o = configuration.get(Remoting.SOCKET_CREATION_CLIENT_LISTENER);
 
825
      if (o != null)
 
826
      {
 
827
         if (o instanceof SocketCreationListener)
 
828
         {
 
829
            SocketCreationListener listener = (SocketCreationListener) o;
 
830
            if (secondaryServerSocket instanceof CreationListenerServerSocket)
 
831
            {
 
832
               CreationListenerServerSocket clss = (CreationListenerServerSocket) secondaryServerSocket;
 
833
               clss.setListener(listener);
 
834
            }
 
835
            else
 
836
            {
 
837
               secondaryServerSocket = new CreationListenerServerSocket(secondaryServerSocket, listener);
 
838
            }
 
839
         }
 
840
         else
 
841
         {
 
842
            log.error("socket creation listener of invalid type: " + o);
 
843
         }
 
844
      }
 
845
      else
 
846
      {
 
847
         if (secondaryServerSocket instanceof CreationListenerServerSocket)
 
848
         {
 
849
            CreationListenerServerSocket clss = (CreationListenerServerSocket) secondaryServerSocket;
 
850
            secondaryServerSocket = clss.getServerSocket();
 
851
         }
 
852
      }
 
853
      
 
854
      return secondaryServerSocket;
 
855
   }
 
856
 
 
857
 
 
858
   protected Object handleInternalInvocation(InternalInvocation ii,
 
859
                                             InvocationRequest ir,
 
860
                                             ServerInvocationHandler handler)
 
861
   throws Throwable
 
862
   {
 
863
      if(Bisocket.GET_SECONDARY_INVOKER_LOCATOR.equals(ii.getMethodName()))
 
864
      {
 
865
         return secondaryLocator;
 
866
      }
 
867
 
 
868
      Object response = super.handleInternalInvocation(ii, ir, handler);
 
869
 
 
870
      if(InternalInvocation.ADDCLIENTLISTENER.equals(ii.getMethodName()))
 
871
      {
 
872
         Map metadata = ir.getRequestPayload();
 
873
         if(metadata != null)
 
874
         {
 
875
            String listenerId = (String) metadata.get(Client.LISTENER_ID_KEY);
 
876
            if (listenerId != null)
 
877
            {
 
878
               listenerIdToServerInvokerMap.put(listenerId, this);
 
879
            }
 
880
         }
 
881
      }
 
882
      else if(InternalInvocation.REMOVECLIENTLISTENER.equals(ii.getMethodName()))
 
883
      {
 
884
         Map metadata = ir.getRequestPayload();
 
885
         if(metadata != null)
 
886
         {
 
887
            String listenerId = (String) metadata.get(Client.LISTENER_ID_KEY);
 
888
            if (listenerId != null)
 
889
            {
 
890
               listenerIdToServerInvokerMap.remove(listenerId);
 
891
               BisocketClientInvoker.removeBisocketClientInvoker(listenerId);
 
892
               destroyControlConnection(listenerId);
 
893
            }
 
894
         }
 
895
      }
 
896
 
 
897
      return response;
 
898
   }
 
899
 
 
900
 
 
901
   class ControlConnectionThread extends Thread
 
902
   {
 
903
      private static final int MAX_INITIAL_ATTEMPTS = 5;
 
904
      private Socket controlSocket;
 
905
      private String listenerId;
 
906
      private DataInputStream dis;
 
907
      private boolean running;
 
908
      private int errorCount;
 
909
      private long lastPing = -1;
 
910
      private int initialAttempts;
 
911
 
 
912
      ControlConnectionThread(Socket socket, String listenerId) throws IOException
 
913
      {
 
914
         controlSocket = socket;
 
915
         this.listenerId = listenerId;
 
916
         dis = new DataInputStream(socket.getInputStream());
 
917
      }
 
918
 
 
919
      void shutdown()
 
920
      {
 
921
         running = false;
 
922
 
 
923
         try
 
924
         {
 
925
            controlSocket.close();
 
926
         }
 
927
         catch (IOException e)
 
928
         {
 
929
            log.warn("unable to close controlSocket");
 
930
         }
 
931
         interrupt();
 
932
      }
 
933
 
 
934
      boolean checkConnection()
 
935
      {
 
936
         if (lastPing < 0 && initialAttempts++ < MAX_INITIAL_ATTEMPTS)
 
937
         {
 
938
            return true;
 
939
         }
 
940
         else if (lastPing < 0)
 
941
         {
 
942
            return false;
 
943
         }
 
944
         
 
945
         long currentTime = System.currentTimeMillis();
 
946
 
 
947
         if (log.isTraceEnabled())
 
948
         {
 
949
            log.trace("elapsed: " + (currentTime - lastPing));
 
950
         }
 
951
         return (currentTime - lastPing <= pingWindow);
 
952
      }
 
953
 
 
954
      String getListenerId()
 
955
      {
 
956
         return listenerId;
 
957
      }
 
958
 
 
959
      public void run()
 
960
      {
 
961
         running = true;
 
962
         while (running)
 
963
         {
 
964
            Socket socket = null;
 
965
 
 
966
            try
 
967
            {
 
968
               int action = dis.read();
 
969
               lastPing = System.currentTimeMillis();
 
970
 
 
971
               switch (action)
 
972
               {
 
973
                  case Bisocket.CREATE_ORDINARY_SOCKET:
 
974
                     InvokerLocator locator = (InvokerLocator) listenerIdToInvokerLocatorMap.get(listenerId);
 
975
                     
 
976
                     IOException savedException = null;
 
977
                     final String finalHost = locator.getHost();
 
978
                     final int finalPort = locator.getPort();
 
979
 
 
980
                     for (int i = 0; i < socketCreationRetries; i++)
 
981
                     {
 
982
                        try
 
983
                        {
 
984
                           socket = (Socket) AccessController.doPrivileged( new PrivilegedExceptionAction()
 
985
                           {
 
986
                              public Object run() throws Exception
 
987
                              {
 
988
                                 Socket s = null;
 
989
                                 if (socketFactory != null)
 
990
                                    s = socketFactory.createSocket(finalHost, finalPort);
 
991
                                 else
 
992
                                    s = new Socket(finalHost, finalPort);
 
993
                                 return s;
 
994
                              }
 
995
                           });
 
996
                        }
 
997
                        catch (PrivilegedActionException e)
 
998
                        {
 
999
                           IOException ioe = (IOException) e.getCause();
 
1000
                           log.debug("Error creating a socket", ioe);
 
1001
                           savedException = ioe;
 
1002
                        }
 
1003
                        
 
1004
                        if (socket != null)
 
1005
                           break;
 
1006
                        
 
1007
                        try
 
1008
                        {
 
1009
                           Thread.sleep(1000);
 
1010
                        }
 
1011
                        catch (InterruptedException e)
 
1012
                        {
 
1013
                           if (running)
 
1014
                           {
 
1015
                              log.debug("received unexpected interrupt");
 
1016
                              continue;
 
1017
                           }
 
1018
                           else
 
1019
                           {
 
1020
                              return;
 
1021
                           }
 
1022
                        }
 
1023
                     }
 
1024
                     
 
1025
                     if (socket == null)
 
1026
                     {
 
1027
                        log.error("Unable to create socket after " + socketCreationRetries 
 
1028
                                  + " retries", savedException);
 
1029
                        continue;
 
1030
                     }
 
1031
                     
 
1032
                     DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
 
1033
                     dos.write(Bisocket.CREATE_ORDINARY_SOCKET);
 
1034
                     dos.writeUTF(listenerId);
 
1035
                     break;
 
1036
 
 
1037
                  case Bisocket.PING:
 
1038
                     continue;
 
1039
 
 
1040
                  case -1:
 
1041
                     shutdown();
 
1042
                     return;
 
1043
 
 
1044
                  default:
 
1045
                     log.error("unrecognized action on ControlConnectionThread (" +
 
1046
                               listenerId + "): " +  action);
 
1047
                     continue;
 
1048
               }
 
1049
            }
 
1050
            catch (IOException e)
 
1051
            {
 
1052
               if (running)
 
1053
               {
 
1054
                  if ("Socket closed".equalsIgnoreCase(e.getMessage()) ||
 
1055
                      "Socket is closed".equalsIgnoreCase(e.getMessage()) ||
 
1056
                      "Connection reset".equalsIgnoreCase(e.getMessage()))
 
1057
                  {
 
1058
                     shutdown();
 
1059
                     return;
 
1060
                  }
 
1061
                  log.error("Unable to process control connection: " + e.getMessage(), e);
 
1062
                  if (++errorCount > 5)
 
1063
                  {
 
1064
                     shutdown();
 
1065
                     return;
 
1066
                  }
 
1067
                  continue;
 
1068
               }
 
1069
 
 
1070
               return;
 
1071
            }
 
1072
 
 
1073
            if (!running)
 
1074
            {
 
1075
               return;
 
1076
            }
 
1077
            
 
1078
            try
 
1079
            {
 
1080
               processInvocation(socket);
 
1081
            }
 
1082
            catch (Exception e)
 
1083
            {
 
1084
               log.error("Unable to create new ServerThread: " + e.getMessage(), e);
 
1085
            }
 
1086
         }
 
1087
      }
 
1088
   }
 
1089
 
 
1090
 
 
1091
   class SecondaryServerSocketThread extends Thread
 
1092
   {
 
1093
      private ServerSocket secondaryServerSocket;
 
1094
      boolean running = true;
 
1095
 
 
1096
      SecondaryServerSocketThread(ServerSocket secondaryServerSocket) throws IOException
 
1097
      {
 
1098
         this.secondaryServerSocket = secondaryServerSocket;
 
1099
      }
 
1100
 
 
1101
      void shutdown()
 
1102
      {
 
1103
         running = false;
 
1104
         interrupt();
 
1105
      }
 
1106
 
 
1107
      public void run()
 
1108
      {
 
1109
         while (running)
 
1110
         {
 
1111
            try
 
1112
            {
 
1113
               Socket socket = null;
 
1114
               try
 
1115
               {
 
1116
                   socket = (Socket)AccessController.doPrivileged( new PrivilegedExceptionAction()
 
1117
                   {
 
1118
                      public Object run() throws Exception
 
1119
                      {
 
1120
                          return secondaryServerSocket.accept();
 
1121
                      }
 
1122
                   });
 
1123
               }
 
1124
               catch (PrivilegedActionException e)
 
1125
               {
 
1126
                   throw (IOException) e.getCause();
 
1127
               }
 
1128
               
 
1129
               if (log.isTraceEnabled()) log.trace("accepted: " + socket);
 
1130
               DataInputStream dis = new DataInputStream(socket.getInputStream());
 
1131
               int action = dis.read();
 
1132
               String listenerId = dis.readUTF();
 
1133
 
 
1134
               switch (action)
 
1135
               {
 
1136
                  case Bisocket.CREATE_CONTROL_SOCKET:
 
1137
                     BisocketClientInvoker.transferSocket(listenerId, socket, true);
 
1138
                     if (log.isTraceEnabled()) 
 
1139
                        log.trace("SecondaryServerSocketThread: created control socket: (" + socket + ")"+ listenerId);
 
1140
                     break;
 
1141
                     
 
1142
                  case Bisocket.RECREATE_CONTROL_SOCKET:
 
1143
                     BisocketClientInvoker invoker =  BisocketClientInvoker.getBisocketCallbackClientInvoker(listenerId);
 
1144
                     if (invoker == null)
 
1145
                     {
 
1146
                        log.debug("received new control socket for unrecognized listenerId: " + listenerId);
 
1147
                     }
 
1148
                     else
 
1149
                     {
 
1150
                        invoker.replaceControlSocket(socket);
 
1151
                        if (log.isTraceEnabled())
 
1152
                           log.trace("SecondaryServerSocketThread: recreated control socket: " + listenerId);
 
1153
                     }
 
1154
                     break;
 
1155
 
 
1156
                  case Bisocket.CREATE_ORDINARY_SOCKET:
 
1157
                     BisocketClientInvoker.transferSocket(listenerId, socket, false);
 
1158
                     if (log.isTraceEnabled())
 
1159
                        log.trace("SecondaryServerSocketThread: transferred socket: " + listenerId);
 
1160
                     break;
 
1161
 
 
1162
                  default:
 
1163
                     log.error("unrecognized action on SecondaryServerSocketThread: " + action);
 
1164
               }
 
1165
            }
 
1166
            catch (IOException e)
 
1167
            {
 
1168
               if (running)
 
1169
                  log.error("Failed to accept socket connection", e);
 
1170
               else
 
1171
                  return;
 
1172
 
 
1173
            }
 
1174
         }
 
1175
      }
 
1176
 
 
1177
      ServerSocket getServerSocket()
 
1178
      {
 
1179
         return secondaryServerSocket;
 
1180
      }
 
1181
   }
 
1182
 
 
1183
 
 
1184
   static class ControlMonitorTimerTask extends TimerTask
 
1185
   {
 
1186
      private boolean running = true;
 
1187
      private BisocketServerInvoker invoker;
 
1188
      private Map listenerIdToInvokerLocatorMap;
 
1189
      private Map controlConnectionThreadMap;
 
1190
      private Map controlConnectionRestartsMap;
 
1191
      private int controlConnectionRestarts;
 
1192
      
 
1193
      ControlMonitorTimerTask(BisocketServerInvoker invoker)
 
1194
      {
 
1195
         this.invoker = invoker;
 
1196
         listenerIdToInvokerLocatorMap = invoker.listenerIdToInvokerLocatorMap;
 
1197
         controlConnectionThreadMap = invoker.controlConnectionThreadMap;
 
1198
         controlConnectionRestartsMap = invoker.controlConnectionRestartsMap;
 
1199
         controlConnectionRestarts = invoker.controlConnectionRestarts;
 
1200
      }
 
1201
 
 
1202
      synchronized void shutdown()
 
1203
      {
 
1204
         // Note that there is a race between shutdown() and run().  But if run()
 
1205
         // were synchronized, then shutdown() could be held up waiting on network
 
1206
         // i/o, including invocations on a server that no longer is accessible.
 
1207
         // So only minimal synchronization is imposed on run(), enough to avoid
 
1208
         // NullPointerExceptions.
 
1209
         
 
1210
         running = false;
 
1211
         invoker = null;
 
1212
         listenerIdToInvokerLocatorMap = null;
 
1213
         controlConnectionThreadMap = null;
 
1214
         cancel();
 
1215
 
 
1216
         try
 
1217
         {
 
1218
            Method purge = getDeclaredMethod(Timer.class, "purge", new Class[]{});
 
1219
            purge.invoke(timer, new Object[]{});
 
1220
         }
 
1221
         catch (Exception e)
 
1222
         {
 
1223
            log.debug("running with jdk 1.4: unable to purge Timer");
 
1224
         }
 
1225
      }
 
1226
 
 
1227
      public void run()
 
1228
      {
 
1229
         if (!running)
 
1230
            return;
 
1231
         
 
1232
         if (log.isTraceEnabled())
 
1233
            log.trace("checking connections");
 
1234
 
 
1235
         Collection controlConnectionThreads = null;
 
1236
         synchronized (this)
 
1237
         {
 
1238
            if (!running)
 
1239
               return;
 
1240
            
 
1241
            controlConnectionThreads = new HashSet(controlConnectionThreadMap.values());
 
1242
         }
 
1243
         
 
1244
         Iterator it = controlConnectionThreads.iterator();
 
1245
         while (it.hasNext())
 
1246
         {
 
1247
            final ControlConnectionThread t = (ControlConnectionThread) it.next();
 
1248
            final String listenerId = t.getListenerId();
 
1249
            final Object locator;
 
1250
            
 
1251
            synchronized (this)
 
1252
            {
 
1253
               if (!running)
 
1254
                  return;
 
1255
                  
 
1256
               locator = listenerIdToInvokerLocatorMap.get(listenerId);
 
1257
            }
 
1258
            
 
1259
            if (!t.checkConnection())
 
1260
            {
 
1261
               t.shutdown();
 
1262
               
 
1263
               synchronized (this)
 
1264
               {
 
1265
                  if (!running)
 
1266
                     return;
 
1267
                  
 
1268
                  controlConnectionThreadMap.remove(listenerId);
 
1269
                  Object o = controlConnectionRestartsMap.get(listenerId);
 
1270
                  int restarts = ((Integer)o).intValue();
 
1271
                  
 
1272
                  if (restarts + 1 > controlConnectionRestarts)
 
1273
                  {
 
1274
                     log.warn(this + ": detected failure on control connection " + t);
 
1275
                     log.warn("Control connection " + listenerId + " has been recreated " + restarts + " times.");
 
1276
                     log.warn("Assuming it is a connection to an old server, and will not restart");
 
1277
                     controlConnectionRestartsMap.remove(listenerId);
 
1278
                     continue;
 
1279
                  }
 
1280
                  
 
1281
                  log.warn(this + ": detected failure on control connection " + t + 
 
1282
                                  " (" + listenerId + 
 
1283
                                  ": requesting new control connection");
 
1284
               }
 
1285
               
 
1286
               Thread t2 = new Thread()
 
1287
               {
 
1288
                  public void run()
 
1289
                  {
 
1290
                     if (!running)
 
1291
                        return;
 
1292
                     
 
1293
                     try
 
1294
                     {
 
1295
                        invoker.createControlConnection(listenerId, false);
 
1296
                     }
 
1297
                     catch (ClientUnavailableException e)
 
1298
                     {
 
1299
                        log.debug("Unable to recreate control connection: " + locator, e);
 
1300
                     }
 
1301
                     catch (IOException e)
 
1302
                     {
 
1303
                        if (running)
 
1304
                           log.error("Unable to recreate control connection: " + locator, e);
 
1305
                        else
 
1306
                           log.debug("Unable to recreate control connection: " + locator, e);
 
1307
                     }
 
1308
                  }
 
1309
               };
 
1310
               t2.setName("controlConnectionRecreate:" + t.getName());
 
1311
               t2.start();
 
1312
            }
 
1313
         }
 
1314
      }
 
1315
   }
 
1316
   
 
1317
   static class ClientUnavailableException extends IOException
 
1318
   {
 
1319
      private static final long serialVersionUID = 2846502029152028732L;
 
1320
   }
 
1321
   
 
1322
   static private Method getDeclaredMethod(final Class c, final String name, final Class[] parameterTypes)
 
1323
   throws NoSuchMethodException
 
1324
   {
 
1325
      if (SecurityUtility.skipAccessControl())
 
1326
      {
 
1327
         Method m = c.getDeclaredMethod(name, parameterTypes);
 
1328
         m.setAccessible(true);
 
1329
         return m;
 
1330
      }
 
1331
 
 
1332
      try
 
1333
      {
 
1334
         return (Method) AccessController.doPrivileged( new PrivilegedExceptionAction()
 
1335
         {
 
1336
            public Object run() throws NoSuchMethodException
 
1337
            {
 
1338
               Method m = c.getDeclaredMethod(name, parameterTypes);
 
1339
               m.setAccessible(true);
 
1340
               return m;
 
1341
            }
 
1342
         });
 
1343
      }
 
1344
      catch (PrivilegedActionException e)
 
1345
      {
 
1346
         throw (NoSuchMethodException) e.getCause();
 
1347
      }
 
1348
   }
 
1349
}
 
 
b'\\ No newline at end of file'