4:01 p.m.
Author: ron.sigal(a)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)