You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mina.apache.org by be...@apache.org on 2012/07/02 20:15:24 UTC
svn commit: r1356386 - in
/mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh: pom.xml
src/main/java/org/apache/vysper/xmpp/extension/xep0124/BoshBackedSessionContext.java
src/main/java/org/apache/vysper/xmpp/extension/xep0124/InactivityChecker.java
Author: berndf
Date: Mon Jul 2 18:15:23 2012
New Revision: 1356386
URL: http://svn.apache.org/viewvc?rev=1356386&view=rev
Log:
BOSH: simplify inactivity checker, call endSession() to correctly terminate a timed-out BOSH session
Modified:
mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/pom.xml
mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/src/main/java/org/apache/vysper/xmpp/extension/xep0124/BoshBackedSessionContext.java
mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/src/main/java/org/apache/vysper/xmpp/extension/xep0124/InactivityChecker.java
Modified: mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/pom.xml
URL: http://svn.apache.org/viewvc/mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/pom.xml?rev=1356386&r1=1356385&r2=1356386&view=diff
==============================================================================
--- mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/pom.xml (original)
+++ mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/pom.xml Mon Jul 2 18:15:23 2012
@@ -30,6 +30,12 @@
<dependencies>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>12.0</version>
+ </dependency>
+
<dependency>
<groupId>org.apache.vysper</groupId>
<artifactId>spec-compliance</artifactId>
@@ -54,7 +60,7 @@
<scope>compile</scope>
</dependency>
- <!-- Test dependencies -->
+ <!-- Test dependencies -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
Modified: mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/src/main/java/org/apache/vysper/xmpp/extension/xep0124/BoshBackedSessionContext.java
URL: http://svn.apache.org/viewvc/mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/src/main/java/org/apache/vysper/xmpp/extension/xep0124/BoshBackedSessionContext.java?rev=1356386&r1=1356385&r2=1356386&view=diff
==============================================================================
--- mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/src/main/java/org/apache/vysper/xmpp/extension/xep0124/BoshBackedSessionContext.java (original)
+++ mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/src/main/java/org/apache/vysper/xmpp/extension/xep0124/BoshBackedSessionContext.java Mon Jul 2 18:15:23 2012
@@ -55,11 +55,11 @@ public class BoshBackedSessionContext ex
public final static String HTTP_SESSION_ATTRIBUTE = "org.apache.vysper.xmpp.extension.xep0124.BoshBackedSessionContext";
- private final int maxpause = 120;
+ private final int maxpauseSeconds = 120;
- private final int inactivity = 60;
+ private final int inactivitySeconds = 60;
- private final int polling = 15;
+ private final int pollingSeconds = 15;
private final int maximumSentResponses = 10;
@@ -67,7 +67,7 @@ public class BoshBackedSessionContext ex
* The number of milliseconds that will have to pass for a response to be reported missing to the client by
* responding with a report and time attributes. See Response Acknowledgements in XEP-0124.
*/
- private final int brokenConnectionReportTimeout = 1000;
+ private final int brokenConnectionReportTimeoutMillis = 1000;
/*
* Keeps the suspended HTTP requests (does not respond to them) until the server has an asynchronous message
@@ -99,7 +99,7 @@ public class BoshBackedSessionContext ex
private int hold = 1;
- private int currentInactivity = inactivity;
+ private int currentInactivitySeconds = inactivitySeconds;
/*
* The highest RID that can be read and processed, this is the highest (rightmost) contiguous RID.
@@ -163,7 +163,7 @@ public class BoshBackedSessionContext ex
private synchronized void updateInactivityChecker() {
Long newInactivityExpireTime = null;
if (requestsWindow.isEmpty()) {
- newInactivityExpireTime = latestWriteTimestamp + currentInactivity * 1000;
+ newInactivityExpireTime = latestWriteTimestamp + currentInactivitySeconds * 1000;
if (newInactivityExpireTime.equals(lastInactivityExpireTime)) {
return;
}
@@ -262,10 +262,14 @@ public class BoshBackedSessionContext ex
final AsyncContext asyncContext = this.saveResponse(req, boshResponse);
asyncContext.dispatch();
- latestWriteTimestamp = System.currentTimeMillis();
+ setLatestWriteTimestamp();
updateInactivityChecker();
}
-
+
+ private void setLatestWriteTimestamp() {
+ latestWriteTimestamp = System.currentTimeMillis();
+ }
+
private static boolean isResponseSavable(BoshRequest req, Stanza response) {
// responses to pause requests are not saved
if (req.getBody().getAttributeValue("pause") != null) {
@@ -322,9 +326,6 @@ public class BoshBackedSessionContext ex
asyncContext.dispatch();
}
- serverRuntimeContext.getResourceRegistry().unbindSession(this);
- sessionStateHolder.setState(SessionState.CLOSED);
-
inactivityChecker.updateExpireTime(this, lastInactivityExpireTime, null);
lastInactivityExpireTime = null;
@@ -357,7 +358,7 @@ public class BoshBackedSessionContext ex
* @return
*/
public int getMaxPause() {
- return maxpause;
+ return maxpauseSeconds;
}
/**
@@ -453,7 +454,7 @@ public class BoshBackedSessionContext ex
* @return the BOSH 'inactivity' parameter
*/
public int getInactivity() {
- return inactivity;
+ return inactivitySeconds;
}
/**
@@ -461,7 +462,7 @@ public class BoshBackedSessionContext ex
* @return the BOSH 'polling' parameter
*/
public int getPolling() {
- return polling;
+ return pollingSeconds;
}
/**
@@ -504,7 +505,7 @@ public class BoshBackedSessionContext ex
LOGGER.debug("rid = {} - inserting new BOSH request", rid);
// reset the inactivity
- currentInactivity = inactivity;
+ currentInactivitySeconds = inactivitySeconds;
final AsyncContext context = br.getHttpServletRequest().startAsync();
this.addContinuationExpirationListener(context);
@@ -540,14 +541,14 @@ public class BoshBackedSessionContext ex
if (requestsWindow.size() + 1 == currentRequests && !"terminate".equals(stanza.getAttributeValue("type"))
&& stanza.getAttributeValue("pause") == null && stanza.getInnerElements().isEmpty()) {
if (!requestsWindow.isEmpty()
- && br.getTimestamp() - requestsWindow.get(requestsWindow.lastKey()).getTimestamp() < polling * 1000) {
+ && br.getTimestamp() - requestsWindow.get(requestsWindow.lastKey()).getTimestamp() < pollingSeconds * 1000) {
LOGGER.warn("BOSH Overactivity: Too frequent requests");
error(br, "policy-violation");
return;
}
}
if ((wait == 0 || hold == 0) && stanza.getInnerElements().isEmpty()) {
- if (latestEmptyPollingRequest != null && br.getTimestamp() - latestEmptyPollingRequest.getTimestamp() < polling * 1000) {
+ if (latestEmptyPollingRequest != null && br.getTimestamp() - latestEmptyPollingRequest.getTimestamp() < pollingSeconds * 1000) {
LOGGER.warn("BOSH Overactivity for polling: Too frequent requests");
error(br, "policy-violation");
return;
@@ -587,7 +588,7 @@ public class BoshBackedSessionContext ex
long ack = Long.parseLong(stanza.getAttributeValue("ack"));
if (ack < sentResponses.lastKey() && sentResponses.containsKey(ack + 1)) {
long delta = System.currentTimeMillis() - sentResponses.get(ack + 1).getTimestamp();
- if (delta >= brokenConnectionReportTimeout) {
+ if (delta >= brokenConnectionReportTimeoutMillis) {
sendBrokenConnectionReport(ack + 1, delta);
return;
}
@@ -609,7 +610,7 @@ public class BoshBackedSessionContext ex
return;
}
pause = Math.max(0, pause);
- pause = Math.min(pause, maxpause);
+ pause = Math.min(pause, maxpauseSeconds);
respondToPause(pause);
return;
}
@@ -638,7 +639,7 @@ public class BoshBackedSessionContext ex
protected void respondToPause(int pause) {
LOGGER.debug("Setting inactivity period to {}", pause);
- currentInactivity = pause;
+ currentInactivitySeconds = pause;
for (;;) {
BoshRequest boshRequest = getNextRequest();
if (boshRequest == null) {
@@ -688,7 +689,7 @@ public class BoshBackedSessionContext ex
final AsyncContext asyncContext = this.saveResponse(br, boshResponse);
asyncContext.dispatch();
- this.latestWriteTimestamp = System.currentTimeMillis();
+ setLatestWriteTimestamp();
this.updateInactivityChecker();
}
Modified: mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/src/main/java/org/apache/vysper/xmpp/extension/xep0124/InactivityChecker.java
URL: http://svn.apache.org/viewvc/mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/src/main/java/org/apache/vysper/xmpp/extension/xep0124/InactivityChecker.java?rev=1356386&r1=1356385&r2=1356386&view=diff
==============================================================================
--- mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/src/main/java/org/apache/vysper/xmpp/extension/xep0124/InactivityChecker.java (original)
+++ mina/vysper/trunk/server/extensions/xep0124-xep0206-bosh/src/main/java/org/apache/vysper/xmpp/extension/xep0124/InactivityChecker.java Mon Jul 2 18:15:23 2012
@@ -19,16 +19,15 @@
*/
package org.apache.vysper.xmpp.extension.xep0124;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-import java.util.SortedMap;
-import java.util.TreeMap;
-
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.ListMultimap;
+import org.apache.vysper.xmpp.server.SessionContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.HashSet;
+import java.util.List;
+
/**
* Periodically checks the BOSH sessions to see if there are inactive sessions,
* in this case it will close the inactive sessions.
@@ -49,10 +48,19 @@ public class InactivityChecker extends T
private final static Logger LOGGER = LoggerFactory.getLogger(InactivityChecker.class);
+ /**
+ * reduces the key resolution to 10th of a second
+ * @param timestampMillis
+ * @return timestampDeci
+ */
+ public static long convertToKey(long timestampMillis) {
+ return timestampMillis / 100;
+ }
+
/*
* The interval in milliseconds between two consecutive inactivity checks.
*/
- private final int CHECKING_INTERVAL = 1000;
+ private final int CHECKING_INTERVAL_MILLIS = 1000;
/*
* Keeps the BOSH sessions sorted according to the time they expire (the key of the map).
@@ -60,12 +68,12 @@ public class InactivityChecker extends T
* The value associated with a key in the map can be a BoshBackedSessionContext
* or a Set<BoshBackedSessionContext> (in case more than one session expires at the same time).
*/
- private final SortedMap<Long, Object> sessions;
+ private final ListMultimap<Long, BoshBackedSessionContext> sessions = ArrayListMultimap.create();
+
public InactivityChecker() {
setName(InactivityChecker.class.getSimpleName());
setDaemon(true);
- sessions = new TreeMap<Long, Object>();
}
/**
@@ -86,45 +94,31 @@ public class InactivityChecker extends T
* then the session will be removed from the inactivity checker)
* @return true if the inactivity checker is watching the session (to detect inactivity), false otherwise
*/
- @SuppressWarnings({ "unchecked", "rawtypes" })
+ @SuppressWarnings({"rawtypes"})
public boolean updateExpireTime(BoshBackedSessionContext session, Long oldExpireTime, Long newExpireTime) {
boolean ret = session.isWatchedByInactivityChecker();
- if ((oldExpireTime == null && newExpireTime == null)
- || (newExpireTime != null && newExpireTime.equals(oldExpireTime))) {
+ if (oldExpireTime == null && newExpireTime == null) {
+ return ret;
+ }
+ Long oldExpireTimeKey = oldExpireTime == null ? null : InactivityChecker.convertToKey(oldExpireTime);
+ Long newExpireTimeKey = newExpireTime == null ? null : InactivityChecker.convertToKey(newExpireTime);
+ if (newExpireTimeKey != null && newExpireTimeKey.equals(oldExpireTimeKey)) {
+ // do not update the key, if it didn't change
return ret;
}
synchronized (sessions) {
- if (oldExpireTime != null) {
- Object oldValue = sessions.get(oldExpireTime);
- if (oldValue instanceof Set) {
- ((Set) oldValue).remove(session);
- if (((Set) oldValue).isEmpty()) {
- sessions.remove(oldExpireTime);
- }
- } else if (oldValue != null) {
- sessions.remove(oldExpireTime);
- }
+ if (oldExpireTimeKey != null) {
+ sessions.remove(oldExpireTimeKey, session);
ret = false;
}
- if (newExpireTime != null) {
- Object value = sessions.get(newExpireTime);
- if (value instanceof Set) {
- ((Set) value).add(session);
- } else if (value == null) {
- sessions.put(newExpireTime, session);
- } else {
- Set<BoshBackedSessionContext> set = new HashSet<BoshBackedSessionContext>();
- sessions.put(newExpireTime, set);
- set.add((BoshBackedSessionContext) value);
- set.add(session);
- }
+ if (newExpireTimeKey != null) {
+ sessions.put(newExpireTimeKey, session);
ret = true;
}
}
return ret;
}
- @SuppressWarnings("unchecked")
@Override
public void run() {
for (;;) {
@@ -134,48 +128,28 @@ public class InactivityChecker extends T
synchronized (this) {
try {
- wait(CHECKING_INTERVAL);
+ wait(CHECKING_INTERVAL_MILLIS);
} catch (InterruptedException e) {
break;
}
}
- long time = System.currentTimeMillis();
-
- // the inactive sessions are saved in a list to close them after the synchronized block to prevent a deadlock
- // that could happen when trying to close the sessions inside the synchronized block
- List<BoshBackedSessionContext> inactiveSessions = null;
+ long nowKey = convertToKey(System.currentTimeMillis());
+ final HashSet<Long> keys;
synchronized (sessions) {
- // get the oldest key
- Long expireTime = sessions.isEmpty() ? null : sessions.firstKey();
-
- while (expireTime != null) {
- // as long as we find expired sessions we save them in the list and remove them from our sorted map
- if (time >= expireTime) {
- if (inactiveSessions == null) {
- inactiveSessions = new ArrayList<BoshBackedSessionContext>();
- }
- Object value = sessions.get(expireTime);
- if (value instanceof Set) {
- inactiveSessions.addAll((Set<BoshBackedSessionContext>) value);
- } else if (value != null) {
- inactiveSessions.add((BoshBackedSessionContext) value);
- }
- sessions.remove(expireTime);
- expireTime = sessions.isEmpty() ? null : sessions.firstKey();
- } else {
- // at the first non-expired session, we know that all the next sessions are more recent and cannot
- // be expired if the current session is Ok, so we break the loop
- break;
- }
- }
+ // get all keys
+ keys = new HashSet<Long>(sessions.keySet());
}
- if (inactiveSessions != null) {
- for (BoshBackedSessionContext session : inactiveSessions) {
- LOGGER.error("BOSH session {} reached maximum inactivity period, closing session...", session);
- session.close();
+ for (Long expireTime : keys) {
+ // expire old sessions
+ if (nowKey >= expireTime) {
+ final List<BoshBackedSessionContext> inactiveSessions = sessions.removeAll(expireTime);
+ for (BoshBackedSessionContext session : inactiveSessions) {
+ LOGGER.error("BOSH session {} reached maximum inactivity period, closing session...", session);
+ session.endSession(SessionContext.SessionTerminationCause.CONNECTION_ABORT);
+ }
}
}
}