Add SimpUserRegistry with multi-server support

This change introduces SimpUserRegistry exposing an API to access
information about connected users, their sessions, and subscriptions
with STOMP/WebSocket messaging. Provides are methods to access users
as well as a method to find subscriptions given a Matcher strategy.

The DefaultSimpUserRegistry implementation is also a
SmartApplicationListener which listesn for ApplicationContext events
when users connect, disconnect, subscribe, and unsubscribe to
destinations.

The MultiServerUserRegistry implementation is a composite that
aggregates user information from the local SimpUserRegistry as well
as snapshots of user  on remote application servers.

UserRegistryMessageHandler is used with MultiServerUserRegistry. It
broadcats user registry information through the broker and listens
for similar broadcasts from other servers. This must be enabled
explicitly when configuring the STOMP broker relay.

The existing UserSessionRegistry which was primiarly used internally
to resolve a user name to session id's has been deprecated and is no
longer used. If an application configures a custom UserSessionRegistr
still, it will be adapted accordingly to SimpUserRegistry but the
effect is rather limited (comparable to pre-existing functionality)
and will not work in multi-server scenarios.

Issue: SPR-12029

spring-messaging/src/test/java/org/springframework/messaging/simp/config/MessageBrokerConfigurationTests.java

@@ -16,6 +16,9 @@
 
 package org.springframework.messaging.simp.config;
 
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
@@ -51,8 +54,10 @@ import org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler;
 import org.springframework.messaging.simp.stomp.StompBrokerRelayMessageHandler;
 import org.springframework.messaging.simp.stomp.StompCommand;
 import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
+import org.springframework.messaging.simp.user.MultiServerUserRegistry;
+import org.springframework.messaging.simp.user.SimpUserRegistry;
 import org.springframework.messaging.simp.user.UserDestinationMessageHandler;
-import org.springframework.messaging.simp.user.UserSessionRegistry;
+import org.springframework.messaging.simp.user.UserRegistryMessageHandler;
 import org.springframework.messaging.support.AbstractSubscribableChannel;
 import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.messaging.support.ChannelInterceptorAdapter;
@@ -66,9 +71,6 @@ import org.springframework.validation.Errors;
 import org.springframework.validation.Validator;
 import org.springframework.validation.beanvalidation.OptionalValidatorFactoryBean;
 
