From f0175bc42d823abe3cc210b731ea9b408bc65ed5 Mon Sep 17 00:00:00 2001 From: Juergen Hoeller Date: Tue, 7 Jul 2015 00:25:47 +0200 Subject: [PATCH] DefaultSubscriptionRegistry uses deep LinkedMultiValueMap copies between accessCache and updateCache Issue: SPR-13185 --- .../broker/DefaultSubscriptionRegistry.java | 53 +++++++++---------- 1 file changed, 26 insertions(+), 27 deletions(-) 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 index f1f7b830d5..9de465e23a 100644 --- 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 @@ -38,15 +38,13 @@ import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.simp.SimpMessageHeaderAccessor; +import org.springframework.messaging.support.MessageHeaderAccessor; import org.springframework.util.AntPathMatcher; import org.springframework.util.Assert; import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; import org.springframework.util.PathMatcher; -import static org.springframework.messaging.support.MessageHeaderAccessor.getAccessor; - - /** * Implementation of {@link SubscriptionRegistry} that stores subscriptions * in memory and uses a {@link org.springframework.util.PathMatcher PathMatcher} @@ -181,7 +179,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry { protected MultiValueMap findSubscriptionsInternal(String destination, Message message) { - MultiValueMap result = this.destinationCache.getSubscriptions(destination); + LinkedMultiValueMap result = this.destinationCache.getSubscriptions(destination); if (result != null) { return filterSubscriptions(result, message); } @@ -258,39 +256,39 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry { private class DestinationCache { /** Map from destination -> for fast look-ups */ - private final Map> accessCache = - new ConcurrentHashMap>(DEFAULT_CACHE_LIMIT); + private final Map> accessCache = + new ConcurrentHashMap>(DEFAULT_CACHE_LIMIT); /** Map from destination -> with locking */ @SuppressWarnings("serial") - private final Map> updateCache = - new LinkedHashMap>(DEFAULT_CACHE_LIMIT, 0.75f, true) { + private final Map> updateCache = + new LinkedHashMap>(DEFAULT_CACHE_LIMIT, 0.75f, true) { @Override - protected boolean removeEldestEntry(Map.Entry> eldest) { + protected boolean removeEldestEntry(Map.Entry> eldest) { return size() > getCacheLimit(); } }; - public MultiValueMap getSubscriptions(String destination) { + public LinkedMultiValueMap getSubscriptions(String destination) { return this.accessCache.get(destination); } - public void addSubscriptions(String destination, MultiValueMap subscriptions) { + public void addSubscriptions(String destination, LinkedMultiValueMap subscriptions) { synchronized (this.updateCache) { - this.updateCache.put(destination, new LinkedMultiValueMap(subscriptions)); + this.updateCache.put(destination, subscriptions.deepCopy()); this.accessCache.put(destination, subscriptions); } } public void updateAfterNewSubscription(String destination, String sessionId, String subsId) { synchronized (this.updateCache) { - for (Map.Entry> entry : this.updateCache.entrySet()) { + for (Map.Entry> entry : this.updateCache.entrySet()) { String cachedDestination = entry.getKey(); if (getPathMatcher().match(destination, cachedDestination)) { - MultiValueMap subs = entry.getValue(); + LinkedMultiValueMap subs = entry.getValue(); subs.add(sessionId, subsId); - this.accessCache.put(cachedDestination, new LinkedMultiValueMap(subs)); + this.accessCache.put(cachedDestination, subs.deepCopy()); } } } @@ -299,9 +297,9 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry { public void updateAfterRemovedSubscription(String sessionId, String subsId) { synchronized (this.updateCache) { Set destinationsToRemove = new HashSet(); - for (Map.Entry> entry : this.updateCache.entrySet()) { + for (Map.Entry> entry : this.updateCache.entrySet()) { String destination = entry.getKey(); - MultiValueMap sessionMap = entry.getValue(); + LinkedMultiValueMap sessionMap = entry.getValue(); List subscriptions = sessionMap.get(sessionId); if (subscriptions != null) { subscriptions.remove(subsId); @@ -312,7 +310,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry { destinationsToRemove.add(destination); } else { - this.accessCache.put(destination, new LinkedMultiValueMap(sessionMap)); + this.accessCache.put(destination, sessionMap.deepCopy()); } } } @@ -326,15 +324,15 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry { public void updateAfterRemovedSession(SessionSubscriptionInfo info) { synchronized (this.updateCache) { Set destinationsToRemove = new HashSet(); - for (Map.Entry> entry : this.updateCache.entrySet()) { + for (Map.Entry> entry : this.updateCache.entrySet()) { String destination = entry.getKey(); - MultiValueMap sessionMap = entry.getValue(); + LinkedMultiValueMap sessionMap = entry.getValue(); if (sessionMap.remove(info.getSessionId()) != null) { if (sessionMap.isEmpty()) { destinationsToRemove.add(destination); } else { - this.accessCache.put(destination, new LinkedMultiValueMap(sessionMap)); + this.accessCache.put(destination, sessionMap.deepCopy()); } } } @@ -351,6 +349,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry { } } + /** * Provide access to session subscriptions by sessionId. */ @@ -360,7 +359,6 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry { private final ConcurrentMap sessions = new ConcurrentHashMap(); - public SessionSubscriptionInfo getSubscriptions(String sessionId) { return this.sessions.get(sessionId); } @@ -394,6 +392,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry { } } + /** * Hold subscriptions for a session. */ @@ -407,7 +406,6 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry { private final Object monitor = new Object(); - public SessionSubscriptionInfo(String sessionId) { Assert.notNull(sessionId, "sessionId must not be null"); this.sessionId = sessionId; @@ -473,19 +471,18 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry { } } + private static class Subscription { private final String id; private final Expression selectorExpression; - public Subscription(String id, Expression selector) { this.id = id; this.selectorExpression = selector; } - public String getId() { return this.id; } @@ -496,10 +493,11 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry { @Override public String toString() { - return "Subscription id='" + this.id; + return "subscription(id=" + this.id + ")"; } } + private static class SimpMessageHeaderPropertyAccessor implements PropertyAccessor { @Override @@ -515,7 +513,8 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry { @Override public TypedValue read(EvaluationContext context, Object target, String name) throws AccessException { MessageHeaders headers = (MessageHeaders) target; - SimpMessageHeaderAccessor accessor = getAccessor(headers, SimpMessageHeaderAccessor.class); + SimpMessageHeaderAccessor accessor = + MessageHeaderAccessor.getAccessor(headers, SimpMessageHeaderAccessor.class); Object value; if ("destination".equalsIgnoreCase(name)) { value = accessor.getDestination();