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 11:59:01 -04:00
parent 0f70ac74cd
commit 7ff915a01a
2 changed files with 44 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@@ -21,9 +21,6 @@ import java.util.Collections;
import java.util.Iterator; import java.util.Iterator;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test; import org.junit.Test;
@@ -31,11 +28,10 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor; import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType; import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.support.MessageBuilder; import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.AntPathMatcher;
import org.springframework.util.MultiValueMap; import org.springframework.util.MultiValueMap;
import org.springframework.util.PathMatcher;
import static org.junit.Assert.*; import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
/** /**
* Test fixture for * Test fixture for
@@ -402,37 +398,22 @@ public class DefaultSubscriptionRegistryTests {
// no ConcurrentModificationException // no ConcurrentModificationException
} }
@Test // SPR-13204 @Test // SPR-13555
public void findSubscriptionsWithConcurrentUnregisterAllSubscriptions() throws Exception { public void cacheLimitExceeded() throws Exception {
final CountDownLatch iterationPausedLatch = new CountDownLatch(1); this.registry.setCacheLimit(1);
final CountDownLatch iterationResumeLatch = new CountDownLatch(1);
final CountDownLatch iterationDoneLatch = new CountDownLatch(1);
PathMatcher pathMatcher = new PausingPathMatcher(iterationPausedLatch, iterationResumeLatch);
this.registry.setPathMatcher(pathMatcher);
this.registry.registerSubscription(subscribeMessage("sess1", "1", "/foo")); this.registry.registerSubscription(subscribeMessage("sess1", "1", "/foo"));
this.registry.registerSubscription(subscribeMessage("sess1", "2", "/bar"));
assertEquals(1, this.registry.findSubscriptions(createMessage("/foo")).size());
assertEquals(1, this.registry.findSubscriptions(createMessage("/bar")).size());
this.registry.registerSubscription(subscribeMessage("sess2", "1", "/foo")); this.registry.registerSubscription(subscribeMessage("sess2", "1", "/foo"));
this.registry.registerSubscription(subscribeMessage("sess2", "2", "/bar"));
AtomicReference<MultiValueMap<String, String>> subscriptions = new AtomicReference<>(); assertEquals(2, this.registry.findSubscriptions(createMessage("/foo")).size());
new Thread(() -> { assertEquals(2, this.registry.findSubscriptions(createMessage("/bar")).size());
subscriptions.set(registry.findSubscriptions(createMessage("/foo")));
iterationDoneLatch.countDown();
}).start();
assertTrue(iterationPausedLatch.await(10, TimeUnit.SECONDS));
this.registry.unregisterAllSubscriptions("sess1");
this.registry.unregisterAllSubscriptions("sess2");
iterationResumeLatch.countDown();
assertTrue(iterationDoneLatch.await(10, TimeUnit.SECONDS));
MultiValueMap<String, String> result = subscriptions.get();
assertNotNull(result);
assertEquals(0, result.size());
} }
private Message<?> createMessage(String destination) { private Message<?> createMessage(String destination) {
SimpMessageHeaderAccessor accessor = SimpMessageHeaderAccessor.create(); SimpMessageHeaderAccessor accessor = SimpMessageHeaderAccessor.create();
accessor.setDestination(destination); accessor.setDestination(destination);
@@ -468,35 +449,4 @@ public class DefaultSubscriptionRegistryTests {
return list; return list;
} }
/**
* An extension of AntPathMatcher with a pair of CountDownLatches to pause
* while matching, allowing another thread to something, and resume when the
* other thread signals it's okay to do so.
*/
private static class PausingPathMatcher extends AntPathMatcher {
private final CountDownLatch iterationPausedLatch;
private final CountDownLatch iterationResumeLatch;
public PausingPathMatcher(CountDownLatch iterationPausedLatch, CountDownLatch iterationResumeLatch) {
this.iterationPausedLatch = iterationPausedLatch;
this.iterationResumeLatch = iterationResumeLatch;
}
@Override
public boolean match(String pattern, String path) {
try {
this.iterationPausedLatch.countDown();
assertTrue(this.iterationResumeLatch.await(10, TimeUnit.SECONDS));
return super.match(pattern, path);
}
catch (InterruptedException ex) {
ex.printStackTrace();
return false;
}
}
}
} }