-import static org.junit.Assert.*;
-import static org.mockito.Mockito.*;
-
 /**
  * Test fixture for {@link AbstractMessageBrokerConfiguration}.
  *
@@ -235,26 +237,6 @@ public class MessageBrokerConfigurationTests {
 		assertEquals("bar", new String((byte[]) message.getPayload()));
 	}
 
-	@Test
-	public void brokerChannelUsedByUserDestinationMessageHandler() {
-		TestChannel channel = this.simpleBrokerContext.getBean("brokerChannel", TestChannel.class);
-		UserDestinationMessageHandler messageHandler = this.simpleBrokerContext.getBean(UserDestinationMessageHandler.class);
-
-		this.simpleBrokerContext.getBean(UserSessionRegistry.class).registerSessionId("joe", "s1");
-
-		StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.SEND);
-		headers.setDestination("/user/joe/foo");
-		Message<?> message = MessageBuilder.createMessage(new byte[0], headers.getMessageHeaders());
-
-		messageHandler.handleMessage(message);
-
-		message = channel.messages.get(0);
-		headers = StompHeaderAccessor.wrap(message);
-
-		assertEquals(SimpMessageType.MESSAGE, headers.getMessageType());
-		assertEquals("/foo-users1", headers.getDestination());
-	}
-
 	@Test
 	public void brokerChannelCustomized() {
 		AbstractSubscribableChannel channel = this.customContext.getBean(
@@ -272,7 +254,7 @@ public class MessageBrokerConfigurationTests {
 
 	@Test
 	public void configureMessageConvertersDefault() {
-		AbstractMessageBrokerConfiguration config = new AbstractMessageBrokerConfiguration() {};
+		AbstractMessageBrokerConfiguration config = new BaseTestMessageBrokerConfig();
 		CompositeMessageConverter compositeConverter = config.brokerMessageConverter();
 
 		List<MessageConverter> converters = compositeConverter.getConverters();
@@ -305,7 +287,7 @@ public class MessageBrokerConfigurationTests {
 	@Test
 	public void configureMessageConvertersCustom() {
 		final MessageConverter testConverter = mock(MessageConverter.class);
-		AbstractMessageBrokerConfiguration config = new AbstractMessageBrokerConfiguration() {
+		AbstractMessageBrokerConfiguration config = new BaseTestMessageBrokerConfig() {
 			@Override
 			protected boolean configureMessageConverters(List<MessageConverter> messageConverters) {
 				messageConverters.add(testConverter);
@@ -323,7 +305,7 @@ public class MessageBrokerConfigurationTests {
 	public void configureMessageConvertersCustomAndDefault() {
 		final MessageConverter testConverter = mock(MessageConverter.class);
 
-		AbstractMessageBrokerConfiguration config = new AbstractMessageBrokerConfiguration() {
+		AbstractMessageBrokerConfiguration config = new BaseTestMessageBrokerConfig() {
 			@Override
 			protected boolean configureMessageConverters(List<MessageConverter> messageConverters) {
 				messageConverters.add(testConverter);
@@ -355,7 +337,7 @@ public class MessageBrokerConfigurationTests {
 
 	@Test
 	public void simpValidatorDefault() {
-		AbstractMessageBrokerConfiguration config = new AbstractMessageBrokerConfiguration() {};
+		AbstractMessageBrokerConfiguration config = new BaseTestMessageBrokerConfig() {};
 		config.setApplicationContext(new StaticApplicationContext());
 
 		assertThat(config.simpValidator(), Matchers.notNullValue());
@@ -365,7 +347,7 @@ public class MessageBrokerConfigurationTests {
 	@Test
 	public void simpValidatorCustom() {
 		final Validator validator = mock(Validator.class);
-		AbstractMessageBrokerConfiguration config = new AbstractMessageBrokerConfiguration() {
+		AbstractMessageBrokerConfiguration config = new BaseTestMessageBrokerConfig() {
 			@Override
 			public Validator getValidator() {
 				return validator;
@@ -379,7 +361,7 @@ public class MessageBrokerConfigurationTests {
 	public void simpValidatorMvc() {
 		StaticApplicationContext appCxt = new StaticApplicationContext();
 		appCxt.registerSingleton("mvcValidator", TestValidator.class);
-		AbstractMessageBrokerConfiguration config = new AbstractMessageBrokerConfiguration() {};
+		AbstractMessageBrokerConfiguration config = new BaseTestMessageBrokerConfig() {};
 		config.setApplicationContext(appCxt);
 
 		assertThat(config.simpValidator(), Matchers.notNullValue());
@@ -405,12 +387,35 @@ public class MessageBrokerConfigurationTests {
 	}
 
 	@Test
-	public void userDestinationBroadcast() throws Exception {
+	public void userBroadcasts() throws Exception {
+		SimpUserRegistry userRegistry = this.brokerRelayContext.getBean(SimpUserRegistry.class);
+		assertEquals(MultiServerUserRegistry.class, userRegistry.getClass());
+
+		UserDestinationMessageHandler handler1 = this.brokerRelayContext.getBean(UserDestinationMessageHandler.class);
+		assertEquals("/topic/unresolved-user-destination", handler1.getBroadcastDestination());
+
+		UserRegistryMessageHandler handler2 = this.brokerRelayContext.getBean(UserRegistryMessageHandler.class);
+		assertEquals("/topic/simp-user-registry", handler2.getBroadcastDestination());
+
 		StompBrokerRelayMessageHandler relay = this.brokerRelayContext.getBean(StompBrokerRelayMessageHandler.class);
-		UserDestinationMessageHandler userHandler = this.brokerRelayContext.getBean(UserDestinationMessageHandler.class);
-		assertEquals("/topic/unresolved", userHandler.getUserDestinationBroadcast());
 		assertNotNull(relay.getSystemSubscriptions());
-		assertSame(userHandler, relay.getSystemSubscriptions().get("/topic/unresolved"));
+		assertEquals(2, relay.getSystemSubscriptions().size());
+		assertSame(handler1, relay.getSystemSubscriptions().get("/topic/unresolved-user-destination"));
+		assertSame(handler2, relay.getSystemSubscriptions().get("/topic/simp-user-registry"));
+	}
+
+	@Test
+	public void userBroadcastsDisabledWithSimpleBroker() throws Exception {
+		SimpUserRegistry registry = this.simpleBrokerContext.getBean(SimpUserRegistry.class);
+		assertNotNull(registry);
+		assertNotEquals(MultiServerUserRegistry.class, registry.getClass());
+
+		UserDestinationMessageHandler handler = this.simpleBrokerContext.getBean(UserDestinationMessageHandler.class);
+		assertNull(handler.getBroadcastDestination());
+
+		String name = "userRegistryMessageHandler";
+		MessageHandler messageHandler = this.simpleBrokerContext.getBean(name, MessageHandler.class);
+		assertNotEquals(UserRegistryMessageHandler.class, messageHandler.getClass());
 	}
 
 
@@ -430,9 +435,17 @@ public class MessageBrokerConfigurationTests {
 		}
 	}
 
+	static class BaseTestMessageBrokerConfig extends AbstractMessageBrokerConfiguration {
+
+		@Override
+		protected SimpUserRegistry createLocalUserRegistry() {
+			return mock(SimpUserRegistry.class);
+		}
+	}
+
 	@SuppressWarnings("unused")
 	@Configuration
-	static class SimpleBrokerConfig extends AbstractMessageBrokerConfiguration {
+	static class SimpleBrokerConfig extends BaseTestMessageBrokerConfig {
 
 		@Bean
 		public TestController subscriptionController() {
@@ -463,17 +476,18 @@ public class MessageBrokerConfigurationTests {
 
 		@Override
 		public void configureMessageBroker(MessageBrokerRegistry registry) {
-			registry.enableStompBrokerRelay("/topic", "/queue").setAutoStartup(true);
-			registry.setUserDestinationBroadcast("/topic/unresolved");
+			registry.enableStompBrokerRelay("/topic", "/queue").setAutoStartup(true)
+					.setUserDestinationBroadcast("/topic/unresolved-user-destination")
+					.setUserRegistryBroadcast("/topic/simp-user-registry");
 		}
 	}
 
 	@Configuration
-	static class DefaultConfig extends AbstractMessageBrokerConfiguration {
+	static class DefaultConfig extends BaseTestMessageBrokerConfig {
 	}
 
 	@Configuration
-	static class CustomConfig extends AbstractMessageBrokerConfiguration {
+	static class CustomConfig extends BaseTestMessageBrokerConfig {
 
 		private ChannelInterceptor interceptor = new ChannelInterceptorAdapter() {};
 

spring-messaging/src/test/java/org/springframework/messaging/simp/config/StompBrokerRelayRegistrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -29,7 +29,8 @@ import org.springframework.messaging.simp.stomp.StompBrokerRelayMessageHandler;
 import static org.junit.Assert.*;
 
 /**
- * Unit tests for {@link org.springframework.messaging.simp.config.StompBrokerRelayRegistration}.
+ * Unit tests for
+ * {@link org.springframework.messaging.simp.config.StompBrokerRelayRegistration}.
  *
  * @author Rossen Stoyanchev
  */
