Author: ron.sigal(a)jboss.com Date: 2009-05-10 01:40:51 -0400 (Sun, 10 May 2009) New Revision: 5170 Modified: remoting2/branches/2.2/src/main/org/jboss/remoting/LeasePinger.java Log: (1) JBREM-1128: Adopted use of leasePingerId; (2) JBREM-1129: adopted use of time stamp. Modified: remoting2/branches/2.2/src/main/org/jboss/remoting/LeasePinger.java =================================================================== --- remoting2/branches/2.2/src/main/org/jboss/remoting/LeasePinger.java 2009-05-10 05:36:45 UTC (rev 5169) +++ remoting2/branches/2.2/src/main/org/jboss/remoting/LeasePinger.java 2009-05-10 05:40:51 UTC (rev 5170) @@ -25,6 +25,9 @@ public static final long DEFAULT_LEASE_PERIOD = 5000; public static final int DEFAULT_DISCONNECT_TIMEOUT = -1; + + static final String LEASE_PINGER_ID = "leasePingerId"; + static final String TIME_STAMP = "timeStamp"; // Static --------------------------------------------------------------------------------------- @@ -39,11 +42,15 @@ private ClientInvoker invoker = null; private String invokerSessionID = null; + private Map clientSessionIds = new ConcurrentHashMap(); private Map clients = new ConcurrentHashMap(); private TimerTask timerTask = null; private long pingPeriod = -1; private int disconnectTimeout = DEFAULT_DISCONNECT_TIMEOUT; + + private String leasePingerId; + private boolean useClientConnectionIdentity; // Constructors --------------------------------------------------------------------------------- @@ -84,6 +91,18 @@ timerTask.cancel(); timerTask = null; + if (useClientConnectionIdentity) + { + Iterator it = clients.values().iterator(); + while (it.hasNext()) + { + Client client = (Client) it.next(); + if (trace) log.trace(this + " calling " + client + ".notifyAndDisconnect()"); + client.notifyListeners(); + it.remove(); + } + } + try { // sending request map with no ClientHolders will indicate to server @@ -91,7 +110,7 @@ HashMap metadata = null; // If disconnectTimeout == 0, skip network i/o. - log.debug(this + ": disconnectTimeout: " + disconnectTimeout); + if (trace) log.trace(this + ": disconnectTimeout: " + disconnectTimeout); if (disconnectTimeout != 0) { if (disconnectTimeout > 0) @@ -110,6 +129,25 @@ e.initCause(throwable); throw e; } + + if (trace) + { + log.trace(this + " shut down"); + if (!clientSessionIds.isEmpty()) + { + log.trace(this + " " + clientSessionIds.size() + " remaining clients:"); + Iterator it = clientSessionIds.keySet().iterator(); + while (it.hasNext()) + { + log.trace(this + ": " + it.next()); + } + clientSessionIds.clear(); + } + else + { + log.trace(this + " No remaining clients"); + } + } } } @@ -122,8 +160,17 @@ if(trace) { log.trace(this + " adding new client with session ID " + sessionID + " and lease period " + leasePeriod); } + if (useClientConnectionIdentity) + { + Client client = (Client) configuration.remove(Client.CLIENT); + if (client != null) + { + clients.put(sessionID, client); + } + } + ClientHolder newClient = new ClientHolder(sessionID, configuration, leasePeriod); - clients.put(sessionID, newClient); + clientSessionIds.put(sessionID, newClient); sendClientPing(); @@ -148,7 +195,9 @@ if(trace) { log.trace(this + " removing client with session ID " + sessionID); } - ClientHolder holder = (ClientHolder)clients.remove(sessionID); + // Don't remove holder until after client has been removed from server side Lease, to + // avoid a race with LeaseTimerTask sending a PING without the Client being removed. + ClientHolder holder = (ClientHolder)clientSessionIds.get(sessionID); if (holder != null) { @@ -176,14 +225,20 @@ log.warn(this + " failed sending disconnect for client lease for " + "client with session ID " + sessionID); } + + clientSessionIds.remove(sessionID); + if (useClientConnectionIdentity) + { + clients.remove(sessionID); + } } else { - log.warn(this + " tried to remove lease for client with session ID " + sessionID + - ", but no such lease was found"); + log.debug(this + " tried to remove lease for client with session ID " + sessionID + + ", but no such lease was found: probably it was registered with an older LeasePinger"); } - if (clients.isEmpty()) + if (clientSessionIds.isEmpty()) { isLastClientLease = true; if(trace) { log.trace(this + " has no more client leases"); } @@ -195,7 +250,7 @@ long tempPingPeriod = defaultPingPeriod; - for (Iterator i = clients.values().iterator(); i.hasNext(); ) + for (Iterator i = clientSessionIds.values().iterator(); i.hasNext(); ) { ClientHolder clientHolder = (ClientHolder)i.next(); long clientHolderLeasePeriod = clientHolder.getLeasePeriod(); @@ -231,7 +286,7 @@ } // look to see if the client is still amont those serviced by this lease pinger - if (clients.containsKey(sessionID)) + if (clientSessionIds.containsKey(sessionID)) { return pingPeriod; } @@ -243,7 +298,7 @@ public String toString() { - return "LeasePinger[" + invoker + "(" + invokerSessionID + ")]"; + return "LeasePinger[" + leasePingerId + ":" + invoker + "(" + invokerSessionID + ")]"; } // Package protected ---------------------------------------------------------------------------- @@ -259,9 +314,29 @@ protected void setDisconnectTimeout(int disconnectTimeout) { this.disconnectTimeout = disconnectTimeout; - log.debug(this + " setting disconnect timeout to: " + disconnectTimeout); + if (trace) log.trace(this + " setting disconnect timeout to: " + disconnectTimeout); } + protected String getLeasePingerId() + { + return leasePingerId; + } + + protected void setLeasePingerId(String leasePingerId) + { + this.leasePingerId = leasePingerId; + } + + boolean isUseClientConnectionIdentity() + { + return useClientConnectionIdentity; + } + + void setUseClientConnectionIdentity(boolean useClientConnectionIdentity) + { + this.useClientConnectionIdentity = useClientConnectionIdentity; + } + // Private -------------------------------------------------------------------------------------- private void sendClientPing() @@ -271,9 +346,9 @@ if(trace) { StringBuffer sb = new StringBuffer(); - if(clients != null) + if(clientSessionIds != null) { - for(Iterator i = clients.values().iterator(); i.hasNext(); ) + for(Iterator i = clientSessionIds.values().iterator(); i.hasNext(); ) { ClientHolder h = (ClientHolder)i.next(); sb.append(" ").append(h.getSessionId()).append('\n'); @@ -284,13 +359,13 @@ "for following clients:\n" + sb.toString()); } - Map clientsClone = new ConcurrentHashMap(clients); + Map clientsClone = new ConcurrentHashMap(clientSessionIds); Map requestClients = new ConcurrentHashMap(); requestClients.put(ClientHolder.CLIENT_HOLDER_KEY, clientsClone); - - InvocationRequest ir = - new InvocationRequest(invokerSessionID, null, "$PING$", requestClients, null, null); - + requestClients.put(LeasePinger.LEASE_PINGER_ID, leasePingerId); + requestClients.put(TIME_STAMP, Long.toString(System.currentTimeMillis())); + + InvocationRequest ir = new InvocationRequest(invokerSessionID, null, "$PING$", requestClients, null, null); invoker.invoke(ir); if(trace) { log.trace(this + " successfully pinged the server"); }