Find Usages Diff Raw Download HTML Widget
Start line:  
End line:  

Snippet Preview

Snippet HTML Code

Stack Overflow Questions 
1
   package org.jboss.remoting.transport.socket;
2
   
3
   import  org.jboss.logging.Logger;
21
  import  org.jboss.util.propertyeditor.PropertyEditors;
22
  
30
  import java.net.Socket;
38
  import java.util.HashMap;
41
  import java.util.List;
42
  import java.util.Map;
45
  
46
  import  EDU.oswego.cs.dl.util.concurrent.Semaphore;

SocketClientInvoker uses Sockets to remotely connect to the a remote ServerInvoker, which must be a SocketServerInvoker.

Author(s):
Jeff Haynie
Tom Elrod
Ovidiu Feodorov
Version:
$Revision: 6179 $
57
  
58
  public class MicroSocketClientInvoker extends RemoteClientInvoker
59
  {
60
     // Constants ------------------------------------------------------------------------------------
61
  
62
     private static final Logger log = Logger.getLogger(MicroSocketClientInvoker.class);

   
Can be either true or false and will indicate if client socket should have TCP_NODELAY turned on or off. TCP_NODELAY is for a specific purpose; to disable the Nagle buffering algorithm. It should only be set for applications that send frequent small bursts of information without getting an immediate response; where timely delivery of data is required (the canonical example is mouse movements). The default is false. 
70
  
71
     public static final String TCP_NODELAY_FLAG = "enableTcpNoDelay";

   
The client side maximum number of threads. The default is MAX_POOL_SIZE. 
75
  
76
     public static final String MAX_POOL_SIZE_FLAG = "clientMaxPoolSize";

   
Specifies the fully qualified class name for the custom SocketWrapper implementation to use on the client. Note, will need to make sure this is marked as a client parameter (using the 'isParam' attribute). Making this change will not affect the marshaller/unmarshaller that is used, which may also be a requirement. 
83
  
84
     public static final String CLIENT_SOCKET_CLASS_FLAG = "clientSocketClass";
   
   
Key for setting timeout used by OnewayConnectionTask
86
  
87
     public static final String ONEWAY_CONNECTION_TIMEOUT = "onewayConnectionTimeout";
   
   
Key to determine if client side oneway invocations should wait to read version. See JBREM-706. 
91
  
92
     public static final String USE_ONEWAY_CONNECTION_TIMEOUT = "useOnewayConnectionTimeout";

   
Key for setting time to wait to get permission to get a connection
94
  
95
     public static final String CONNECTION_WAIT = "connectionWait";
   
   
Key for setting socket write timeout
97
  
98
     public static final String WRITE_TIMEOUT = "writeTimeout";
   
   
Default value for enable TCP nodelay. Value is false. 
102
 
103
    public static final boolean TCP_NODELAY_DEFAULT = false;

   
Default maximum number of times a invocation will be made when it gets a SocketException. Default is 3. 
108
 
109
    public static final int MAX_CALL_RETRIES = 3;

   
Default maximum number of socket connections allowed at any point in time. Default is 50. 
113
 
114
    public static final int MAX_POOL_SIZE = 50;

   
Default timeout value used by OnewayConnectionTask. Value is 2 seconds.
116
 
117
    public static final int ONEWAY_CONNECTION_TIMEOUT_DEFAULT = 2000;
   
   
Default time to wait to get permission to get a connection
119
 
120
    public static final int CONNECTION_WAIT_DEFAULT = 30000;
121
 
122
    // Static ---------------------------------------------------------------------------------------
123
 
124
    private static boolean trace = .isTraceEnabled();

   
Used for debugging (tracing) connections leaks 
128
 
129
    static int counter = 0;
130
 
131
    protected static final Map connectionPools = new HashMap();
132
    
133
    protected static final Map semaphores = new HashMap();
134
 
135
    // Performance measurements
136
    public static long getSocketTime = 0;
137
    public static long readTime = 0;
138
    public static long writeTime = 0;
139
    public static long serializeTime = 0;
140
    public static long deserializeTime = 0;
141
    
142
    private static final String patternString = "^.*(?:connection.*reset|connection.*closed|connection.*abort|broken.*pipe|connection.*shutdown).*$";
143
    private static final Pattern RETRIABLE_ERROR_MESSAGE = Pattern.compile(, .);
   
   
Close all sockets in a specific pool. 
147
 
148
    public static void clearPool(LinkedList thepool)
149
    {
150
       try
151
       {
152
          if (thepool == null)
153
          {
154
             return;
155
          }
156
          synchronized (thepool)
157
          {
158
             int size = thepool.size();
159
             for (int i = 0; i < size; i++)
160
             {
161
                SocketWrapper socketWrapper = (SocketWrapper)thepool.removeFirst();
162
                try
163
                {
164
                   socketWrapper.close();
165
                   socketWrapper = null;
166
                }
167
                catch (Exception ignored)
168
                {
169
                }
170
             }
171
          }
172
       }
173
       catch (Exception ex)
174
       {
175
          .debug("Failure", ex);
176
       }
177
    }

   
Close all sockets in all pools. 
181
 
182
    public static void clearPools()
183
    {
184
       synchronized ()
185
       {
186
          for(Iterator i = .keySet().iterator(); i.hasNext();)
187
          {
188
             ServerAddress sa = (ServerAddress) i.next();
189
 
190
             if () { .trace("clearing pool for " + sa); }
191
             clearPool((LinkedList) .get(sa));
192
             i.remove();
193
          }
194
          .clear();
195
       }
196
    }
197
 
198
    // Attributes -----------------------------------------------------------------------------------
199
 
201
    private boolean reuseAddress;
202
 
203
    protected InetAddress addr;
204
    protected int port;
205
 
206
    // flag being set on true by a disconnect request. If trying to create a connection goes on in a
207
    // loop and a disconnect request arrives, this flag will be used to sent this information into
208
    // the connect loop
209
    // private volatile boolean bailOut;
210
 
   
Indicates if will check the socket connection when getting from pool by sending byte over the connection to validate is still good. 
214
 
215
    protected boolean shouldCheckConnection;

   
If the TcpNoDelay option should be used on the socket. 
219
 
220
    protected boolean enableTcpNoDelay;
221
 
222
    protected String clientSocketClassName;
223
    protected Class clientSocketClass;
224
    protected int numberOfCallRetries;
225
    protected int maxPoolSize;
226
    protected int onewayConnectionTimeout;
227
    protected boolean useOnewayConnectionTimeout = true;
228
    protected int connectionWait = ;

   
Pool for this invoker. This is shared between all instances of proxies attached to a specific invoker. 
233
 
234
    protected LinkedList pool;
235
    
236
    //Semaphore is also shared between all proxies - must 1-1 correspondence between pool and semaphore
237
    protected Semaphore semaphore;
   

   
connection information 
242
 
243
    protected ServerAddress address;
244
    protected Home home;
   
   
Socket configuration parameters. 
248
 
249
    protected boolean keepAlive;
250
    protected boolean keepAliveSet;
251
    protected boolean oOBInline;
252
    protected boolean oOBInlineSet;
253
    protected int receiveBufferSize = - 1;
254
    protected int sendBufferSize = -1;
255
    protected boolean soLinger;
256
    protected boolean soLingerSet;
257
    protected int soLingerDuration = -1;
258
    protected int trafficClass = -1;
   
   
If true, an IOException with message such as "Connection reset by peer: socket write error" will be treated like a SocketException. 
263
 
264
    protected boolean generalizeSocketException;
265
    
266
    protected int writeTimeout = -1;
267
 
268
    // Constructors ---------------------------------------------------------------------------------
269
 
271
    {
272
       this(locator, null);
273
    }
274
 
275
    public MicroSocketClientInvoker(InvokerLocator locator, Map configuration)
276
    {
277
       super(locator, configuration);
278
 
279
        = null;
280
        = true;
281
        = false;
284
        = null;
286
        = null;
287
        = ;
289
 
290
       try
291
       {
292
          setup();
293
       }
294
       catch (Exception ex)
295
       {
296
          .debug("Error setting up " + this, ex);
297
          throw new RuntimeException(ex.getMessage(), ex);
298
       }
299
 
300
       .debug(this + " constructed");
301
    }
302
 
303
    // Public ---------------------------------------------------------------------------------------
304
 
   
Indicates if will check socket connection when returning from pool by sending byte to the server. Default value will be false. 
308
 
309
    public boolean checkingConnection()
310
    {
311
       return ;
312
    }

   
Returns if newly created sockets will have SO_REUSEADDR enabled. Default is for this to be true. 
317
 
318
    public boolean getReuseAddress()
319
    {
320
       return ;
321
    }

   
Sets if newly created socket should have SO_REUSEADDR enable. Default is true. 
325
 
326
    public void setReuseAddress(boolean reuse)
327
    {
328
        = reuse;
329
    }
330
 
331
    public boolean isKeepAlive()
332
    {
333
       return ;
334
    }
335
 
336
    public void setKeepAlive(boolean keepAlive)
337
    {
338
       this. = keepAlive;
339
        = true;
340
    }
341
 
342
    public boolean isOOBInline()
343
    {
344
       return ;
345
    }
346
 
347
    public void setOOBInline(boolean inline)
348
    {
349
        = inline;
350
        = true;
351
    }
352
 
353
    public int getReceiveBufferSize()
354
    {
355
       return ;
356
    }
357
 
358
    public void setReceiveBufferSize(int receiveBufferSize)
359
    {
360
       this. = receiveBufferSize;
361
    }
362
 
363
    public int getSendBufferSize()
364
    {
365
       return ;
366
    }
367
 
368
    public void setSendBufferSize(int sendBufferSize)
369
    {
370
       this. = sendBufferSize;
371
    }
372
 
373
    public boolean isSoLinger()
374
    {
375
       return ;
376
    }
377
    
378
    public int getSoLingerDuration()
379
    {
380
       return ;
381
    }
382
 
383
    public void setSoLinger(boolean soLinger)
384
    {
385
       this. = soLinger;
386
        = true;
387
    }
388
 
389
    public void setSoLingerDuration(int soLingerDuration)
390
    {
391
       this. = soLingerDuration;
392
    }
393
 
394
    public int getTrafficClass()
395
    {
396
       return ;
397
    }
398
 
399
    public void setTrafficClass(int trafficClass)
400
    {
401
       this. = trafficClass;
402
    }
403
 
404
    public int getWriteTimeout()
405
    {
406
       return ;
407
    }
408
 
409
    public void setWriteTimeout(int writeTimeout)
410
    {
411
       this. = writeTimeout;
412
    }
413
 
414
    public boolean isGeneralizeSocketException()
415
    {
416
       return ;
417
    }
418
 
419
    public void setGeneralizeSocketException(boolean generalizeSocketException)
420
    {
421
       this. = generalizeSocketException;
422
    }
423
 
424
    public synchronized void disconnect()
425
    {
426
       .debug(this + " disconnecting ...");
427
 //      bailOut = true;
428
       super.disconnect();
429
    }
430
 
431
    public void flushConnectionPool()
432
    {
433
       synchronized ()
434
       {
435
          while ( != null && .size() > 0)
436
          {
437
             SocketWrapper socketWrapper = (SocketWrapper).removeFirst();
438
             try
439
             {
440
                socketWrapper.close();
441
             }
442
             catch (IOException e)
443
             {
444
                .debug("Failed to close socket wrapper", e);
445
             }
446
          }
447
       }
448
    }
449
    
450
    public int getConnectionWait()
451
    {
452
       return ;
453
    }
454
 
455
    public void setConnectionWait(int connectionWait)
456
    {
457
       this. = connectionWait;
458
    }
459
 
460
    public Home getHomeInUse()
461
    {
462
       return ;
463
    }

   
Sets the number of times an invocation will retry based on getting SocketException. 
467
 
468
    public void setNumberOfCallRetries(int numberOfCallRetries)
469
    {
470
       if (numberOfCallRetries < 1)
471
       {
472
          this. = ;
473
       }
474
       else
475
       {
476
          this. = numberOfCallRetries;
477
       }
478
    }
479
 
480
    public int getNumberOfCallRetries()
481
    {
482
       return ;
483
    }

   
Sets the number of retries to get a socket connection.

Parameters:
numberOfRetries Must be a number greater than 0.
489
 
490
    public void setNumberOfRetries(int numberOfRetries)
491
    {
492
       .warn("numberOfRetries is no longer used");
493
    }
494
 
495
    public int getNumberOfRetries()
496
    {
497
       .warn("numberOfRetries is no longer used");
498
       return -1;
499
    }

   
The name of of the server. 
503
 
504
    public String getServerHostName() throws Exception
505
    {
506
       return .;
507
    }
508
    
509
    public int getNumberOfUsedConnections()
510
    {
511
       if ( == null)
512
          return 0;
513
       
514
       return  - (int) .permits();
515
    }
516
    
517
    public int getNumberOfAvailableConnections()
518
    {
519
       if ( == null)
520
          return 0;
521
       
522
       return (int) .permits();
523
    }
524
 
525
    // Package protected ----------------------------------------------------------------------------
526
 
527
    // Protected ------------------------------------------------------------------------------------
528
 
529
    protected void setup() throws Exception
530
    {
531
       Properties props = new Properties();
532
       props.putAll();
533
       mapJavaBeanProperties(MicroSocketClientInvoker.this, props, false);
534
       configureParameters();
535
 
537
       {
538
           = getAddressByName(.getHost());
539
           = .getPort();
540
           = createServerAddress(, );
541
       }
542
       else
543
       {
544
          List homes = .getConnectHomeList();
545
          if (homes.size() == 1)
546
          {
547
             // Treat as in non MULTIHOME case.
548
             Home home = (Home) homes.iterator().next();
549
              = getAddressByName(home.host);
550
              = createServerAddress(, home.port);
551
          }
552
       }
553
    }
554
 
555
    protected void configureParameters()
556
    {
557
       Map params = ;
558
 
559
       if (params == null)
560
       {
561
          return;
562
       }
563
 
564
       // look for enableTcpNoDelay param
565
       Object val = params.get();
566
       if (val != null)
567
       {
568
          try
569
          {
570
              = Boolean.valueOf((String)val).booleanValue();
571
             .debug(this + " setting enableTcpNoDelay to " + );
572
          }
573
          catch (Exception e)
574
          {
575
             .warn(this + " could not convert " +  + " value of " +
576
                      val + " to a boolean value.");
577
          }
578
       }
579
 
580
       // look for maxPoolSize param
581
       val = params.get();
582
       if (val != null)
583
       {
584
          try
585
          {
586
              = Integer.valueOf((String)val).intValue();
587
             .debug(this + " setting maxPoolSize to " + );
588
          }
589
          catch (Exception e)
590
          {
591
             .warn(this + " could not convert " +  + " value of " +
592
                      val + " to a int value");
593
          }
594
       }
595
 
596
       // look for client socket class name
597
       val = params.get();
598
       if (val != null)
599
       {
600
          String value = (String)val;
601
          if (value.length() > 0)
602
          {
603
              = value;
604
             .debug(this + " setting client socket wrapper class name to " + );
605
          }
606
       }
607
 
608
       val = params.get(.);
609
       if (val != null && ((String)val).length() > 0)
610
       {
611
          String value = (String) val;
612
           = Boolean.valueOf(value).booleanValue();
613
          .debug(this + " setting shouldCheckConnection to " + );
614
       }
615
       else if (getVersion() == .)
616
       {
617
           = true;
618
          .debug(this + " setting shouldCheckConnection to " + );
619
       }
620
       
621
       // look for onewayConnectionTimeout param
622
       val = params.get();
623
       if (val != null)
624
       {
625
          try
626
          {
627
              = Integer.valueOf((String)val).intValue();
628
             .debug(this + " setting onewayConnectionTimeout to " + );
629
          }
630
          catch (Exception e)
631
          {
632
             .warn(this + " could not convert " +  + " value of " +
633
                      val + " to an int value");
634
          }
635
       }
636
       
637
       // look for useOnewayConnectionTimeout param
638
       val = params.get();
639
       if (val != null)
640
       {
641
          try
642
          {
643
              = Boolean.valueOf((String)val).booleanValue();
644
             .debug(this + " setting useOnewayConnectionTimeout to " + );
645
          }
646
          catch (Exception e)
647
          {
648
             .warn(this + " could not convert " +  + " value of " +
649
                      val + " to a boolean value");
650
          }
651
       }
652
       
653
       // look for writeTimeout param
654
       val = params.get();
655
       if (val != null)
656
       {
657
          try
658
          {
659
              = Integer.valueOf((String)val).intValue();
660
             .debug(this + " setting writeTimeout to " + );
661
          }
662
          catch (Exception e)
663
          {
664
             .warn(this + " could not convert " +  + " value of " +
665
                      val + " to an int value");
666
          }
667
       }
668
    }
669
 
670
    protected ServerAddress createServerAddress(InetAddress addr, int port)
671
    {
672
       return new ServerAddress(addr.getHostAddress(), port, , -1, );
673
    }
674
 
675
    protected void finalize() throws Throwable
676
    {
677
       disconnect();
678
       super.finalize();
679
    }
680
 
681
    protected synchronized void handleConnect() throws ConnectionFailedException
682
    {
683
       initPool();
684
       
686
       {
687
           = getUsableAddress();
688
          if ( == null)
689
          {
690
             throw new ConnectionFailedException(this + " unable to find a usable address for: " + );
691
          }
692
          .setHomeInUse();
693
       }
694
       else
695
       {
696
           = new Home(.getHost(), .getPort());
697
       }
698
    }
699
 
700
    protected Home getUsableAddress(InvokerLocator locator)
701
    {
702
       List homes = getConnectHomes();
703
       Iterator it = homes.iterator();
704
       Home home = null;
705
       
706
       while (it.hasNext())
707
       {
708
          try
709
          {
710
             home = (Home) it.next();
711
              = getAddressByName(home.host);
712
              = createServerAddress(, home.port);
713
             invoke(new InvocationRequest(null, null, ., null, null, null));
714
             if () .trace(this + " able to contact server at: " + home);
715
             return home;
716
          }
717
          catch (Throwable e)
718
          {
719
             .debug(this + " unable to contact server at: " + home);
720
          }
721
       }
722
    
723
       return null;
724
    }
725
 
726
    protected synchronized void handleDisconnect()
727
    {
728
       clearPools();
729
       clearPool();
730
    }

   
Each implementation of the remote client invoker should have a default data type that is used in the case it is not specified in the invoker locator URI. 
735
 
736
    protected String getDefaultDataType()
737
    {
738
       return .;
739
    }
740
 
741
    protected Object transport(String sessionID, Object invocation, Map metadata,
742
                               Marshaller marshaller, UnMarshaller unmarshaller)
744
    {
745
       SocketWrapper socketWrapper = null;
746
       Object response = null;
747
       boolean oneway = false;
748
 
749
       // tempTimeout < 0 will indicate there is no per invocation timeout.
750
       int tempTimeout = -1;
751
       int timeLeft = -1;
752
       int savedTimeout = -1;
753
       long start = -1;
754
 
755
       if(metadata != null)
756
       {
757
          // check to see if is one way invocation and return after writing invocation if is
758
          Object val = metadata.get(....);
759
          if(val != null && val instanceof String && Boolean.valueOf((String)val).booleanValue())
760
          {
761
             oneway = true;
762
          }
763
 
764
          // look for temporary timeout values
765
          String tempTimeoutString = (String) metadata.get(.);
766
          {
767
             if (tempTimeoutString != null)
768
             {
769
                try
770
                {
771
                   tempTimeout = Integer.valueOf(tempTimeoutString).intValue();
772
                   .debug(this + " setting timeout to " + tempTimeout + " for this invocation");
773
                }
774
                catch (Exception e)
775
                {
776
                   .warn(this + " could not convert " + . + " value of " +
777
                            tempTimeoutString + " to an integer value.");
778
                }
779
             }
780
          }
781
       }
782
       
783
       if (tempTimeout >= 0)
784
       {
785
          start = System.currentTimeMillis();
786
       }
787
 
788
       boolean serverSideOneway = false;
789
       if (oneway && invocation instanceof InvocationRequest)
790
       {
791
          InvocationRequest ir = (InvocationRequest) invocation;
792
          if (ir.getParameter() instanceof OnewayInvocation)
793
             serverSideOneway = true;
794
       }
795
       
796
       int retryCount = 0;
797
       Exception sockEx = null;
798
 
799
       for (; retryCount < ; retryCount++)
800
       {
801
          if () .trace(this + " retryCount: " + retryCount);
802
          if (0 < tempTimeout)
803
          {
804
             // If a per invocation timeout has been set, the time spent retrying
805
             // should count toward the elapsed time.
806
             timeLeft = (int) (tempTimeout - (System.currentTimeMillis() - start));
807
             if (timeLeft <= 0)
808
                break;
809
          }
810
 
811
          try
812
          {
813
             boolean tryPool = retryCount < ( - 1)
814
                                  ||  == 1
815
                                  ||  == 1;
816
             socketWrapper = getConnection(marshaller, unmarshaller, tryPool, timeLeft);
817
             .trace(this + " got socketWrapper: " + socketWrapper);
818
          }
819
          catch (InterruptedException e)
820
          {
821
             .release();
822
             if () .trace(this + " released semaphore: " + .permits(), e);
823
             throw new RuntimeException(e);
824
          }
825
          catch (Exception e)
826
          {
827
 //            if (bailOut)
828
 //               return null;
829
             .release();
830
             if () .trace(this + " released semaphore: " + .permits(), e);
831
             sockEx =  new CannotConnectException(
832
                   "Can not get connection to server. Problem establishing " +
833
                   "socket connection for " + , e);
834
             continue;
835
          }
836
 
837
          if (tempTimeout >= 0)
838
          {
839
             timeLeft = (int) (tempTimeout - (System.currentTimeMillis() - start));
840
             if (timeLeft <= 0)
841
                break;
842
             savedTimeout = socketWrapper.getTimeout();            
843
             socketWrapper.setTimeout(timeLeft);
844
          }
845
 
846
          try
847
          {
848
             int version = getVersion();
849
             boolean performVersioning = Version.performVersioning(version);
850
 
851
             OutputStream outputStream = socketWrapper.getOutputStream();
852
             .trace(this + "got outputStream: " + outputStream);
853
             if (performVersioning)
854
             {
855
                .trace(this + " writing version");
856
                writeVersion(outputStream, version);
857
                .trace(this + " wrote version");
858
             }
859
 
860
             //TODO: -TME so this is messed up as now ties remoting versioning to using a marshaller type
861
             versionedWrite(outputStream, marshaller, invocation, version);
862
 
863
             if (serverSideOneway)
864
             {
865
                if() { .trace(this + " sent oneway invocation, so not waiting for response, returning null"); }
866
             }
867
             else if (oneway)
868
             {
869
                if (performVersioning && )
870
                {
871
                   int onewaySavedTimeout = socketWrapper.getTimeout();
872
                   socketWrapper.setTimeout();
873
                   InputStream inputStream = socketWrapper.getInputStream();
874
                   version = readVersion(inputStream);
875
                   if (version == -1)
876
                   {
877
                      throw new EOFException("end of file");
878
                   }
879
                   if (version == .)
880
                   {
881
                      .trace(this + " received version 254: treating as end of file");
882
                      throw new EOFException("end of file");
883
                   }
884
 
885
                   // Note that if an exception is thrown, the socket is thrown away,
886
                   // so there's no need to reset the timeout value.
887
                   socketWrapper.setTimeout(onewaySavedTimeout);
888
                }
889
             }
890
             else
891
             {
892
                InputStream inputStream = socketWrapper.getInputStream();
893
                if (performVersioning)
894
                {
895
                   version = readVersion(inputStream);
896
                   if (version == -1)
897
                   {
898
                      throw new EOFException("end of file");
899
                   }
900
                   if (version == .)
901
                   {
902
                      .trace(this + " received version 254: treating as end of file");
903
                      throw new EOFException("end of file");
904
                   }
905
                }
906
 
907
                response = versionedRead(inputStream, unmarshaller, version);
908
             }
909
 
910
             // Note that resetting the timeout value after closing the socket results
911
             // in an exception, so the reset is not done in a finally clause.  However,
912
             // if a catch clause is ever added that does not close the socket, care
913
             // must be taken to reset the timeout in that case.
914
             if (tempTimeout >= 0)
915
             {
916
                socketWrapper.setTimeout(savedTimeout);
917
             }
918
          }
919
          catch (SocketException sex)
920
          {
921
             handleRetriableException(socketWrapper, sex, retryCount);
922
             sockEx = sex;
923
             continue;
924
          }
925
          catch (EOFException ex)
926
          {
927
             handleRetriableException(socketWrapper, ex, retryCount);
928
             sockEx = ex;
929
             continue;
930
          }
931
          catch (IOException e)
932
          {
933
             if (isGeneralizeSocketException() && e.getMessage() != null && .matcher(e.getMessage()).matches())
934
             {
935
                handleRetriableException(socketWrapper, e, retryCount);
936
                sockEx = new SocketException(e.getMessage());
937
                continue;
938
             }
939
             else
940
             {
941
                return handleOtherException(e, , socketWrapper, oneway);
942
             }
943
          }
944
          catch (Exception ex)
945
          {
946
             return handleOtherException(ex, , socketWrapper, oneway);
947
          }
948
 
949
          // call worked, so no need to retry
950
          break;
951
       }
952
 
953
       // need to check if ran out of retries
954
       if (retryCount >= )
955
       {
956
          handleException(sockEx, socketWrapper);
957
       }
958
       
959
       if (response == null && tempTimeout > 0 && timeLeft <= 0)
960
       {
961
          if (sockEx == null)
962
          {
963
             sockEx =  new CannotConnectException(
964
                          "Can not get connection to server. Timed out establishing " +
965
                          "socket connection for " + );
966
          }
967
          handleException(sockEx, socketWrapper);
968
       }
969
 
970
       // Put socket back in pool for reuse
971
       synchronized ()
972
       {
973
          if (.size() < )
974
          {
975
             .add(socketWrapper);
976
             if () { .trace(this + " returned " + socketWrapper + " to pool"); }
977
          }
978
          else
979
          {
980
             if () { .trace(this + "'s pool is full, will close the connection"); }
981
             try
982
             {
983
                socketWrapper.close();
984
             }
985
             catch (Exception ignored)
986
             {
987
             }
988
          }         
989
          .release();
990
          if () .trace(this + " released semaphore: " + .permits());
991
       }
992
 
993
       if ( && !oneway) { .trace(this + " received response " + response);  }
994
       return response;
995
    }
996
 
997
    protected Object handleException(Exception ex, SocketWrapper socketWrapper)
999
    {
1000
      if (ex instanceof ClassNotFoundException)
1001
      {
1002
         //TODO: -TME Add better exception handling for class not found exception
1003
         .debug("Error loading classes from remote call result.", ex);
1004
         throw (ClassNotFoundException)ex;
1005
      }
1006
      
1007
      if (ex instanceof CannotConnectException)
1008
      {
1009
         .debug(this, ex);
1010
         throw (CannotConnectException) ex;
1011
      }
1012
      
1013
      if (ex instanceof InterruptedException)
1014
      {
1015
         .debug(this, ex);
1016
         throw new RuntimeException(ex);
1017
      }
1019
      throw new InvocationFailureException("Unable to perform invocation", ex);
1020
   }
1021
   
1022
   protected void handleRetriableException(SocketWrapper socketWrapper, Exception e, int retryCount)
1023
   {
1024
      if () .trace(this + "(" + socketWrapper + ") got Exception: " + e);
1026
      try
1027
      {
1028
         .release();
1029
         if () .trace(this + " released semaphore: " + .permits());
1030
         socketWrapper.close();            
1031
      }
1032
      catch (Exception ex)
1033
      {
1034
         if () { .trace(this + " couldn't successfully close its socketWrapper", ex); }
1035
      }

      
About to run out of retries and pool may be full of timed out sockets, so want to flush the pool and try with fresh socket as a last effort. 
1043
      if (retryCount == ( - 2))
1044
      {
1045
         flushConnectionPool();
1046
      }
1047
      
1048
      if ()
1049
      {
1050
         if (retryCount < ( - 1))
1051
         {
1052
            .trace(this + " will try again, retries: " + retryCount + " < " + );
1053
         }
1054
         else
1055
         {
1056
            .trace(this + " retries exhausted");               
1057
         }
1058
      }
1059
   }
1061
   protected Object handleOtherException(Exception ex, Semaphore semaphore, SocketWrapper socketWrapper, boolean oneway)
1063
   {
1064
      .debug(this + " got exception: " + socketWrapper, ex);
1066
      try
1067
      {
1068
         semaphore.release();
1069
         if () .trace(this + " released semaphore: " + semaphore.permits());
1070
         socketWrapper.close();
1071
      }
1072
      catch (Exception ignored)
1073
      {
1074
      }
1075
      
1076
      if (oneway)
1077
         return null;
1078
      else
1079
         return handleException(ex, socketWrapper);
1080
   }
1081
   
1082
   protected void initPool()
1083
   {
1084
      synchronized ()
1085
      {
1086
          = (LinkedList).get();
1087
          = (Semaphore).get();
1088
         if ( == null)
1089
         {
1090
             = new LinkedList();
1091
            .put(, );
1092
            .debug("Creating semaphore with size " + );
1093
             = new Semaphore();
1094
            .put(, );
1095
            
1096
            if ()
1097
            {
1098
               synchronized ()
1099
               {
1100
                  .trace(this + " added new pool (" +  + ") as " + );
1101
               }
1102
            }
1103
         }
1104
         else
1105
         {
1106
            if ()
1107
            {
1108
               synchronized ()
1109
               {
1110
                  .trace(this + " using pool (" +  + ") already defined for " + );
1111
               }
1112
            }
1113
         }
1114
      }
1115
   }
1116
   
1117
   protected SocketWrapper getConnection(Marshaller marshaller,
1118
                                         UnMarshaller unmarshaller,
1119
                                         boolean tryPool, int timeAllowed)
1120
      throws Exception
1121
   {
1122
      long start = System.currentTimeMillis();
1123
      long timeToWait = (timeAllowed > 0) ? timeAllowed : ;
1124
      boolean timedout = !.attempt(timeToWait);
1125
      if () .trace(this + " obtained semaphore: " + .permits());
1126
      
1127
      if (timedout)
1128
      {
1129
         throw new IllegalStateException("Timeout waiting for a free socket");
1130
      }
1131
      
1132
      SocketWrapper pooled = null;
1134
      if (tryPool)
1135
      {
1136
         synchronized ()
1137
         {
1138
            // if connection within pool, use it
1139
            if (.size() > 0)
1140
            {
1141
               pooled = getPooledConnection();
1142
               if () .trace(this + " reusing pooled connection: " + pooled);
1143
            }
1144
         }
1145
      }
1146
      else
1147
      {
1148
         if () .trace(this + " avoiding connection pool, creating new socket");
1149
      }
1151
      if (pooled == null)
1152
      {
1153
         //Need to create a new one  
1154
         Socket socket = null;
1156
         if () { .trace(this + " creating socket "); }
1158
         // timeAllowed < 0 indicates no per invocation timeout has been set.
1159
         int timeRemaining = -1;
1160
         if (0 <= timeAllowed)
1161
         {
1162
            timeRemaining = (int) (timeAllowed - (System.currentTimeMillis() - start));
1163
         }
1164
         
1165
         socket = createSocket(., ., timeRemaining);
1166
         if () .trace(this + " created socket: " + socket);
1168
         socket.setTcpNoDelay(.);
1170
         Map metadata = getLocator().getParameters();
1171
         if (metadata == null)
1172
         {
1173
            metadata = new HashMap(2);
1174
         }
1175
         else
1176
         {
1177
            metadata = new HashMap(metadata);
1178
         }
1179
         metadata.put(., marshaller);
1180
         metadata.put(., unmarshaller);
1181
         if ( > 0)
1182
         {
1183
            metadata.put(., new Integer());
1184
         }
1185
         if (timeAllowed > 0)
1186
         {
1187
            timeRemaining = (int) (timeAllowed - (System.currentTimeMillis() - start));
1188
            
1189
            if (timeRemaining <= 0)
1190
               throw new IllegalStateException("Timeout creating a new socket");
1191
            
1192
            metadata.put(., new Integer(timeRemaining));
1193
         }
1194
         
1195
         pooled = createClientSocket(socket, ., metadata);
1196