Avoid ConcurrentModificationException

sdeleuze · rstoyanchev · commit 96da77ef759e · 2014-05-06T20:22:05.000-04:00
Removal of cached destination is now moved outside the for loop
that removes subscriptions to avoid ConcurrentModificationException.

Also since updateCache is a LinkedHashMap with accessOrder=true,
a simple access with updateCache.get() modify the map.
By iterating over updateCache.entrySet(), we avoid this update.

Issue: SPR-11755
diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/broker/DefaultSubscriptionRegistry.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/broker/DefaultSubscriptionRegistry.java
@@ -55,7 +55,6 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
 	private PathMatcher pathMatcher = new AntPathMatcher();
 
 
-
 	/**
 	 * Specify the maximum number of entries for the resolved destination cache.
 	 * Default is 1024.
@@ -115,7 +114,7 @@ public void unregisterAllSubscriptions(String sessionId) {
 
 	@Override
 	protected MultiValueMap<String, String> findSubscriptionsInternal(String destination, Message<?> message) {
-		MultiValueMap<String,String> result = this.destinationCache.getSubscriptions(destination);
+		MultiValueMap<String, String> result = this.destinationCache.getSubscriptions(destination);
 		if (result != null) {
 			return result;
 		}
@@ -129,7 +128,7 @@ protected MultiValueMap<String, String> findSubscriptionsInternal(String destina
 				}
 			}
 		}
-		if(!result.isEmpty()) {
+		if (!result.isEmpty()) {
 			this.destinationCache.addSubscriptions(destination, result);
 		}
 		return result;
@@ -175,10 +174,11 @@ public void addSubscriptions(String destination, MultiValueMap<String, String> s
 		}
 
 		public void updateAfterNewSubscription(String destination, String sessionId, String subsId) {
-			synchronized(this.updateCache) {
-				for (String cachedDestination : this.updateCache.keySet()) {
+			synchronized (this.updateCache) {
+				for (Map.Entry<String, MultiValueMap<String, String>> entry : this.updateCache.entrySet()) {
+					String cachedDestination = entry.getKey();
 					if (getPathMatcher().match(destination, cachedDestination)) {
-						MultiValueMap<String, String> subs = this.updateCache.get(cachedDestination);
+						MultiValueMap<String, String> subs = entry.getValue();
 						subs.add(sessionId, subsId);
 						this.accessCache.put(cachedDestination, new LinkedMultiValueMap<String, String>(subs));
 					}
@@ -187,43 +187,53 @@ public void updateAfterNewSubscription(String destination, String sessionId, Str
 		}
 
 		public void updateAfterRemovedSubscription(String destination, String sessionId, String subsId) {
-			synchronized(this.updateCache) {
-				for (String cachedDestination : this.updateCache.keySet()) {
+			synchronized (this.updateCache) {
+				Set<String> destinationsToRemove = new HashSet<String>();
+				for (Map.Entry<String, MultiValueMap<String, String>> entry : this.updateCache.entrySet()) {
+					String cachedDestination = entry.getKey();
 					if (getPathMatcher().match(destination, cachedDestination)) {
-						MultiValueMap<String, String> subs = this.updateCache.get(cachedDestination);
+						MultiValueMap<String, String> subs = entry.getValue();
 						List<String> subsIds = subs.get(sessionId);
 						subsIds.remove(subsId);
 						if (subsIds.isEmpty()) {
 							subs.remove(sessionId);
 						}
 						if (subs.isEmpty()) {
-							this.updateCache.remove(cachedDestination);
-							this.accessCache.remove(cachedDestination);
+							destinationsToRemove.add(cachedDestination);
 						}
 						else {
 							this.accessCache.put(cachedDestination, new LinkedMultiValueMap<String, String>(subs));
 						}
 					}
 				}
+				for (String d : destinationsToRemove) {
+					this.updateCache.remove(d);
+					this.accessCache.remove(d);
+				}
 			}
 		}
 
 		public void updateAfterRemovedSession(SessionSubscriptionInfo info) {
-			synchronized(this.updateCache) {
+			synchronized (this.updateCache) {
 				for (String destination : info.getDestinations()) {
-					for (String cachedDestination : this.updateCache.keySet()) {
+					Set<String> destinationsToRemove = new HashSet<String>();
+					for (Map.Entry<String, MultiValueMap<String, String>> entry : this.updateCache.entrySet()) {
+						String cachedDestination = entry.getKey();
 						if (getPathMatcher().match(destination, cachedDestination)) {
-							MultiValueMap<String, String> subs = this.updateCache.get(cachedDestination);
+							MultiValueMap<String, String> subs = entry.getValue();
 							subs.remove(info.getSessionId());
 							if (subs.isEmpty()) {
-								this.updateCache.remove(cachedDestination);
-								this.accessCache.remove(cachedDestination);
+								destinationsToRemove.add(cachedDestination);
 							}
 							else {
 								this.accessCache.put(cachedDestination,new LinkedMultiValueMap<String, String>(subs));
 							}
 						}
 					}
+					for (String d : destinationsToRemove) {
+						this.updateCache.remove(d);
+						this.accessCache.remove(d);
+					}
 				}
 			}
 		}
diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/broker/DefaultSubscriptionRegistryTests.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/broker/DefaultSubscriptionRegistryTests.java
@@ -185,6 +185,41 @@ public void registerSubscriptionsWithSimpleAndPatternDestinations() {
 		assertEquals(0, actual.size());
 	}
 
+	// SPR-11755
+
+	@Test
+	public void registerAndUnregisterMultipleDestinations() {
+
+		String sess1 = "sess01";
+		String sess2 = "sess02";
+
+		String subs1 = "subs01";
+		String subs2 = "subs02";
+		String subs3 = "subs03";
+		String subs4 = "subs04";
+		String subs5 = "subs05";
+
+		this.registry.registerSubscription(subscribeMessage(sess1, subs1, "/topic/PRICE.STOCK.NASDAQ.IBM"));
+		this.registry.registerSubscription(subscribeMessage(sess1, subs2, "/topic/PRICE.STOCK.NYSE.IBM"));
+		this.registry.registerSubscription(subscribeMessage(sess1, subs3, "/topic/PRICE.STOCK.NASDAQ.GOOG"));
+
+		this.registry.findSubscriptions(message("/topic/PRICE.STOCK.NYSE.IBM"));
+		this.registry.findSubscriptions(message("/topic/PRICE.STOCK.NASDAQ.GOOG"));
+		this.registry.findSubscriptions(message("/topic/PRICE.STOCK.NASDAQ.IBM"));
+
+		this.registry.unregisterSubscription(unsubscribeMessage(sess1, subs1));
+		this.registry.unregisterSubscription(unsubscribeMessage(sess1, subs2));
+		this.registry.unregisterSubscription(unsubscribeMessage(sess1, subs3));
+
+		this.registry.registerSubscription(subscribeMessage(sess1, subs1, "/topic/PRICE.STOCK.NASDAQ.IBM"));
+		this.registry.registerSubscription(subscribeMessage(sess1, subs2, "/topic/PRICE.STOCK.NYSE.IBM"));
+		this.registry.registerSubscription(subscribeMessage(sess1, subs3, "/topic/PRICE.STOCK.NASDAQ.GOOG"));
+		this.registry.registerSubscription(subscribeMessage(sess1, subs4, "/topic/PRICE.STOCK.NYSE.IBM"));
+		this.registry.registerSubscription(subscribeMessage(sess2, subs5, "/topic/PRICE.STOCK.NASDAQ.GOOG"));
+		this.registry.unregisterAllSubscriptions(sess1);
+		this.registry.unregisterAllSubscriptions(sess2);
+	}
+
 	@Test
 	public void registerSubscriptionWithDestinationPatternRegex() {
 

Original file line number	Diff line number	Diff line change
`@@ -55,7 +55,6 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {`
`55`	`55`	`private PathMatcher pathMatcher = new AntPathMatcher();`
`56`	`56`
`57`	`57`
`58`		`-`
`59`	`58`	`/**`
`60`	`59`	`* Specify the maximum number of entries for the resolved destination cache.`
`61`	`60`	`* Default is 1024.`
`@@ -115,7 +114,7 @@ public void unregisterAllSubscriptions(String sessionId) {`
`115`	`114`
`116`	`115`	`@Override`
`117`	`116`	`protected MultiValueMap<String, String> findSubscriptionsInternal(String destination, Message<?> message) {`
`118`		`- MultiValueMap<String,String> result = this.destinationCache.getSubscriptions(destination);`
	`117`	`+ MultiValueMap<String, String> result = this.destinationCache.getSubscriptions(destination);`
`119`	`118`	`if (result != null) {`
`120`	`119`	`return result;`
`121`	`120`	`}`
`@@ -129,7 +128,7 @@ protected MultiValueMap<String, String> findSubscriptionsInternal(String destina`
`129`	`128`	`}`
`130`	`129`	`}`
`131`	`130`	`}`
`132`		`- if(!result.isEmpty()) {`
	`131`	`+ if (!result.isEmpty()) {`
`133`	`132`	`this.destinationCache.addSubscriptions(destination, result);`
`134`	`133`	`}`
`135`	`134`	`return result;`
`@@ -175,10 +174,11 @@ public void addSubscriptions(String destination, MultiValueMap<String, String> s`
`175`	`174`	`}`
`176`	`175`
`177`	`176`	`public void updateAfterNewSubscription(String destination, String sessionId, String subsId) {`
`178`		`- synchronized(this.updateCache) {`
`179`		`- for (String cachedDestination : this.updateCache.keySet()) {`
	`177`	`+ synchronized (this.updateCache) {`
	`178`	`+ for (Map.Entry<String, MultiValueMap<String, String>> entry : this.updateCache.entrySet()) {`
	`179`	`+ String cachedDestination = entry.getKey();`
`180`	`180`	`if (getPathMatcher().match(destination, cachedDestination)) {`
`181`		`- MultiValueMap<String, String> subs = this.updateCache.get(cachedDestination);`
	`181`	`+ MultiValueMap<String, String> subs = entry.getValue();`
`182`	`182`	`subs.add(sessionId, subsId);`
`183`	`183`	`this.accessCache.put(cachedDestination, new LinkedMultiValueMap<String, String>(subs));`
`184`	`184`	`}`
`@@ -187,43 +187,53 @@ public void updateAfterNewSubscription(String destination, String sessionId, Str`
`187`	`187`	`}`
`188`	`188`
`189`	`189`	`public void updateAfterRemovedSubscription(String destination, String sessionId, String subsId) {`
`190`		`- synchronized(this.updateCache) {`
`191`		`- for (String cachedDestination : this.updateCache.keySet()) {`
	`190`	`+ synchronized (this.updateCache) {`
	`191`	`+ Set<String> destinationsToRemove = new HashSet<String>();`
	`192`	`+ for (Map.Entry<String, MultiValueMap<String, String>> entry : this.updateCache.entrySet()) {`
	`193`	`+ String cachedDestination = entry.getKey();`
`192`	`194`	`if (getPathMatcher().match(destination, cachedDestination)) {`
`193`		`- MultiValueMap<String, String> subs = this.updateCache.get(cachedDestination);`
	`195`	`+ MultiValueMap<String, String> subs = entry.getValue();`
`194`	`196`	`List<String> subsIds = subs.get(sessionId);`
`195`	`197`	`subsIds.remove(subsId);`
`196`	`198`	`if (subsIds.isEmpty()) {`
`197`	`199`	`subs.remove(sessionId);`
`198`	`200`	`}`
`199`	`201`	`if (subs.isEmpty()) {`
`200`		`- this.updateCache.remove(cachedDestination);`
`201`		`- this.accessCache.remove(cachedDestination);`
	`202`	`+ destinationsToRemove.add(cachedDestination);`
`202`	`203`	`}`
`203`	`204`	`else {`
`204`	`205`	`this.accessCache.put(cachedDestination, new LinkedMultiValueMap<String, String>(subs));`
`205`	`206`	`}`
`206`	`207`	`}`
`207`	`208`	`}`
	`209`	`+ for (String d : destinationsToRemove) {`
	`210`	`+ this.updateCache.remove(d);`
	`211`	`+ this.accessCache.remove(d);`
	`212`	`+ }`
`208`	`213`	`}`
`209`	`214`	`}`
`210`	`215`
`211`	`216`	`public void updateAfterRemovedSession(SessionSubscriptionInfo info) {`
`212`		`- synchronized(this.updateCache) {`
	`217`	`+ synchronized (this.updateCache) {`
`213`	`218`	`for (String destination : info.getDestinations()) {`
`214`		`- for (String cachedDestination : this.updateCache.keySet()) {`
	`219`	`+ Set<String> destinationsToRemove = new HashSet<String>();`
	`220`	`+ for (Map.Entry<String, MultiValueMap<String, String>> entry : this.updateCache.entrySet()) {`
	`221`	`+ String cachedDestination = entry.getKey();`
`215`	`222`	`if (getPathMatcher().match(destination, cachedDestination)) {`
`216`		`- MultiValueMap<String, String> subs = this.updateCache.get(cachedDestination);`
	`223`	`+ MultiValueMap<String, String> subs = entry.getValue();`
`217`	`224`	`subs.remove(info.getSessionId());`
`218`	`225`	`if (subs.isEmpty()) {`
`219`		`- this.updateCache.remove(cachedDestination);`
`220`		`- this.accessCache.remove(cachedDestination);`
	`226`	`+ destinationsToRemove.add(cachedDestination);`
`221`	`227`	`}`
`222`	`228`	`else {`
`223`	`229`	`this.accessCache.put(cachedDestination,new LinkedMultiValueMap<String, String>(subs));`
`224`	`230`	`}`
`225`	`231`	`}`
`226`	`232`	`}`
	`233`	`+ for (String d : destinationsToRemove) {`
	`234`	`+ this.updateCache.remove(d);`
	`235`	`+ this.accessCache.remove(d);`
	`236`	`+ }`
`227`	`237`	`}`
`228`	`238`	`}`
`229`	`239`	`}`