diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/config/AbstractMessageBrokerConfiguration.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/config/AbstractMessageBrokerConfiguration.java index fa475b6811..0eb766c84c 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/config/AbstractMessageBrokerConfiguration.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/config/AbstractMessageBrokerConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -52,6 +52,7 @@ import org.springframework.messaging.support.ExecutorSubscribableChannel; import org.springframework.messaging.support.ImmutableMessageChannelInterceptor; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.MimeTypeUtils; import org.springframework.util.PathMatcher; @@ -315,8 +316,11 @@ public abstract class AbstractMessageBrokerConfiguration implements ApplicationC if (getBrokerRegistry().getUserRegistryBroadcast() == null) { return new NoOpMessageHandler(); } - return new UserRegistryMessageHandler(userRegistry(), brokerMessagingTemplate(), - getBrokerRegistry().getUserRegistryBroadcast(), messageBrokerTaskScheduler()); + SimpUserRegistry userRegistry = userRegistry(); + Assert.isInstanceOf(MultiServerUserRegistry.class, userRegistry); + return new UserRegistryMessageHandler((MultiServerUserRegistry) userRegistry, + brokerMessagingTemplate(), getBrokerRegistry().getUserRegistryBroadcast(), + messageBrokerTaskScheduler()); } // Expose alias for 4.1 compatibility diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/user/MultiServerUserRegistry.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/user/MultiServerUserRegistry.java index 386797c6e1..b8d7007f13 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/user/MultiServerUserRegistry.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/user/MultiServerUserRegistry.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -35,10 +35,11 @@ import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; /** - * A user registry that is a composite of the "local" user registry as well as - * snapshots of remote user registries. For use with - * {@link UserRegistryMessageHandler} which broadcasts periodically the content - * of the local registry and receives updates from other servers. + * {@code SimpUserRegistry} that looks up users in a "local" user registry as + * well as a set of "remote" user registries. The local registry is provided as + * a constructor argument while remote registries are updated via broadcasts + * handled by {@link UserRegistryMessageHandler} which in turn notifies this + * registry when updates are received. * * @author Rossen Stoyanchev * @since 4.2 @@ -50,10 +51,9 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati private final SimpUserRegistry localRegistry; - private final SmartApplicationListener listener; + private final Map remoteRegistries = new ConcurrentHashMap(); - private final Map remoteRegistries = - new ConcurrentHashMap(); + private final boolean delegateApplicationEvents; /** @@ -61,13 +61,11 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati */ public MultiServerUserRegistry(SimpUserRegistry localRegistry) { Assert.notNull(localRegistry, "'localRegistry' is required."); - this.localRegistry = localRegistry; - this.listener = (this.localRegistry instanceof SmartApplicationListener ? - (SmartApplicationListener) this.localRegistry : new NoOpSmartApplicationListener()); this.id = generateId(); + this.localRegistry = localRegistry; + this.delegateApplicationEvents = this.localRegistry instanceof SmartApplicationListener; } - private static String generateId() { String host; try { @@ -80,14 +78,43 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati } + @Override + public int getOrder() { + return (this.delegateApplicationEvents ? + ((SmartApplicationListener) this.localRegistry).getOrder() : Ordered.LOWEST_PRECEDENCE); + } + + // SmartApplicationListener methods + + @Override + public boolean supportsEventType(Class eventType) { + return (this.delegateApplicationEvents && + ((SmartApplicationListener) this.localRegistry).supportsEventType(eventType)); + } + + @Override + public boolean supportsSourceType(Class sourceType) { + return (this.delegateApplicationEvents && + ((SmartApplicationListener) this.localRegistry).supportsSourceType(sourceType)); + } + + @Override + public void onApplicationEvent(ApplicationEvent event) { + if (this.delegateApplicationEvents) { + ((SmartApplicationListener) this.localRegistry).onApplicationEvent(event); + } + } + + // SimpUserRegistry methods + @Override public SimpUser getUser(String userName) { SimpUser user = this.localRegistry.getUser(userName); if (user != null) { return user; } - for (UserRegistryDto registry : this.remoteRegistries.values()) { - user = registry.getUsers().get(userName); + for (UserRegistrySnapshot registry : this.remoteRegistries.values()) { + user = registry.getUserMap().get(userName); if (user != null) { return user; } @@ -97,94 +124,89 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati @Override public Set getUsers() { - Set result = new HashSet(this.localRegistry.getUsers()); - for (UserRegistryDto registry : this.remoteRegistries.values()) { - result.addAll(registry.getUsers().values()); + Set result = new HashSet(); + result.addAll(this.localRegistry.getUsers()); + for (UserRegistrySnapshot registry : this.remoteRegistries.values()) { + result.addAll(registry.getUserMap().values()); } return result; } @Override public Set findSubscriptions(SimpSubscriptionMatcher matcher) { - Set result = new HashSet(this.localRegistry.findSubscriptions(matcher)); - for (UserRegistryDto registry : this.remoteRegistries.values()) { + Set result = new HashSet(); + result.addAll(this.localRegistry.findSubscriptions(matcher)); + for (UserRegistrySnapshot registry : this.remoteRegistries.values()) { result.addAll(registry.findSubscriptions(matcher)); } return result; } - @Override - public boolean supportsEventType(Class eventType) { - return this.listener.supportsEventType(eventType); - } - - @Override - public boolean supportsSourceType(Class sourceType) { - return this.listener.supportsSourceType(sourceType); - } - - @Override - public void onApplicationEvent(ApplicationEvent event) { - this.listener.onApplicationEvent(event); - } - - @Override - public int getOrder() { - return this.listener.getOrder(); - } + // Internal methods for UserRegistryMessageHandler to manage broadcasts Object getLocalRegistryDto() { - return new UserRegistryDto(this.id, this.localRegistry); + return new UserRegistrySnapshot(this.id, this.localRegistry); } void addRemoteRegistryDto(Message message, MessageConverter converter, long expirationPeriod) { - UserRegistryDto registryDto = (UserRegistryDto) converter.fromMessage(message, UserRegistryDto.class); - if (registryDto != null && !registryDto.getId().equals(this.id)) { - long expirationTime = System.currentTimeMillis() + expirationPeriod; - registryDto.setExpirationTime(expirationTime); - registryDto.restoreParentReferences(); - this.remoteRegistries.put(registryDto.getId(), registryDto); + UserRegistrySnapshot registry = (UserRegistrySnapshot) converter.fromMessage(message, UserRegistrySnapshot.class); + if (registry != null && !registry.getId().equals(this.id)) { + registry.init(expirationPeriod); + this.remoteRegistries.put(registry.getId(), registry); } } void purgeExpiredRegistries() { long now = System.currentTimeMillis(); - Iterator> iterator = this.remoteRegistries.entrySet().iterator(); + Iterator> iterator = this.remoteRegistries.entrySet().iterator(); while (iterator.hasNext()) { - Map.Entry entry = iterator.next(); - if (now > entry.getValue().getExpirationTime()) { + Map.Entry entry = iterator.next(); + if (entry.getValue().isExpired(now)) { iterator.remove(); } } } + @Override public String toString() { return "local=[" + this.localRegistry + "], remote=" + this.remoteRegistries + "]"; } + /** + * Holds a copy of a SimpUserRegistry for the purpose of broadcasting to and + * receiving broadcasts from other application servers. + */ @SuppressWarnings("unused") - private static class UserRegistryDto { + private static class UserRegistrySnapshot { private String id; - private Map users; + private Map users; private long expirationTime; - public UserRegistryDto() { + + /** + * Default constructor for JSON deserialization. + */ + public UserRegistrySnapshot() { } - public UserRegistryDto(String id, SimpUserRegistry registry) { + /** + * Constructor to create DTO from a local user registry. + */ + public UserRegistrySnapshot(String id, SimpUserRegistry registry) { this.id = id; Set users = registry.getUsers(); - this.users = new HashMap(users.size()); + this.users = new HashMap(users.size()); for (SimpUser user : users) { - this.users.put(user.getName(), new SimpUserDto(user)); + this.users.put(user.getName(), new TransferSimpUser(user)); } } + public void setId(String id) { this.id = id; } @@ -193,18 +215,31 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati return this.id; } - public void setUsers(Map users) { + public void setUserMap(Map users) { this.users = users; } - public Map getUsers() { + public Map getUserMap() { return this.users; } + public boolean isExpired(long now) { + return (now > this.expirationTime); + } + + + public void init(long expirationPeriod) { + this.expirationTime = System.currentTimeMillis() + expirationPeriod; + for (TransferSimpUser user : this.users.values()) { + user.afterDeserialization(); + } + } + + public Set findSubscriptions(SimpSubscriptionMatcher matcher) { Set result = new HashSet(); - for (SimpUserDto user : this.users.values()) { - for (SimpSessionDto session : user.sessions) { + for (TransferSimpUser user : this.users.values()) { + for (TransferSimpSession session : user.sessions) { for (SimpSubscription subscription : session.subscriptions) { if (matcher.match(subscription)) { result.add(subscription); @@ -215,46 +250,46 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati return result; } - public void setExpirationTime(long expirationTime) { - this.expirationTime = expirationTime; - } - public long getExpirationTime() { - return this.expirationTime; - } - - private void restoreParentReferences() { - for (SimpUserDto user : this.users.values()) { - user.restoreParentReferences(); - } - } @Override public String toString() { return "id=" + this.id + ", users=" + this.users; } + } + /** + * SimpUser that can be (de)serialized and broadcast to other servers. + */ @SuppressWarnings("unused") - private static class SimpUserDto implements SimpUser { + private static class TransferSimpUser implements SimpUser { private String name; - private Set sessions; + private Set sessions; - public SimpUserDto() { - this.sessions = new HashSet(1); + + /** + * Default constructor for JSON deserialization. + */ + public TransferSimpUser() { + this.sessions = new HashSet(1); } - public SimpUserDto(SimpUser user) { + /** + * Constructor to create user from a local user. + */ + public TransferSimpUser(SimpUser user) { this.name = user.getName(); Set sessions = user.getSessions(); - this.sessions = new HashSet(sessions.size()); + this.sessions = new HashSet(sessions.size()); for (SimpSession session : sessions) { - this.sessions.add(new SimpSessionDto(session)); + this.sessions.add(new TransferSimpSession(session)); } } + public void setName(String name) { this.name = name; } @@ -270,8 +305,8 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati } @Override - public SimpSessionDto getSession(String sessionId) { - for (SimpSessionDto session : this.sessions) { + public SimpSession getSession(String sessionId) { + for (TransferSimpSession session : this.sessions) { if (session.getId().equals(sessionId)) { return session; } @@ -279,7 +314,7 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati return null; } - public void setSessions(Set sessions) { + public void setSessions(Set sessions) { this.sessions.addAll(sessions); } @@ -288,10 +323,10 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati return new HashSet(this.sessions); } - private void restoreParentReferences() { - for (SimpSessionDto session : this.sessions) { + private void afterDeserialization() { + for (TransferSimpSession session : this.sessions) { session.setUser(this); - session.restoreParentReferences(); + session.afterDeserialization(); } } @@ -311,26 +346,35 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati } } - + /** + * SimpSession that can be (de)serialized and broadcast to other servers. + */ @SuppressWarnings("unused") - private static class SimpSessionDto implements SimpSession { + private static class TransferSimpSession implements SimpSession { private String id; - private SimpUserDto user; + private TransferSimpUser user; - private final Set subscriptions; + private final Set subscriptions; - public SimpSessionDto() { - this.subscriptions = new HashSet(4); + + /** + * Default constructor for JSON deserialization. + */ + public TransferSimpSession() { + this.subscriptions = new HashSet(4); } - public SimpSessionDto(SimpSession session) { + /** + * Constructor to create DTO from the local user session. + */ + public TransferSimpSession(SimpSession session) { this.id = session.getId(); Set subscriptions = session.getSubscriptions(); - this.subscriptions = new HashSet(subscriptions.size()); + this.subscriptions = new HashSet(subscriptions.size()); for (SimpSubscription subscription : subscriptions) { - this.subscriptions.add(new SimpSubscriptionDto(subscription)); + this.subscriptions.add(new TransferSimpSubscription(subscription)); } } @@ -343,16 +387,16 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati return this.id; } - public void setUser(SimpUserDto user) { + public void setUser(TransferSimpUser user) { this.user = user; } @Override - public SimpUserDto getUser() { + public TransferSimpUser getUser() { return this.user; } - public void setSubscriptions(Set subscriptions) { + public void setSubscriptions(Set subscriptions) { this.subscriptions.addAll(subscriptions); } @@ -361,8 +405,8 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati return new HashSet(this.subscriptions); } - private void restoreParentReferences() { - for (SimpSubscriptionDto subscription : this.subscriptions) { + private void afterDeserialization() { + for (TransferSimpSubscription subscription : this.subscriptions) { subscription.setSession(this); } } @@ -383,24 +427,34 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati } } - + /** + * SimpSubscription that can be (de)serialized and broadcast to other servers. + */ @SuppressWarnings("unused") - private static class SimpSubscriptionDto implements SimpSubscription { + private static class TransferSimpSubscription implements SimpSubscription { private String id; - private SimpSessionDto session; + private TransferSimpSession session; private String destination; - public SimpSubscriptionDto() { + + /** + * Default constructor for JSON deserialization. + */ + public TransferSimpSubscription() { } - public SimpSubscriptionDto(SimpSubscription subscription) { + /** + * Constructor to create DTO from a local user subscription. + */ + public TransferSimpSubscription(SimpSubscription subscription) { this.id = subscription.getId(); this.destination = subscription.getDestination(); } + public void setId(String id) { this.id = id; } @@ -410,12 +464,12 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati return this.id; } - public void setSession(SimpSessionDto session) { + public void setSession(TransferSimpSession session) { this.session = session; } @Override - public SimpSessionDto getSession() { + public TransferSimpSession getSession() { return this.session; } @@ -452,27 +506,4 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati } } - - private static class NoOpSmartApplicationListener implements SmartApplicationListener { - - @Override - public boolean supportsEventType(Class eventType) { - return false; - } - - @Override - public boolean supportsSourceType(Class sourceType) { - return false; - } - - @Override - public void onApplicationEvent(ApplicationEvent event) { - } - - @Override - public int getOrder() { - return Ordered.LOWEST_PRECEDENCE; - } - } - } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserRegistryMessageHandler.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserRegistryMessageHandler.java index d5ac6e9008..983127cd84 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserRegistryMessageHandler.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserRegistryMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -31,9 +31,11 @@ import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; /** - * A MessageHandler that is subscribed to listen to broadcasts of user registry - * information from other application servers as well as to periodically - * broadcast the content of the local user registry. The aggregated information + * {@code MessageHandler} that handles user registry broadcasts from other + * application servers and periodically broadcasts the content of the local + * user registry. + * + * The aggregated information * is maintained in a {@link MultiServerUserRegistry}. * * @author Rossen Stoyanchev @@ -56,16 +58,22 @@ public class UserRegistryMessageHandler implements MessageHandler, ApplicationLi private long registryExpirationPeriod = 20 * 1000; - public UserRegistryMessageHandler(SimpUserRegistry userRegistry, SimpMessagingTemplate brokerTemplate, - String broadcastDestination, TaskScheduler scheduler) { + /** + * Constructor. + * @param userRegistry the registry with local and remote user registry information + * @param brokerTemplate template for broadcasting local registry information + * @param broadcastDestination the destination to broadcast to + * @param scheduler + */ + public UserRegistryMessageHandler(MultiServerUserRegistry userRegistry, + SimpMessagingTemplate brokerTemplate, String broadcastDestination, TaskScheduler scheduler) { Assert.notNull(userRegistry, "'userRegistry' is required"); - Assert.isInstanceOf(MultiServerUserRegistry.class, userRegistry); Assert.notNull(brokerTemplate, "'brokerTemplate' is required"); Assert.hasText(broadcastDestination, "'broadcastDestination' is required"); Assert.notNull(scheduler, "'scheduler' is required"); - this.userRegistry = (MultiServerUserRegistry) userRegistry; + this.userRegistry = userRegistry; this.brokerTemplate = brokerTemplate; this.broadcastDestination = broadcastDestination; this.scheduler = scheduler; @@ -73,20 +81,21 @@ public class UserRegistryMessageHandler implements MessageHandler, ApplicationLi /** - * Return the destination for broadcasting user registry information to. + * Return the configured destination for broadcasting UserRegistry information. */ public String getBroadcastDestination() { return this.broadcastDestination; } /** - * Configure how long before a remote registry snapshot expires. - *