@@ -39,15 +40,11 @@ public class StompBrokerRelayRegistrationTests {
 	@Test
 	public void test() {
 
-		SubscribableChannel clientInboundChannel = new StubMessageChannel();
-		MessageChannel clientOutboundChannel = new StubMessageChannel();
-		SubscribableChannel brokerChannel = new StubMessageChannel();
-
-		String[] destinationPrefixes = new String[] { "/foo", "/bar" };
-
-		StompBrokerRelayRegistration registration = new StompBrokerRelayRegistration(
-				clientInboundChannel, clientOutboundChannel, destinationPrefixes);
+		SubscribableChannel inChannel = new StubMessageChannel();
+		MessageChannel outChannel = new StubMessageChannel();
+		String[] prefixes = new String[] { "/foo", "/bar" };
 
+		StompBrokerRelayRegistration registration = new StompBrokerRelayRegistration(inChannel, outChannel, prefixes);
 		registration.setClientLogin("clientlogin");
 		registration.setClientPasscode("clientpasscode");
 		registration.setSystemLogin("syslogin");
@@ -56,18 +53,16 @@ public class StompBrokerRelayRegistrationTests {
 		registration.setSystemHeartbeatSendInterval(456);
 		registration.setVirtualHost("example.org");
 
-		StompBrokerRelayMessageHandler relayMessageHandler = registration.getMessageHandler(brokerChannel);
+		StompBrokerRelayMessageHandler handler = registration.getMessageHandler(new StubMessageChannel());
 
-		assertEquals(Arrays.asList(destinationPrefixes),
-				new ArrayList<String>(relayMessageHandler.getDestinationPrefixes()));
-
-		assertEquals("clientlogin", relayMessageHandler.getClientLogin());
-		assertEquals("clientpasscode", relayMessageHandler.getClientPasscode());
-		assertEquals("syslogin", relayMessageHandler.getSystemLogin());
-		assertEquals("syspasscode", relayMessageHandler.getSystemPasscode());
-		assertEquals(123, relayMessageHandler.getSystemHeartbeatReceiveInterval());
-		assertEquals(456, relayMessageHandler.getSystemHeartbeatSendInterval());
-		assertEquals("example.org", relayMessageHandler.getVirtualHost());
+		assertArrayEquals(prefixes, handler.getDestinationPrefixes().toArray(new String[2]));
+		assertEquals("clientlogin", handler.getClientLogin());
+		assertEquals("clientpasscode", handler.getClientPasscode());
+		assertEquals("syslogin", handler.getSystemLogin());
+		assertEquals("syspasscode", handler.getSystemPasscode());
+		assertEquals(123, handler.getSystemHeartbeatReceiveInterval());
+		assertEquals(456, handler.getSystemHeartbeatSendInterval());
+		assertEquals("example.org", handler.getVirtualHost());
 	}
 
 }

spring-messaging/src/test/java/org/springframework/messaging/simp/user/DefaultUserDestinationResolverTests.java

@@ -17,6 +17,9 @@
 package org.springframework.messaging.simp.user;
 
 import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+import java.security.Principal;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -36,35 +39,36 @@ import org.springframework.util.StringUtils;
  */
 public class DefaultUserDestinationResolverTests {
 
-	public static final String SESSION_ID = "123";
-
 	private DefaultUserDestinationResolver resolver;
 
-	private UserSessionRegistry registry;
-
-	private TestPrincipal user;
+	private SimpUserRegistry registry;
 
 
 	@Before
 	public void setup() {
-		this.user = new TestPrincipal("joe");
-		this.registry = new DefaultUserSessionRegistry();
-		this.registry.registerSessionId(this.user.getName(), SESSION_ID);
+
+		TestSimpUser simpUser = new TestSimpUser("joe");
+		simpUser.addSessions(new TestSimpSession("123"));
+
+		this.registry = mock(SimpUserRegistry.class);
+		when(this.registry.getUser("joe")).thenReturn(simpUser);
+
 		this.resolver = new DefaultUserDestinationResolver(this.registry);
 	}
 
-
 	@Test
 	public void handleSubscribe() {
+		TestPrincipal user = new TestPrincipal("joe");
 		String sourceDestination = "/user/queue/foo";
-		Message<?> message = createWith(SimpMessageType.SUBSCRIBE, this.user, SESSION_ID, sourceDestination);
+
+		Message<?> message = createMessage(SimpMessageType.SUBSCRIBE, user, "123", sourceDestination);
 		UserDestinationResult actual = this.resolver.resolveDestination(message);
 
 		assertEquals(sourceDestination, actual.getSourceDestination());
 		assertEquals(1, actual.getTargetDestinations().size());
 		assertEquals("/queue/foo-user123", actual.getTargetDestinations().iterator().next());
 		assertEquals(sourceDestination, actual.getSubscribeDestination());
-		assertEquals(this.user.getName(), actual.getUser());
+		assertEquals(user.getName(), actual.getUser());
 	}
 
 	// SPR-11325
@@ -72,32 +76,35 @@ public class DefaultUserDestinationResolverTests {
 	@Test
 	public void handleSubscribeOneUserMultipleSessions() {
 
-		this.registry.registerSessionId("joe", "456");
-		this.registry.registerSessionId("joe", "789");
+		TestSimpUser simpUser = new TestSimpUser("joe");
+		simpUser.addSessions(new TestSimpSession("123"), new TestSimpSession("456"));
+		when(this.registry.getUser("joe")).thenReturn(simpUser);
 
-		Message<?> message = createWith(SimpMessageType.SUBSCRIBE, this.user, SESSION_ID, "/user/queue/foo");
+		TestPrincipal user = new TestPrincipal("joe");
+		Message<?> message = createMessage(SimpMessageType.SUBSCRIBE, user, "456", "/user/queue/foo");
 		UserDestinationResult actual = this.resolver.resolveDestination(message);
 
 		assertEquals(1, actual.getTargetDestinations().size());
-		assertEquals("/queue/foo-user123", actual.getTargetDestinations().iterator().next());
+		assertEquals("/queue/foo-user456", actual.getTargetDestinations().iterator().next());
 	}
 
 	@Test
 	public void handleSubscribeNoUser() {
 		String sourceDestination = "/user/queue/foo";
-		Message<?> message = createWith(SimpMessageType.SUBSCRIBE, null, SESSION_ID, sourceDestination);
+		Message<?> message = createMessage(SimpMessageType.SUBSCRIBE, null, "123", sourceDestination);
 		UserDestinationResult actual = this.resolver.resolveDestination(message);
 
 		assertEquals(sourceDestination, actual.getSourceDestination());
 		assertEquals(1, actual.getTargetDestinations().size());
-		assertEquals("/queue/foo-user" + SESSION_ID, actual.getTargetDestinations().iterator().next());
+		assertEquals("/queue/foo-user" + "123", actual.getTargetDestinations().iterator().next());
 		assertEquals(sourceDestination, actual.getSubscribeDestination());
 		assertNull(actual.getUser());
 	}
 
 	@Test
 	public void handleUnsubscribe() {
-		Message<?> message = createWith(SimpMessageType.UNSUBSCRIBE, this.user, SESSION_ID, "/user/queue/foo");
+		TestPrincipal user = new TestPrincipal("joe");
+		Message<?> message = createMessage(SimpMessageType.UNSUBSCRIBE, user, "123", "/user/queue/foo");
 		UserDestinationResult actual = this.resolver.resolveDestination(message);
 
 		assertEquals(1, actual.getTargetDestinations().size());
@@ -106,32 +113,37 @@ public class DefaultUserDestinationResolverTests {
 
 	@Test
 	public void handleMessage() {
+		TestPrincipal user = new TestPrincipal("joe");
 		String sourceDestination = "/user/joe/queue/foo";
-		Message<?> message = createWith(SimpMessageType.MESSAGE, this.user, SESSION_ID, sourceDestination);
+		Message<?> message = createMessage(SimpMessageType.MESSAGE, user, "123", sourceDestination);
 		UserDestinationResult actual = this.resolver.resolveDestination(message);
 
 		assertEquals(sourceDestination, actual.getSourceDestination());
 		assertEquals(1, actual.getTargetDestinations().size());
 		assertEquals("/queue/foo-user123", actual.getTargetDestinations().iterator().next());
 		assertEquals("/user/queue/foo", actual.getSubscribeDestination());
-		assertEquals(this.user.getName(), actual.getUser());
+		assertEquals(user.getName(), actual.getUser());
 	}
 
 	// SPR-12444
+	
 	@Test
 	public void handleMessageToOtherUser() {
-		final String OTHER_SESSION_ID = "456";
-		final String OTHER_USER_NAME = "anna";
+		
+		TestSimpUser otherSimpUser = new TestSimpUser("anna");
+		otherSimpUser.addSessions(new TestSimpSession("456"));
+		when(this.registry.getUser("anna")).thenReturn(otherSimpUser);
+
+		TestPrincipal user = new TestPrincipal("joe");
+		TestPrincipal otherUser = new TestPrincipal("anna");
+		String sourceDestination = "/user/anna/queue/foo";
+		Message<?> message = createMessage(SimpMessageType.MESSAGE, user, "456", sourceDestination);
 
-		String sourceDestination = "/user/"+OTHER_USER_NAME+"/queue/foo";
-		TestPrincipal otherUser = new TestPrincipal(OTHER_USER_NAME);
-		this.registry.registerSessionId(otherUser.getName(), OTHER_SESSION_ID);
-		Message<?> message = createWith(SimpMessageType.MESSAGE, this.user, SESSION_ID, sourceDestination);
 		UserDestinationResult actual = this.resolver.resolveDestination(message);
 
 		assertEquals(sourceDestination, actual.getSourceDestination());
 		assertEquals(1, actual.getTargetDestinations().size());
-		assertEquals("/queue/foo-user" + OTHER_SESSION_ID, actual.getTargetDestinations().iterator().next());
+		assertEquals("/queue/foo-user456", actual.getTargetDestinations().iterator().next());
 		assertEquals("/user/queue/foo", actual.getSubscribeDestination());
 		assertEquals(otherUser.getName(), actual.getUser());
 	}
@@ -140,9 +152,14 @@ public class DefaultUserDestinationResolverTests {
 	public void handleMessageEncodedUserName() {
 
 		String userName = "http://joe.openid.example.org/";
-		this.registry.registerSessionId(userName, "openid123");
+
+		TestSimpUser simpUser = new TestSimpUser(userName);
+		simpUser.addSessions(new TestSimpSession("openid123"));
+		when(this.registry.getUser(userName)).thenReturn(simpUser);
+
 		String destination = "/user/" + StringUtils.replace(userName, "/", "%2F") + "/queue/foo";
-		Message<?> message = createWith(SimpMessageType.MESSAGE, this.user, null, destination);
+
+		Message<?> message = createMessage(SimpMessageType.MESSAGE, new TestPrincipal("joe"), null, destination);
 		UserDestinationResult actual = this.resolver.resolveDestination(message);
 
 		assertEquals(1, actual.getTargetDestinations().size());
@@ -151,8 +168,8 @@ public class DefaultUserDestinationResolverTests {
 
 	@Test
 	public void handleMessageWithNoUser() {
-		String sourceDestination = "/user/" + SESSION_ID + "/queue/foo";
-		Message<?> message = createWith(SimpMessageType.MESSAGE, null, SESSION_ID, sourceDestination);
+		String sourceDestination = "/user/" + "123" + "/queue/foo";
+		Message<?> message = createMessage(SimpMessageType.MESSAGE, null, "123", sourceDestination);
 		UserDestinationResult actual = this.resolver.resolveDestination(message);
 
 		assertEquals(sourceDestination, actual.getSourceDestination());
@@ -166,28 +183,28 @@ public class DefaultUserDestinationResolverTests {
 	public void ignoreMessage() {
 
 		// no destination
-		Message<?> message = createWith(SimpMessageType.MESSAGE, this.user, SESSION_ID, null);
+		TestPrincipal user = new TestPrincipal("joe");
+		Message<?> message = createMessage(SimpMessageType.MESSAGE, user, "123", null);
 		UserDestinationResult actual = this.resolver.resolveDestination(message);
 		assertNull(actual);
 
 		// not a user destination
-		message = createWith(SimpMessageType.MESSAGE, this.user, SESSION_ID, "/queue/foo");
+		message = createMessage(SimpMessageType.MESSAGE, user, "123", "/queue/foo");
 		actual = this.resolver.resolveDestination(message);
 		assertNull(actual);
 
 		// subscribe + not a user destination
-		message = createWith(SimpMessageType.SUBSCRIBE, this.user, SESSION_ID, "/queue/foo");
+		message = createMessage(SimpMessageType.SUBSCRIBE, user, "123", "/queue/foo");
 		actual = this.resolver.resolveDestination(message);
 		assertNull(actual);
 
 		// no match on message type
-		message = createWith(SimpMessageType.CONNECT, this.user, SESSION_ID, "user/joe/queue/foo");
+		message = createMessage(SimpMessageType.CONNECT, user, "123", "user/joe/queue/foo");
 		actual = this.resolver.resolveDestination(message);
 		assertNull(actual);
 	}
 
-
-	private Message<?> createWith(SimpMessageType type, TestPrincipal user, String sessionId, String destination) {
+	private Message<?> createMessage(SimpMessageType type, Principal user, String sessionId, String destination) {
 		SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.create(type);
 		if (destination != null) {
 			headers.setDestination(destination);

spring-messaging/src/test/java/org/springframework/messaging/simp/user/DefaultUserSessionRegistryTests.java

@@ -1,82 +0,0 @@
-/*
- * Copyright 2002-2015 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.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.messaging.simp.user;
-
-import static org.junit.Assert.*;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.LinkedHashSet;
-import java.util.List;
-
-import org.junit.Test;
-
-/**
- * Test fixture for
- * {@link org.springframework.messaging.simp.user.DefaultUserSessionRegistry}
- *
- * @author Rossen Stoyanchev
- * @since 4.0
- */
-public class DefaultUserSessionRegistryTests {
-
-	private static final String user = "joe";
-
-	private static final List<String> sessionIds = Arrays.asList("sess01", "sess02", "sess03");
-
-
-	@Test
-	public void addOneSessionId() {
-
-		DefaultUserSessionRegistry resolver = new DefaultUserSessionRegistry();
-		resolver.registerSessionId(user, sessionIds.get(0));
-
-		assertEquals(Collections.singleton(sessionIds.get(0)), resolver.getSessionIds(user));
-		assertSame(Collections.emptySet(), resolver.getSessionIds("jane"));
-	}
-
-	@Test
-	public void addMultipleSessionIds() {
-
-		DefaultUserSessionRegistry resolver = new DefaultUserSessionRegistry();
-		for (String sessionId : sessionIds) {
-			resolver.registerSessionId(user, sessionId);
-		}
-
-		assertEquals(new LinkedHashSet<>(sessionIds), resolver.getSessionIds(user));
-		assertEquals(Collections.<String>emptySet(), resolver.getSessionIds("jane"));
-	}
-
-	@Test
-	public void removeSessionIds() {
-
-		DefaultUserSessionRegistry resolver = new DefaultUserSessionRegistry();
-		for (String sessionId : sessionIds) {
-			resolver.registerSessionId(user, sessionId);
-		}
-
-		assertEquals(new LinkedHashSet<>(sessionIds), resolver.getSessionIds(user));
-
-		resolver.unregisterSessionId(user, sessionIds.get(1));
-		resolver.unregisterSessionId(user, sessionIds.get(2));
-		assertEquals(Collections.singleton(sessionIds.get(0)), resolver.getSessionIds(user));
-
-		resolver.unregisterSessionId(user, sessionIds.get(0));
-		assertSame(Collections.emptySet(), resolver.getSessionIds(user));
-	}
-
-}

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

@@ -0,0 +1,167 @@
+/*
+ * Copyright 2002-2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.messaging.simp.user;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.MappingJackson2MessageConverter;
+import org.springframework.messaging.converter.MessageConverter;
+
+/**
+ * Unit tests for {@link MultiServerUserRegistry}.
+ *
+ * @author Rossen Stoyanchev
+ */
+public class MultiServerUserRegistryTests {
+
+	private SimpUserRegistry localRegistry;
+
+	private MultiServerUserRegistry multiServerRegistry;
+
+	private MessageConverter converter;
+
+
+	@Before
+	public void setUp() throws Exception {
+		this.localRegistry = Mockito.mock(SimpUserRegistry.class);
+		this.multiServerRegistry = 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"));
+	}
+
+	@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));
+
+		MultiServerUserRegistry remoteRegistry = new MultiServerUserRegistry(remoteUserRegistry);
+		Message<?> message = this.converter.toMessage(remoteRegistry.getLocalRegistryDto(), null);
+
+		this.multiServerRegistry.addRemoteRegistryDto(message, this.converter, 20000);
+		assertEquals(1, this.multiServerRegistry.getUsers().size());
+
+		SimpUser user = this.multiServerRegistry.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());
+		assertEquals("/remote-dest", subscription.getDestination());
+	}
+
+	@Test
+	public void findUserFromRemoteRegistry() throws Exception {
+
+		TestSimpSubscription subscription1 = new TestSimpSubscription("sub1", "/match");
+		TestSimpSession session1 = new TestSimpSession("sess1");
+		session1.addSubscriptions(subscription1);
+		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");
+		user3.addSessions(session3);
+
+		SimpUserRegistry remoteUserRegistry = mock(SimpUserRegistry.class);
+		when(remoteUserRegistry.getUsers()).thenReturn(new HashSet<SimpUser>(Arrays.asList(user1, user2, user3)));
+
+		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(2, matches.size());
+
+		Iterator<SimpSubscription> iterator = matches.iterator();
+		Set<String> sessionIds = new HashSet<>(2);
+		sessionIds.add(iterator.next().getSession().getId());
+		sessionIds.add(iterator.next().getSession().getId());
+		assertEquals(new HashSet<>(Arrays.asList("sess1", "sess2")), sessionIds);
+	}
+
+	@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));
+
+		MultiServerUserRegistry remoteRegistry = new MultiServerUserRegistry(remoteUserRegistry);
+		Message<?> message = this.converter.toMessage(remoteRegistry.getLocalRegistryDto(), null);
+
+		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());
+	}
+
+}

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2002-2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.messaging.simp.user;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+
+
+public class TestSimpSession implements SimpSession {
+
+	private String id;
+
+	private TestSimpUser user;
+
+	private Set<SimpSubscription> subscriptions = new HashSet<>();
+
+
+	public TestSimpSession(String id) {
+		this.id = id;
+	}
+
+	@Override
+	public String getId() {
+		return id;
+	}
+
+	@Override
+	public TestSimpUser getUser() {
+		return user;
+	}
+
+	public void setUser(TestSimpUser user) {
+		this.user = user;
+	}
+
+	@Override
+	public Set<SimpSubscription> getSubscriptions() {
+		return subscriptions;
+	}
+
+	public void addSubscriptions(TestSimpSubscription... subscriptions) {
+		for (TestSimpSubscription subscription : subscriptions) {
+			subscription.setSession(this);
+			this.subscriptions.add(subscription);
+		}
+	}
+
+}

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

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2002-2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.messaging.simp.user;
+
+
+public class TestSimpSubscription implements SimpSubscription {
+
+	private String id;
+
+	private TestSimpSession session;
+
+	private String destination;
+
+
+	public TestSimpSubscription(String id, String destination) {
+		this.destination = destination;
+		this.id = id;
+	}
+
+	@Override
+	public String getId() {
+		return id;
+	}
+
+	@Override
+	public TestSimpSession getSession() {
+		return this.session;
+	}
+
+	public void setSession(TestSimpSession session) {
+		this.session = session;
+	}
+
+	@Override
+	public String getDestination() {
+		return destination;
+	}
+
+}

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2002-2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.messaging.simp.user;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+
+public class TestSimpUser implements SimpUser {
+
+	private String name;
+
+	private Map<String, SimpSession> sessions = new HashMap<>();
+
+
+	public TestSimpUser(String name) {
+		this.name = name;
+	}
+
+	@Override
+	public String getName() {
+		return name;
+	}
+
+	@Override
+	public Set<SimpSession> getSessions() {
+		return new HashSet<>(this.sessions.values());
+	}
+
+	@Override
+	public boolean hasSessions() {
+		return !this.sessions.isEmpty();
+	}
+
+	@Override
+	public SimpSession getSession(String sessionId) {
+		return this.sessions.get(sessionId);
+	}
+
+	public void addSessions(TestSimpSession... sessions) {
+		for (TestSimpSession session : sessions) {
+			session.setUser(this);
+			this.sessions.put(session.getId(), session);
+		}
+	}
+
+}

spring-messaging/src/test/java/org/springframework/messaging/simp/user/UserDestinationMessageHandlerTests.java

@@ -25,9 +25,7 @@ import java.nio.charset.Charset;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
-import org.mockito.Mock;
 import org.mockito.Mockito;
-import org.mockito.MockitoAnnotations;
 
 import org.springframework.messaging.Message;
 import org.springframework.messaging.StubMessageChannel;
@@ -50,16 +48,15 @@ public class UserDestinationMessageHandlerTests {
 
 	private UserDestinationMessageHandler handler;
 
-	private UserSessionRegistry registry;
+	private SimpUserRegistry registry;
 
-	@Mock
 	private SubscribableChannel brokerChannel;
 
 
 	@Before
 	public void setup() {
-		MockitoAnnotations.initMocks(this);
-		this.registry = new DefaultUserSessionRegistry();
+		this.registry = mock(SimpUserRegistry.class);
+		this.brokerChannel = mock(SubscribableChannel.class);
 		UserDestinationResolver resolver = new DefaultUserDestinationResolver(this.registry);
 		this.handler = new UserDestinationMessageHandler(new StubMessageChannel(), this.brokerChannel, resolver);
 	}
@@ -91,7 +88,9 @@ public class UserDestinationMessageHandlerTests {
 
 	@Test
 	public void handleMessage() {
-		this.registry.registerSessionId("joe", "123");
+		TestSimpUser simpUser = new TestSimpUser("joe");
+		simpUser.addSessions(new TestSimpSession("123"));
+		when(this.registry.getUser("joe")).thenReturn(simpUser);
 		given(this.brokerChannel.send(Mockito.any(Message.class))).willReturn(true);
 		this.handler.handleMessage(createWith(SimpMessageType.MESSAGE, "joe", "123", "/user/joe/queue/foo"));
 
@@ -105,7 +104,7 @@ public class UserDestinationMessageHandlerTests {
 
 	@Test
 	public void handleMessageWithoutActiveSession() {
-		this.handler.setUserDestinationBroadcast("/topic/unresolved");
+		this.handler.setBroadcastDestination("/topic/unresolved");
 		given(this.brokerChannel.send(Mockito.any(Message.class))).willReturn(true);
 		this.handler.handleMessage(createWith(SimpMessageType.MESSAGE, "joe", "123", "/user/joe/queue/foo"));
 
@@ -126,9 +125,11 @@ public class UserDestinationMessageHandlerTests {
 	@Test
 	public void handleMessageFromBrokerWithActiveSession() {
 
-		this.registry.registerSessionId("joe", "123");
+		TestSimpUser simpUser = new TestSimpUser("joe");
+		simpUser.addSessions(new TestSimpSession("123"));
+		when(this.registry.getUser("joe")).thenReturn(simpUser);
 
-		this.handler.setUserDestinationBroadcast("/topic/unresolved");
+		this.handler.setBroadcastDestination("/topic/unresolved");
 		given(this.brokerChannel.send(Mockito.any(Message.class))).willReturn(true);
 
 		StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.MESSAGE);
@@ -152,7 +153,7 @@ public class UserDestinationMessageHandlerTests {
 
 	@Test
 	public void handleMessageFromBrokerWithoutActiveSession() {
-		this.handler.setUserDestinationBroadcast("/topic/unresolved");
+		this.handler.setBroadcastDestination("/topic/unresolved");
 		given(this.brokerChannel.send(Mockito.any(Message.class))).willReturn(true);
 
 		StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.MESSAGE);

spring-messaging/src/test/java/org/springframework/messaging/simp/user/UserRegistryMessageHandlerTests.java

@@ -0,0 +1,183 @@
+/*
+ * Copyright 2002-2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.messaging.simp.user;
+
+import static org.junit.Assert.*;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.*;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.concurrent.ScheduledFuture;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.converter.MappingJackson2MessageConverter;
+import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
+import org.springframework.messaging.simp.SimpMessagingTemplate;
+import org.springframework.messaging.simp.broker.BrokerAvailabilityEvent;
+import org.springframework.scheduling.TaskScheduler;
+
+/**
+ * User tests for {@link UserRegistryMessageHandler}.
+ * @author Rossen Stoyanchev
+ */
+public class UserRegistryMessageHandlerTests {
+
+	private UserRegistryMessageHandler handler;
+
+	private SimpUserRegistry localRegistry;
+
+	private MultiServerUserRegistry multiServerRegistry;
+
+	private MessageConverter converter;
+
+	@Mock
+	private MessageChannel brokerChannel;
+
+	@Mock
+	private TaskScheduler taskScheduler;
+
+
+	@Before
+	public void setUp() throws Exception {
+
+		MockitoAnnotations.initMocks(this);
+
+		when(this.brokerChannel.send(any())).thenReturn(true);
+		this.converter = new MappingJackson2MessageConverter();
+
+		SimpMessagingTemplate brokerTemplate = new SimpMessagingTemplate(this.brokerChannel);
+		brokerTemplate.setMessageConverter(this.converter);
+
+		this.localRegistry = mock(SimpUserRegistry.class);
+		this.multiServerRegistry = new MultiServerUserRegistry(this.localRegistry);
+
+		this.handler = new UserRegistryMessageHandler(this.multiServerRegistry, brokerTemplate,
+				"/topic/simp-user-registry", this.taskScheduler);
+	}
+
+	@Test
+	public void brokerAvailableEvent() throws Exception {
+		Runnable runnable = getUserRegistryTask();
+		assertNotNull(runnable);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void brokerUnavailableEvent() throws Exception {
+
+		ScheduledFuture future = Mockito.mock(ScheduledFuture.class);
+		when(this.taskScheduler.scheduleWithFixedDelay(any(Runnable.class), any(Long.class))).thenReturn(future);
+
+		BrokerAvailabilityEvent event = new BrokerAvailabilityEvent(true, this);
+		this.handler.onApplicationEvent(event);
+		verifyNoMoreInteractions(future);
+
+		event = new BrokerAvailabilityEvent(false, this);
+		this.handler.onApplicationEvent(event);
+		verify(future).cancel(true);
+	}
+
+	@Test
+	public void broadcastRegistry() throws Exception {
+
+		TestSimpUser simpUser1 = new TestSimpUser("joe");
+		TestSimpUser simpUser2 = new TestSimpUser("jane");
+
+		simpUser1.addSessions(new TestSimpSession("123"));
+		simpUser1.addSessions(new TestSimpSession("456"));
+
+		HashSet<SimpUser> simpUsers = new HashSet<>(Arrays.asList(simpUser1, simpUser2));
+		when(this.localRegistry.getUsers()).thenReturn(simpUsers);
+
+		getUserRegistryTask().run();
+
+		ArgumentCaptor<Message> captor = ArgumentCaptor.forClass(Message.class);
+		verify(this.brokerChannel).send(captor.capture());
+
+		Message<?> message = captor.getValue();
+		assertNotNull(message);
+		MessageHeaders headers = message.getHeaders();
+		assertEquals("/topic/simp-user-registry", SimpMessageHeaderAccessor.getDestination(headers));
+
+		MultiServerUserRegistry remoteRegistry = new MultiServerUserRegistry(mock(SimpUserRegistry.class));
+		remoteRegistry.addRemoteRegistryDto(message, this.converter, 20000);
+		assertEquals(2, remoteRegistry.getUsers().size());
+		assertNotNull(remoteRegistry.getUser("joe"));
+		assertNotNull(remoteRegistry.getUser("jane"));
+	}
+
+	@Test
+	public void handleMessage() throws Exception {
+
+		TestSimpUser simpUser1 = new TestSimpUser("joe");
+		TestSimpUser simpUser2 = new TestSimpUser("jane");
+
+		simpUser1.addSessions(new TestSimpSession("123"));
+		simpUser2.addSessions(new TestSimpSession("456"));
+
+		HashSet<SimpUser> simpUsers = new HashSet<>(Arrays.asList(simpUser1, simpUser2));
+		SimpUserRegistry remoteUserRegistry = mock(SimpUserRegistry.class);
+		when(remoteUserRegistry.getUsers()).thenReturn(simpUsers);
+
+		MultiServerUserRegistry remoteRegistry = new MultiServerUserRegistry(remoteUserRegistry);
+		Message<?> message = this.converter.toMessage(remoteRegistry.getLocalRegistryDto(), null);
+
+		this.handler.handleMessage(message);
+
+		assertEquals(2, remoteRegistry.getUsers().size());
+		assertNotNull(this.multiServerRegistry.getUser("joe"));
+		assertNotNull(this.multiServerRegistry.getUser("jane"));
+	}
+
+	@Test
+	public void handleMessageFromOwnBroadcast() throws Exception {
+
+		TestSimpUser simpUser = new TestSimpUser("joe");
+		simpUser.addSessions(new TestSimpSession("123"));
+		when(this.localRegistry.getUsers()).thenReturn(Collections.singleton(simpUser));
+
+		assertEquals(1, this.multiServerRegistry.getUsers().size());
+
+		Message<?> message = this.converter.toMessage(this.multiServerRegistry.getLocalRegistryDto(), null);
+		this.multiServerRegistry.addRemoteRegistryDto(message, this.converter, 20000);
+		assertEquals(1, this.multiServerRegistry.getUsers().size());
+	}
+
+
+	private Runnable getUserRegistryTask() {
+		BrokerAvailabilityEvent event = new BrokerAvailabilityEvent(true, this);
+		this.handler.onApplicationEvent(event);
+
+		ArgumentCaptor<? extends Runnable> captor = ArgumentCaptor.forClass(Runnable.class);
+		verify(this.taskScheduler).scheduleWithFixedDelay(captor.capture(), eq(10000L));
+
+		return captor.getValue();
+	}
+
+}