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

Polish SimpUserRegistry related classes

Browse Source 
Issue: SPR-13800
This commit is contained in:
Rossen Stoyanchev
2016-03-09 16:16:50 -05:00
parent 8222ff465d
commit 6aa216afb6
9 changed files with 391 additions and 282 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-messaging/src/main/java/org/springframework/messaging/simp/config/AbstractMessageBrokerConfiguration.java
View File

@@ -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

285
spring-messaging/src/main/java/org/springframework/messaging/simp/user/MultiServerUserRegistry.java
View File

@@ -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<String, UserRegistrySnapshot> remoteRegistries = new ConcurrentHashMap<String, UserRegistrySnapshot>();
private final Map<String, UserRegistryDto> remoteRegistries =
new ConcurrentHashMap<String, UserRegistryDto>();
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<? extends ApplicationEvent> 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<SimpUser> getUsers() {
Set<SimpUser> result = new HashSet<SimpUser>(this.localRegistry.getUsers());
for (UserRegistryDto registry : this.remoteRegistries.values()) {
result.addAll(registry.getUsers().values());
Set<SimpUser> result = new HashSet<SimpUser>();
result.addAll(this.localRegistry.getUsers());
for (UserRegistrySnapshot registry : this.remoteRegistries.values()) {
result.addAll(registry.getUserMap().values());
}
return result;
}
@Override
public Set<SimpSubscription> findSubscriptions(SimpSubscriptionMatcher matcher) {
Set<SimpSubscription> result = new HashSet<SimpSubscription>(this.localRegistry.findSubscriptions(matcher));
for (UserRegistryDto registry : this.remoteRegistries.values()) {
Set<SimpSubscription> result = new HashSet<SimpSubscription>();
result.addAll(this.localRegistry.findSubscriptions(matcher));
for (UserRegistrySnapshot registry : this.remoteRegistries.values()) {
result.addAll(registry.findSubscriptions(matcher));
}
return result;
}
@Override
public boolean supportsEventType(Class<? extends ApplicationEvent> 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<Map.Entry<String, UserRegistryDto>> iterator = this.remoteRegistries.entrySet().iterator();
Iterator<Map.Entry<String, UserRegistrySnapshot>> iterator = this.remoteRegistries.entrySet().iterator();
while (iterator.hasNext()) {
Map.Entry<String, UserRegistryDto> entry = iterator.next();
if (now > entry.getValue().getExpirationTime()) {
Map.Entry<String, UserRegistrySnapshot> 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<String, SimpUserDto> users;
private Map<String, TransferSimpUser> 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<SimpUser> users = registry.getUsers();
this.users = new HashMap<String, SimpUserDto>(users.size());
this.users = new HashMap<String, TransferSimpUser>(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<String, SimpUserDto> users) {
public void setUserMap(Map<String, TransferSimpUser> users) {
this.users = users;
}
public Map<String, SimpUserDto> getUsers() {
public Map<String, TransferSimpUser> 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<SimpSubscription> findSubscriptions(SimpSubscriptionMatcher matcher) {
Set<SimpSubscription> result = new HashSet<SimpSubscription>();
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<SimpSessionDto> sessions;
private Set<TransferSimpSession> sessions;
public SimpUserDto() {
this.sessions = new HashSet<SimpSessionDto>(1);
/**
* Default constructor for JSON deserialization.
*/
public TransferSimpUser() {
this.sessions = new HashSet<TransferSimpSession>(1);
}
public SimpUserDto(SimpUser user) {
/**
* Constructor to create user from a local user.
*/
public TransferSimpUser(SimpUser user) {
this.name = user.getName();
Set<SimpSession> sessions = user.getSessions();
this.sessions = new HashSet<SimpSessionDto>(sessions.size());
this.sessions = new HashSet<TransferSimpSession>(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<SimpSessionDto> sessions) {
public void setSessions(Set<TransferSimpSession> sessions) {
this.sessions.addAll(sessions);
}
@@ -288,10 +323,10 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
return new HashSet<SimpSession>(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<SimpSubscriptionDto> subscriptions;
private final Set<TransferSimpSubscription> subscriptions;
public SimpSessionDto() {
this.subscriptions = new HashSet<SimpSubscriptionDto>(4);
/**
* Default constructor for JSON deserialization.
*/
public TransferSimpSession() {
this.subscriptions = new HashSet<TransferSimpSubscription>(4);
}
public SimpSessionDto(SimpSession session) {
/**
* Constructor to create DTO from the local user session.
*/
public TransferSimpSession(SimpSession session) {
this.id = session.getId();
Set<SimpSubscription> subscriptions = session.getSubscriptions();
this.subscriptions = new HashSet<SimpSubscriptionDto>(subscriptions.size());
this.subscriptions = new HashSet<TransferSimpSubscription>(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<SimpSubscriptionDto> subscriptions) {
public void setSubscriptions(Set<TransferSimpSubscription> subscriptions) {
this.subscriptions.addAll(subscriptions);
}
@@ -361,8 +405,8 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
return new HashSet<SimpSubscription>(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<? extends ApplicationEvent> 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;
}
}
}

37
spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserRegistryMessageHandler.java
View File

@@ -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.
* <p>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.
* <p>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;
}
/**

41
spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserSessionRegistryAdapter.java
View File

@@ -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.
*
* <p>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<String> sessionIds = this.delegate.getSessionIds(userName);
return (!CollectionUtils.isEmpty(sessionIds) ? new SimpleSimpUser(userName, sessionIds) : null);
Set<String> 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<String, SimpSession> sessions;
public SimpleSimpUser(String name, Set<String> sessionIds) {
public SimpUserAdapter(String name, Set<String> sessionIds) {
this.name = name;
this.sessions = new HashMap<String, SimpSession>(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;
}

105
spring-messaging/src/test/java/org/springframework/messaging/simp/user/MultiServerUserRegistryTests.java
View File

@@ -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<SimpUser> 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<SimpUser>(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<SimpSubscription> matches = this.multiServerRegistry.findSubscriptions(new SimpSubscriptionMatcher() {
@Override
public boolean match(SimpSubscription subscription) {
return subscription.getDestination().equals("/match");
}
});
assertEquals(3, this.registry.getUsers().size());
Set<SimpSubscription> matches = this.registry.findSubscriptions(s -> s.getDestination().equals("/match"));
assertEquals(2, matches.size());
Iterator<SimpSubscription> iterator = matches.iterator();
Set<String> 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());
}
}

17
spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpSession.java
View File

@@ -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;
}
}

27
spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpSubscription.java
View File

@@ -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;
}
}

15
spring-messaging/src/test/java/org/springframework/messaging/simp/user/TestSimpUser.java
View File

@@ -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;
}
}

136
spring-websocket/src/main/java/org/springframework/web/socket/messaging/DefaultSimpUserRegistry.java
View File

@@ -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<String, DefaultSimpUser> users = new ConcurrentHashMap<String, DefaultSimpUser>();
/* Primary lookup that holds all users and their sessions */
private final Map<String, LocalSimpUser> users = new ConcurrentHashMap<String, LocalSimpUser>();
private final Map<String, DefaultSimpSession> sessions = new ConcurrentHashMap<String, DefaultSimpSession>();
/* Secondary lookup across all sessions by id */
private final Map<String, LocalSimpSession> sessions = new ConcurrentHashMap<String, LocalSimpSession>();
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<SimpUser> getUsers() {
return new HashSet<SimpUser>(this.users.values());
}
public Set<SimpSubscription> findSubscriptions(SimpSubscriptionMatcher matcher) {
Set<SimpSubscription> result = new HashSet<SimpSubscription>();
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<? extends ApplicationEvent> 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<SimpUser> getUsers() {
return new HashSet<SimpUser>(this.users.values());
}
public Set<SimpSubscription> findSubscriptions(SimpSubscriptionMatcher matcher) {
Set<SimpSubscription> result = new HashSet<SimpSubscription>();
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<String, SimpSession> sessions =
private final Map<String, SimpSession> userSessions =
new ConcurrentHashMap<String, SimpSession>(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<SimpSession> getSessions() {
return new HashSet<SimpSession>(this.sessions.values());
return new HashSet<SimpSession>(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<String, SimpSubscription> subscriptions = new ConcurrentHashMap<String, SimpSubscription>(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;
}