Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Enforce cacheLimit in DefaultSubscriptionRegistry

Browse Source 
When the cacheLimit is reached and there is an eviction from the
updateCache, the accessCache is now also updated.

This change also ensures that adding a destination to the cache is
protected with synchronization on the updateCache.

Issue: SPR-13555
This commit is contained in:
Rossen Stoyanchev
2015-10-13 12:28:27 -04:00
parent 799a03e98d
commit 0086dd830b
3 changed files with 62 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/AbstractSubscriptionRegistry.java
View File

@@ -23,6 +23,8 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.util.CollectionUtils;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
/**
@@ -35,6 +37,10 @@ import org.springframework.util.MultiValueMap;
*/
public abstract class AbstractSubscriptionRegistry implements SubscriptionRegistry {
private static MultiValueMap<String, String> EMPTY_MAP =
CollectionUtils.unmodifiableMultiValueMap(new LinkedMultiValueMap<String, String>(0));
protected final Log logger = LogFactory.getLog(getClass());
@@ -104,7 +110,7 @@ public abstract class AbstractSubscriptionRegistry implements SubscriptionRegist
String destination = SimpMessageHeaderAccessor.getDestination(headers);
if (destination == null) {
logger.error("No destination in " + message);
return null;
return EMPTY_MAP;
}
return findSubscriptionsInternal(destination, message);

55
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/DefaultSubscriptionRegistry.java
View File

@@ -115,24 +115,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
@Override
protected MultiValueMap<String, String> findSubscriptionsInternal(String destination, Message<?> message) {
MultiValueMap<String, String> result = this.destinationCache.getSubscriptions(destination);
if (result != null) {
return result;
}
result = new LinkedMultiValueMap<String, String>();
for (SessionSubscriptionInfo info : this.subscriptionRegistry.getAllSubscriptions()) {
for (String destinationPattern : info.getDestinations()) {
if (this.pathMatcher.match(destinationPattern, destination)) {
for (String subscriptionId : info.getSubscriptions(destinationPattern)) {
result.add(info.sessionId, subscriptionId);
}
}
}
}
if (!result.isEmpty()) {
this.destinationCache.addSubscriptions(destination, result);
}
return result;
return this.destinationCache.getSubscriptions(destination, message);
}
@Override
@@ -157,20 +140,38 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
new LinkedHashMap<String, MultiValueMap<String, String>>(DEFAULT_CACHE_LIMIT, 0.75f, true) {
@Override
protected boolean removeEldestEntry(Map.Entry<String, MultiValueMap<String, String>> eldest) {
return size() > getCacheLimit();
if (size() > getCacheLimit()) {
accessCache.remove(eldest.getKey());
return true;
}
else {
return false;
}
}
};
public MultiValueMap<String, String> getSubscriptions(String destination) {
return this.accessCache.get(destination);
}
public void addSubscriptions(String destination, MultiValueMap<String, String> subscriptions) {
synchronized (this.updateCache) {
this.updateCache.put(destination, deepCopy(subscriptions));
this.accessCache.put(destination, subscriptions);
public MultiValueMap<String, String> getSubscriptions(String destination, Message<?> message) {
MultiValueMap<String, String> result = this.accessCache.get(destination);
if (result == null) {
synchronized (this.updateCache) {
result = new LinkedMultiValueMap<String, String>();
for (SessionSubscriptionInfo info : subscriptionRegistry.getAllSubscriptions()) {
for (String destinationPattern : info.getDestinations()) {
if (getPathMatcher().match(destinationPattern, destination)) {
for (String subscriptionId : info.getSubscriptions(destinationPattern)) {
result.add(info.sessionId, subscriptionId);
}
}
}
}
if (!result.isEmpty()) {
this.updateCache.put(destination, deepCopy(result));
this.accessCache.put(destination, result);
}
}
}
return result;
}
public void updateAfterNewSubscription(String destination, String sessionId, String subsId) {

38
spring-messaging/src/test/java/org/springframework/messaging/simp/broker/DefaultSubscriptionRegistryTests.java
View File

@@ -69,7 +69,7 @@ public class DefaultSubscriptionRegistryTests {
MultiValueMap<String, String> actual = this.registry.findSubscriptions(message(dest));
assertEquals("Expected one element " + actual, 1, actual.size());
assertEquals(Arrays.asList(subsId), actual.get(sessId));
assertEquals(Collections.singletonList(subsId), actual.get(sessId));
}
@Test
@@ -116,7 +116,7 @@ public class DefaultSubscriptionRegistryTests {
MultiValueMap<String, String> actual = this.registry.findSubscriptions(message(dest));
assertEquals("Expected one element " + actual, 1, actual.size());
assertEquals(Arrays.asList(subsId), actual.get(sessId));
assertEquals(Collections.singletonList(subsId), actual.get(sessId));
}
@Test // SPR-11657
@@ -143,13 +143,13 @@ public class DefaultSubscriptionRegistryTests {
actual = this.registry.findSubscriptions(message("/topic/PRICE.STOCK.NASDAQ.IBM"));
assertEquals(2, actual.size());
assertEquals(Arrays.asList(subs2, subs1), actual.get(sess1));
assertEquals(Arrays.asList(subs1), actual.get(sess2));
assertEquals(Collections.singletonList(subs1), actual.get(sess2));
this.registry.unregisterAllSubscriptions(sess1);
actual = this.registry.findSubscriptions(message("/topic/PRICE.STOCK.NASDAQ.IBM"));
assertEquals(1, actual.size());
assertEquals(Arrays.asList(subs1), actual.get(sess2));
assertEquals(Collections.singletonList(subs1), actual.get(sess2));
this.registry.registerSubscription(subscribeMessage(sess1, subs1, "/topic/PRICE.STOCK.*.IBM"));
this.registry.registerSubscription(subscribeMessage(sess1, subs2, "/topic/PRICE.STOCK.NASDAQ.IBM"));
@@ -157,20 +157,20 @@ public class DefaultSubscriptionRegistryTests {
actual = this.registry.findSubscriptions(message("/topic/PRICE.STOCK.NASDAQ.IBM"));
assertEquals(2, actual.size());
assertEquals(Arrays.asList(subs1, subs2), actual.get(sess1));
assertEquals(Arrays.asList(subs1), actual.get(sess2));
assertEquals(Collections.singletonList(subs1), actual.get(sess2));
this.registry.unregisterSubscription(unsubscribeMessage(sess1, subs2));
actual = this.registry.findSubscriptions(message("/topic/PRICE.STOCK.NASDAQ.IBM"));
assertEquals(2, actual.size());
assertEquals(Arrays.asList(subs1), actual.get(sess1));
assertEquals(Arrays.asList(subs1), actual.get(sess2));
assertEquals(Collections.singletonList(subs1), actual.get(sess1));
assertEquals(Collections.singletonList(subs1), actual.get(sess2));
this.registry.unregisterSubscription(unsubscribeMessage(sess1, subs1));
actual = this.registry.findSubscriptions(message("/topic/PRICE.STOCK.NASDAQ.IBM"));
assertEquals(1, actual.size());
assertEquals(Arrays.asList(subs1), actual.get(sess2));
assertEquals(Collections.singletonList(subs1), actual.get(sess2));
this.registry.unregisterSubscription(unsubscribeMessage(sess2, subs1));
@@ -222,13 +222,13 @@ public class DefaultSubscriptionRegistryTests {
MultiValueMap<String, String> actual = this.registry.findSubscriptions(message);
assertEquals("Expected one element " + actual, 1, actual.size());
assertEquals(Arrays.asList(subsId), actual.get(sessId));
assertEquals(Collections.singletonList(subsId), actual.get(sessId));
message = message("/topic/PRICE.STOCK.NASDAQ.MSFT");
actual = this.registry.findSubscriptions(message);
assertEquals("Expected one element " + actual, 1, actual.size());
assertEquals(Arrays.asList(subsId), actual.get(sessId));
assertEquals(Collections.singletonList(subsId), actual.get(sessId));
message = message("/topic/PRICE.STOCK.NASDAQ.VMW");
actual = this.registry.findSubscriptions(message);
@@ -249,7 +249,7 @@ public class DefaultSubscriptionRegistryTests {
actual = this.registry.findSubscriptions(message("/foo"));
assertEquals("Expected 1 element", 1, actual.size());
assertEquals(Arrays.asList("subs02"), actual.get("sess01"));
assertEquals(Collections.singletonList("subs02"), actual.get("sess01"));
this.registry.unregisterSubscription(unsubscribeMessage("sess01", "subs02"));
@@ -345,6 +345,22 @@ public class DefaultSubscriptionRegistryTests {
// no ConcurrentModificationException
}
@Test // SPR-13555
public void cacheLimitExceeded() throws Exception {
this.registry.setCacheLimit(1);
this.registry.registerSubscription(subscribeMessage("sess1", "1", "/foo"));
this.registry.registerSubscription(subscribeMessage("sess1", "2", "/bar"));
assertEquals(1, this.registry.findSubscriptions(message("/foo")).size());
assertEquals(1, this.registry.findSubscriptions(message("/bar")).size());
this.registry.registerSubscription(subscribeMessage("sess2", "1", "/foo"));
this.registry.registerSubscription(subscribeMessage("sess2", "2", "/bar"));
assertEquals(2, this.registry.findSubscriptions(message("/foo")).size());
assertEquals(2, this.registry.findSubscriptions(message("/bar")).size());
}
private Message<?> subscribeMessage(String sessionId, String subscriptionId, String destination) {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.create(SimpMessageType.SUBSCRIBE);