By default this is set to 20000 (20 seconds). - * @param expirationPeriod the expiration period in milliseconds + * Configure the amount of time (in milliseconds) before a remote user + * registry snapshot is considered expired. + *

By default this is set to 20 seconds (value of 20000). + * @param milliseconds the expiration period in milliseconds */ @SuppressWarnings("unused") - public void setRegistryExpirationPeriod(long expirationPeriod) { - this.registryExpirationPeriod = expirationPeriod; + public void setRegistryExpirationPeriod(long milliseconds) { + this.registryExpirationPeriod = milliseconds; } /** diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserSessionRegistryAdapter.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserSessionRegistryAdapter.java index ffe4eb00d5..1a76435012 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserSessionRegistryAdapter.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserSessionRegistryAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,10 +22,18 @@ import java.util.HashSet; import java.util.Map; import java.util.Set; +import org.springframework.messaging.simp.config.AbstractMessageBrokerConfiguration; import org.springframework.util.CollectionUtils; /** - * A temporary adapter to allow use of deprecated {@link UserSessionRegistry}. + * An adapter that allows a {@code UserSessionRegistry}, which is deprecated in + * favor of {@code SimpUserRegistry}, to be used as a {@code SimpUserRegistry}. + * Due to the more limited information available, methods such as + * {@link #getUsers()} and {@link #findSubscriptions} are not supported. + * + *

As of 4.2 this adapter is used only in applications that explicitly + * register a custom {@code UserSessionRegistry} bean by overriding + * {@link AbstractMessageBrokerConfiguration#userSessionRegistry()}. * * @author Rossen Stoyanchev * @since 4.2 @@ -33,18 +41,18 @@ import org.springframework.util.CollectionUtils; @SuppressWarnings("deprecation") public class UserSessionRegistryAdapter implements SimpUserRegistry { - private final UserSessionRegistry delegate; + private final UserSessionRegistry userSessionRegistry; - public UserSessionRegistryAdapter(UserSessionRegistry delegate) { - this.delegate = delegate; + public UserSessionRegistryAdapter(UserSessionRegistry registry) { + this.userSessionRegistry = registry; } @Override public SimpUser getUser(String userName) { - Set sessionIds = this.delegate.getSessionIds(userName); - return (!CollectionUtils.isEmpty(sessionIds) ? new SimpleSimpUser(userName, sessionIds) : null); + Set sessionIds = this.userSessionRegistry.getSessionIds(userName); + return (!CollectionUtils.isEmpty(sessionIds) ? new SimpUserAdapter(userName, sessionIds) : null); } @Override @@ -58,17 +66,21 @@ public class UserSessionRegistryAdapter implements SimpUserRegistry { } - private static class SimpleSimpUser implements SimpUser { + /** + * Expose the only information available from a UserSessionRegistry (name + * and session id's) as a {@code SimpUser}. + */ + private static class SimpUserAdapter implements SimpUser { private final String name; private final Map sessions; - public SimpleSimpUser(String name, Set sessionIds) { + public SimpUserAdapter(String name, Set sessionIds) { this.name = name; this.sessions = new HashMap(sessionIds.size()); for (String sessionId : sessionIds) { - this.sessions.put(sessionId, new SimpleSimpSession(sessionId)); + this.sessions.put(sessionId, new SimpSessionAdapter(sessionId)); } } @@ -93,12 +105,15 @@ public class UserSessionRegistryAdapter implements SimpUserRegistry { } } - - private static class SimpleSimpSession implements SimpSession { + /** + * Expose the only information available from a UserSessionRegistry (session + * id's but no subscriptions) as a {@code SimpSession}. + */ + private static class SimpSessionAdapter implements SimpSession { private final String id; - public SimpleSimpSession(String id) { + public SimpSessionAdapter(String id) { this.id = id; } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/user/MultiServerUserRegistryTests.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/user/MultiServerUserRegistryTests.java index f4910601a0..44e74c6480 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/simp/user/MultiServerUserRegistryTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/simp/user/MultiServerUserRegistryTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -42,7 +42,7 @@ public class MultiServerUserRegistryTests { private SimpUserRegistry localRegistry; - private MultiServerUserRegistry multiServerRegistry; + private MultiServerUserRegistry registry; private MessageConverter converter; @@ -50,48 +50,47 @@ public class MultiServerUserRegistryTests { @Before public void setUp() throws Exception { this.localRegistry = Mockito.mock(SimpUserRegistry.class); - this.multiServerRegistry = new MultiServerUserRegistry(this.localRegistry); + this.registry = new MultiServerUserRegistry(this.localRegistry); this.converter = new MappingJackson2MessageConverter(); } @Test public void getUserFromLocalRegistry() throws Exception { - SimpUser user = Mockito.mock(SimpUser.class); Set users = Collections.singleton(user); when(this.localRegistry.getUsers()).thenReturn(users); when(this.localRegistry.getUser("joe")).thenReturn(user); - assertEquals(1, this.multiServerRegistry.getUsers().size()); - assertSame(user, this.multiServerRegistry.getUser("joe")); + assertEquals(1, this.registry.getUsers().size()); + assertSame(user, this.registry.getUser("joe")); } @Test public void getUserFromRemoteRegistry() throws Exception { - TestSimpSession remoteSession = new TestSimpSession("remote-sess"); - remoteSession.addSubscriptions(new TestSimpSubscription("remote-sub", "/remote-dest")); - TestSimpUser remoteUser = new TestSimpUser("joe"); - remoteUser.addSessions(remoteSession); - SimpUserRegistry remoteUserRegistry = mock(SimpUserRegistry.class); - when(remoteUserRegistry.getUsers()).thenReturn(Collections.singleton(remoteUser)); + // Prepare broadcast message from remote server + TestSimpUser testUser = new TestSimpUser("joe"); + TestSimpSession testSession = new TestSimpSession("remote-sess"); + testSession.addSubscriptions(new TestSimpSubscription("remote-sub", "/remote-dest")); + testUser.addSessions(testSession); + SimpUserRegistry testRegistry = mock(SimpUserRegistry.class); + when(testRegistry.getUsers()).thenReturn(Collections.singleton(testUser)); + Object registryDto = new MultiServerUserRegistry(testRegistry).getLocalRegistryDto(); + Message message = this.converter.toMessage(registryDto, null); - MultiServerUserRegistry remoteRegistry = new MultiServerUserRegistry(remoteUserRegistry); - Message message = this.converter.toMessage(remoteRegistry.getLocalRegistryDto(), null); + // Add remote registry + this.registry.addRemoteRegistryDto(message, this.converter, 20000); - this.multiServerRegistry.addRemoteRegistryDto(message, this.converter, 20000); - assertEquals(1, this.multiServerRegistry.getUsers().size()); - SimpUser user = this.multiServerRegistry.getUser("joe"); + assertEquals(1, this.registry.getUsers().size()); + SimpUser user = this.registry.getUser("joe"); assertNotNull(user); assertEquals(1, user.getSessions().size()); - SimpSession session = user.getSession("remote-sess"); assertNotNull(session); assertEquals("remote-sess", session.getId()); assertSame(user, session.getUser()); assertEquals(1, session.getSubscriptions().size()); - SimpSubscription subscription = session.getSubscriptions().iterator().next(); assertEquals("remote-sub", subscription.getId()); assertSame(session, subscription.getSession()); @@ -101,42 +100,31 @@ public class MultiServerUserRegistryTests { @Test public void findUserFromRemoteRegistry() throws Exception { - TestSimpSubscription subscription1 = new TestSimpSubscription("sub1", "/match"); - TestSimpSession session1 = new TestSimpSession("sess1"); - session1.addSubscriptions(subscription1); + // Prepare broadcast message from remote server TestSimpUser user1 = new TestSimpUser("joe"); - user1.addSessions(session1); - - TestSimpSubscription subscription2 = new TestSimpSubscription("sub1", "/match"); - TestSimpSession session2 = new TestSimpSession("sess2"); - session2.addSubscriptions(subscription2); TestSimpUser user2 = new TestSimpUser("jane"); - user2.addSessions(session2); - - TestSimpSubscription subscription3 = new TestSimpSubscription("sub1", "/not-a-match"); - TestSimpSession session3 = new TestSimpSession("sess3"); - session3.addSubscriptions(subscription3); TestSimpUser user3 = new TestSimpUser("jack"); + TestSimpSession session1 = new TestSimpSession("sess1"); + TestSimpSession session2 = new TestSimpSession("sess2"); + TestSimpSession session3 = new TestSimpSession("sess3"); + session1.addSubscriptions(new TestSimpSubscription("sub1", "/match")); + session2.addSubscriptions(new TestSimpSubscription("sub1", "/match")); + session3.addSubscriptions(new TestSimpSubscription("sub1", "/not-a-match")); + user1.addSessions(session1); + user2.addSessions(session2); user3.addSessions(session3); + SimpUserRegistry userRegistry = mock(SimpUserRegistry.class); + when(userRegistry.getUsers()).thenReturn(new HashSet<>(Arrays.asList(user1, user2, user3))); + Object registryDto = new MultiServerUserRegistry(userRegistry).getLocalRegistryDto(); + Message message = this.converter.toMessage(registryDto, null); - SimpUserRegistry remoteUserRegistry = mock(SimpUserRegistry.class); - when(remoteUserRegistry.getUsers()).thenReturn(new HashSet(Arrays.asList(user1, user2, user3))); + // Add remote registry + this.registry.addRemoteRegistryDto(message, this.converter, 20000); - MultiServerUserRegistry remoteRegistry = new MultiServerUserRegistry(remoteUserRegistry); - Message message = this.converter.toMessage(remoteRegistry.getLocalRegistryDto(), null); - - this.multiServerRegistry.addRemoteRegistryDto(message, this.converter, 20000); - assertEquals(3, this.multiServerRegistry.getUsers().size()); - - Set matches = this.multiServerRegistry.findSubscriptions(new SimpSubscriptionMatcher() { - @Override - public boolean match(SimpSubscription subscription) { - return subscription.getDestination().equals("/match"); - } - }); + assertEquals(3, this.registry.getUsers().size()); + Set matches = this.registry.findSubscriptions(s -> s.getDestination().equals("/match")); assertEquals(2, matches.size()); - Iterator iterator = matches.iterator(); Set sessionIds = new HashSet<>(2); sessionIds.add(iterator.next().getSession().getId()); @@ -147,20 +135,21 @@ public class MultiServerUserRegistryTests { @Test public void purgeExpiredRegistries() throws Exception { - TestSimpUser remoteUser = new TestSimpUser("joe"); - remoteUser.addSessions(new TestSimpSession("remote-sub")); - SimpUserRegistry remoteUserRegistry = mock(SimpUserRegistry.class); - when(remoteUserRegistry.getUsers()).thenReturn(Collections.singleton(remoteUser)); + // Prepare broadcast message from remote server + TestSimpUser testUser = new TestSimpUser("joe"); + testUser.addSessions(new TestSimpSession("remote-sub")); + SimpUserRegistry testRegistry = mock(SimpUserRegistry.class); + when(testRegistry.getUsers()).thenReturn(Collections.singleton(testUser)); + Object registryDto = new MultiServerUserRegistry(testRegistry).getLocalRegistryDto(); + Message message = this.converter.toMessage(registryDto, null); - MultiServerUserRegistry remoteRegistry = new MultiServerUserRegistry(remoteUserRegistry); - Message message = this.converter.toMessage(remoteRegistry.getLocalRegistryDto(), null); + // Add remote registry + this.registry.addRemoteRegistryDto(message, this.converter, -1); - long expirationPeriod = -1; - this.multiServerRegistry.addRemoteRegistryDto(message, this.converter, expirationPeriod); - assertEquals(1, this.multiServerRegistry.getUsers().size()); - this.multiServerRegistry.purgeExpiredRegistries(); - assertEquals(0, this.multiServerRegistry.getUsers().size()); + assertEquals(1, this.registry.getUsers().size()); + this.registry.purgeExpiredRegistries(); + assertEquals(0, this.registry.getUsers().size()); } } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpSession.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpSession.java index f608e03016..4ab378ad8f 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpSession.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpSession.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -59,4 +59,19 @@ public class TestSimpSession implements SimpSession { } } + @Override + public boolean equals(Object other) { + return (this == other || (other instanceof SimpSession && this.id.equals(((SimpSession) other).getId()))); + } + + @Override + public int hashCode() { + return this.id.hashCode(); + } + + @Override + public String toString() { + return "id=" + this.id + ", subscriptions=" + this.subscriptions; + } + } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpSubscription.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpSubscription.java index 28f92823ce..5a13c23453 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpSubscription.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpSubscription.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,6 +16,8 @@ package org.springframework.messaging.simp.user; +import org.springframework.util.ObjectUtils; + public class TestSimpSubscription implements SimpSubscription { private String id; @@ -49,4 +51,27 @@ public class TestSimpSubscription implements SimpSubscription { return destination; } + @Override + public boolean equals(Object other) { + if (this == other) { + return true; + } + if (!(other instanceof SimpSubscription)) { + return false; + } + SimpSubscription otherSubscription = (SimpSubscription) other; + return (ObjectUtils.nullSafeEquals(getSession(), otherSubscription.getSession()) && + this.id.equals(otherSubscription.getId())); + } + + @Override + public int hashCode() { + return this.id.hashCode() * 31 + ObjectUtils.nullSafeHashCode(getSession()); + } + + @Override + public String toString() { + return "destination=" + this.destination; + } + } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpUser.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpUser.java index 19d2099b2d..1607c20197 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpUser.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpUser.java @@ -59,4 +59,19 @@ public class TestSimpUser implements SimpUser { } } + @Override + public boolean equals(Object other) { + return (this == other || (other instanceof SimpUser && this.name.equals(((SimpUser) other).getName()))); + } + + @Override + public int hashCode() { + return this.name.hashCode(); + } + + @Override + public String toString() { + return "name=" + this.name + ", sessions=" + this.sessions; + } + } diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/messaging/DefaultSimpUserRegistry.java b/spring-websocket/src/main/java/org/springframework/web/socket/messaging/DefaultSimpUserRegistry.java index 4cd575bdbb..9ea63ea25f 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/messaging/DefaultSimpUserRegistry.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/messaging/DefaultSimpUserRegistry.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -37,52 +37,36 @@ import org.springframework.messaging.support.MessageHeaderAccessor; import org.springframework.util.Assert; /** - * Default, mutable, thread-safe implementation of {@link SimpUserRegistry} that - * listens ApplicationContext events of type {@link AbstractSubProtocolEvent} to - * keep track of user presence and subscription information. + * A default implementation of {@link SimpUserRegistry} that relies on + * {@link AbstractSubProtocolEvent} application context events to keep track of + * connected users and their subscriptions. * * @author Rossen Stoyanchev * @since 4.2 */ public class DefaultSimpUserRegistry implements SimpUserRegistry, SmartApplicationListener { - private final Map users = new ConcurrentHashMap(); + /* Primary lookup that holds all users and their sessions */ + private final Map users = new ConcurrentHashMap(); - private final Map sessions = new ConcurrentHashMap(); + /* Secondary lookup across all sessions by id */ + private final Map sessions = new ConcurrentHashMap(); + + private final Object sessionLock = new Object(); @Override - public SimpUser getUser(String userName) { - return this.users.get(userName); + public int getOrder() { + return Ordered.LOWEST_PRECEDENCE; } - @Override - public Set getUsers() { - return new HashSet(this.users.values()); - } - - public Set findSubscriptions(SimpSubscriptionMatcher matcher) { - Set result = new HashSet(); - for (DefaultSimpSession session : this.sessions.values()) { - for (SimpSubscription subscription : session.subscriptions.values()) { - if (matcher.match(subscription)) { - result.add(subscription); - } - } - } - return result; - } + // SmartApplicationListener methods @Override public boolean supportsEventType(Class eventType) { return AbstractSubProtocolEvent.class.isAssignableFrom(eventType); } - @Override - public boolean supportsSourceType(Class sourceType) { - return true; - } - @Override public void onApplicationEvent(ApplicationEvent event) { @@ -92,7 +76,7 @@ public class DefaultSimpUserRegistry implements SimpUserRegistry, SmartApplicati String sessionId = accessor.getSessionId(); if (event instanceof SessionSubscribeEvent) { - DefaultSimpSession session = this.sessions.get(sessionId); + LocalSimpSession session = this.sessions.get(sessionId); if (session != null) { String id = accessor.getSubscriptionId(); String destination = accessor.getDestination(); @@ -108,23 +92,22 @@ public class DefaultSimpUserRegistry implements SimpUserRegistry, SmartApplicati if (user instanceof DestinationUserNameProvider) { name = ((DestinationUserNameProvider) user).getDestinationUserName(); } - synchronized (this) { - DefaultSimpUser simpUser = this.users.get(name); + synchronized (this.sessionLock) { + LocalSimpUser simpUser = this.users.get(name); if (simpUser == null) { - simpUser = new DefaultSimpUser(name, sessionId); + simpUser = new LocalSimpUser(name); this.users.put(name, simpUser); } - else { - simpUser.addSession(sessionId); - } - this.sessions.put(sessionId, (DefaultSimpSession) simpUser.getSession(sessionId)); + LocalSimpSession session = new LocalSimpSession(sessionId, simpUser); + simpUser.addSession(session); + this.sessions.put(sessionId, session); } } else if (event instanceof SessionDisconnectEvent) { - synchronized (this) { - DefaultSimpSession session = this.sessions.remove(sessionId); + synchronized (this.sessionLock) { + LocalSimpSession session = this.sessions.remove(sessionId); if (session != null) { - DefaultSimpUser user = session.getUser(); + LocalSimpUser user = session.getUser(); user.removeSession(sessionId); if (!user.hasSessions()) { this.users.remove(user.getName()); @@ -133,7 +116,7 @@ public class DefaultSimpUserRegistry implements SimpUserRegistry, SmartApplicati } } else if (event instanceof SessionUnsubscribeEvent) { - DefaultSimpSession session = this.sessions.get(sessionId); + LocalSimpSession session = this.sessions.get(sessionId); if (session != null) { String subscriptionId = accessor.getSubscriptionId(); session.removeSubscription(subscriptionId); @@ -142,28 +125,52 @@ public class DefaultSimpUserRegistry implements SimpUserRegistry, SmartApplicati } @Override - public int getOrder() { - return Ordered.LOWEST_PRECEDENCE; + public boolean supportsSourceType(Class sourceType) { + return true; } + // SimpUserRegistry methods + + @Override + public SimpUser getUser(String userName) { + return this.users.get(userName); + } + + @Override + public Set getUsers() { + return new HashSet(this.users.values()); + } + + public Set findSubscriptions(SimpSubscriptionMatcher matcher) { + Set result = new HashSet(); + for (LocalSimpSession session : this.sessions.values()) { + for (SimpSubscription subscription : session.subscriptions.values()) { + if (matcher.match(subscription)) { + result.add(subscription); + } + } + } + return result; + } + + @Override public String toString() { return "users=" + this.users; } - private static class DefaultSimpUser implements SimpUser { + + private static class LocalSimpUser implements SimpUser { private final String name; - private final Map sessions = + private final Map userSessions = new ConcurrentHashMap(1); - public DefaultSimpUser(String userName, String sessionId) { + public LocalSimpUser(String userName) { Assert.notNull(userName); - Assert.notNull(sessionId); this.name = userName; - this.sessions.put(sessionId, new DefaultSimpSession(sessionId, this)); } @Override @@ -173,26 +180,25 @@ public class DefaultSimpUserRegistry implements SimpUserRegistry, SmartApplicati @Override public boolean hasSessions() { - return !this.sessions.isEmpty(); + return !this.userSessions.isEmpty(); } @Override public SimpSession getSession(String sessionId) { - return (sessionId != null ? this.sessions.get(sessionId) : null); + return (sessionId != null ? this.userSessions.get(sessionId) : null); } @Override public Set getSessions() { - return new HashSet(this.sessions.values()); + return new HashSet(this.userSessions.values()); } - void addSession(String sessionId) { - DefaultSimpSession session = new DefaultSimpSession(sessionId, this); - this.sessions.put(sessionId, session); + void addSession(SimpSession session) { + this.userSessions.put(session.getId(), session); } void removeSession(String sessionId) { - this.sessions.remove(sessionId); + this.userSessions.remove(sessionId); } @Override @@ -213,20 +219,20 @@ public class DefaultSimpUserRegistry implements SimpUserRegistry, SmartApplicati @Override public String toString() { - return "name=" + this.name + ", sessions=" + this.sessions; + return "name=" + this.name + ", sessions=" + this.userSessions; } } - private static class DefaultSimpSession implements SimpSession { + private static class LocalSimpSession implements SimpSession { private final String id; - private final DefaultSimpUser user; + private final LocalSimpUser user; private final Map subscriptions = new ConcurrentHashMap(4); - public DefaultSimpSession(String id, DefaultSimpUser user) { + public LocalSimpSession(String id, LocalSimpUser user) { Assert.notNull(id); Assert.notNull(user); this.id = id; @@ -239,7 +245,7 @@ public class DefaultSimpUserRegistry implements SimpUserRegistry, SmartApplicati } @Override - public DefaultSimpUser getUser() { + public LocalSimpUser getUser() { return this.user; } @@ -249,7 +255,7 @@ public class DefaultSimpUserRegistry implements SimpUserRegistry, SmartApplicati } void addSubscription(String id, String destination) { - this.subscriptions.put(id, new DefaultSimpSubscription(id, destination, this)); + this.subscriptions.put(id, new LocalSimpSubscription(id, destination, this)); } void removeSubscription(String id) { @@ -278,16 +284,16 @@ public class DefaultSimpUserRegistry implements SimpUserRegistry, SmartApplicati } } - private static class DefaultSimpSubscription implements SimpSubscription { + private static class LocalSimpSubscription implements SimpSubscription { private final String id; - private final DefaultSimpSession session; + private final LocalSimpSession session; private final String destination; - public DefaultSimpSubscription(String id, String destination, DefaultSimpSession session) { + public LocalSimpSubscription(String id, String destination, LocalSimpSession session) { Assert.notNull(id); Assert.hasText(destination); Assert.notNull(session); @@ -302,7 +308,7 @@ public class DefaultSimpUserRegistry implements SimpUserRegistry, SmartApplicati } @Override - public DefaultSimpSession getSession() { + public LocalSimpSession getSession() { return this.session; }