[jboss-remoting-commits] JBoss Remoting SVN: r5333 - remoting2/branches/2.x/src/main/org/jboss/remoting/transport/bisocket.

[jboss-remoting-commits at lists.jboss.org]

Author: ron.sigal at jboss.com
Date: 2009-08-14 17:01:47 -0400 (Fri, 14 Aug 2009)
New Revision: 5333

Modified:
   remoting2/branches/2.x/src/main/org/jboss/remoting/transport/bisocket/BisocketClientInvoker.java
Log:
JBREM-1147: createSocket() periodically checks for updated controlOuputStream.

Modified: remoting2/branches/2.x/src/main/org/jboss/remoting/transport/bisocket/BisocketClientInvoker.java
===================================================================
--- remoting2/branches/2.x/src/main/org/jboss/remoting/transport/bisocket/BisocketClientInvoker.java	2009-08-14 21:00:16 UTC (rev 5332)
+++ remoting2/branches/2.x/src/main/org/jboss/remoting/transport/bisocket/BisocketClientInvoker.java	2009-08-14 21:01:47 UTC (rev 5333)
@@ -85,6 +85,8 @@
    protected String listenerId;
 
    private int pingFrequency = Bisocket.PING_FREQUENCY_DEFAULT;
+   private int pingWindowFactor = Bisocket.PING_WINDOW_FACTOR_DEFAULT;
+   private int pingWindow = pingWindowFactor * pingFrequency;
    private int maxRetries = Bisocket.MAX_RETRIES_DEFAULT;
    private Socket controlSocket;
    private OutputStream controlOutputStream;
@@ -190,6 +192,26 @@
             }
          }
          
+         val = configuration.get(Bisocket.PING_WINDOW_FACTOR);
+         if (val != null && val instanceof String && ((String) val).length() > 0)
+         {
+            try
+            {
+               pingWindowFactor = Integer.valueOf(((String) val)).intValue();
+               log.debug(this + " setting pingWindowFactor to " + pingWindowFactor);
+            }
+            catch (NumberFormatException e)
+            {
+               log.warn("Invalid format for " + "\"" + Bisocket.PING_WINDOW_FACTOR + "\": " + val);
+            }
+         }
+         else if (val != null)
+         {
+            log.warn("\"" + Bisocket.PING_WINDOW_FACTOR + "\" must be specified as a String");
+         }
+         
+         pingWindow = pingWindowFactor * pingFrequency;
+         
          val = configuration.get(Bisocket.MAX_RETRIES);
          if (val != null)
          {
@@ -231,7 +253,20 @@
       this.pingFrequency = pingFrequency;
    }
 
-
+   
+   public int getPingWindowFactor()
+   {
+      return pingWindowFactor;
+   }
+   
+   
+   public void setPingWindowFactor(int pingWindowFactor)
+   {
+      this.pingWindowFactor = pingWindowFactor;
+      pingWindow = pingWindowFactor * pingFrequency;
+   }
+   
+   
    protected void handleConnect() throws ConnectionFailedException
    {
       // Callback client on server side.
@@ -454,35 +489,47 @@
 
       synchronized (controlLock)
       {
-         controlOutputStream.write(Bisocket.CREATE_ORDINARY_SOCKET);
-      }
-
-      synchronized (sockets)
-      {
-         if (!sockets.isEmpty())
+         if (log.isTraceEnabled()) log.trace(this + " writing Bisocket.CREATE_ORDINARY_SOCKET on " + controlOutputStream);
+         try
          {
-            Iterator it = sockets.iterator();
-            Socket socket = (Socket) it.next();
-            it.remove();
-            log.debug(this + " found socket (" + listenerId + "): " + socket);
-            return socket;
+            controlOutputStream.write(Bisocket.CREATE_ORDINARY_SOCKET);
+            if (log.isTraceEnabled()) log.trace(this + " wrote Bisocket.CREATE_ORDINARY_SOCKET");
+            
+            synchronized (sockets)
+            {
+               if (!sockets.isEmpty())
+               {
+                  Iterator it = sockets.iterator();
+                  Socket socket = (Socket) it.next();
+                  it.remove();
+                  log.debug(this + " found socket (" + listenerId + "): " + socket);
+                  return socket;
+               }
+            }
          }
+         catch (IOException e)
+         {
+            log.debug(this + " unable to write Bisocket.CREATE_ORDINARY_SOCKET", e);
+         }
       }
       
       long timeRemaining = timeout; 
-      long start = System.currentTimeMillis(); 
+      long pingFailedWindow = 2 * pingWindow;
+      long pingFailedTimeRemaining = pingFailedWindow;
+      long start = System.currentTimeMillis();
+      OutputStream savedControlOutputStream = controlOutputStream;
 
-      while (isConnected() && !pingFailed.flag && (timeout == 0 || timeRemaining > 0))
+      while (isConnected() && (!pingFailed.flag || pingFailedTimeRemaining > 0) && (timeout == 0 || timeRemaining > 0))
       {
          synchronized (sockets)
-         {
+         {  
             try
             {
                sockets.wait(1000);
             }
             catch (InterruptedException e)
             {
-               log.debug("unexpected interrupt");
+               log.debug(this + " unexpected interrupt");
             }
             
             if (!sockets.isEmpty())
@@ -494,9 +541,26 @@
                return socket;
             }
          }
-
+         
+         if (savedControlOutputStream != controlOutputStream)
+         {
+            savedControlOutputStream = controlOutputStream;
+            log.debug(this + " rewriting Bisocket.CREATE_ORDINARY_SOCKET on " + controlOutputStream);
+            try
+            {
+               controlOutputStream.write(Bisocket.CREATE_ORDINARY_SOCKET);
+               log.debug(this + " rewrote Bisocket.CREATE_ORDINARY_SOCKET");
+            }
+            catch (IOException e)
+            {
+               log.debug(this + " unable to rewrite Bisocket.CREATE_ORDINARY_SOCKET" + e.getMessage());
+            }
+         }
+         
+         long elapsed = System.currentTimeMillis() - start;
          if (timeout > 0)
-            timeRemaining = timeout - (System.currentTimeMillis() - start);
+            timeRemaining = timeout - elapsed;
+         pingFailedTimeRemaining = pingFailedWindow - elapsed; 
       }
 
       if (!isConnected())
@@ -526,6 +590,7 @@
          controlSocket = socket;
          log.debug(this + " control socket replaced by: " + socket);
          controlOutputStream = controlSocket.getOutputStream();
+         log.debug("controlOutputStream replaced by: " + controlOutputStream);
       }
 
       if (pingTimerTask != null)