From 7f9b5c0515662397e0b9f936e1c22bcc724b9ade Mon Sep 17 00:00:00 2001 From: Rob Winch Date: Mon, 15 Dec 2014 16:13:03 -0600 Subject: [PATCH] Ensure Redis Configured to Send Keyspace Notifications Previously there was a possibility that Session to WebSocket mapping was leaked if keyspace notifications were not enabled in Redis. To resolve this the RedisHttpSessionConfiguration now ensures that Redis is configured to enable Keyspace notifications. Fixes gh-76 gh-81 --- .../config/EmbeddedRedisConfiguration.java | 22 ++- .../sample/EmbeddedRedisConfiguration.java | 22 ++- .../sample/EmbeddedRedisConfiguration.java | 18 +- .../sample/EmbeddedRedisConfiguration.java | 22 ++- .../sample/config/EmbeddedRedisConfig.java | 22 ++- ...edisOperationsSessionRepositoryITests.java | 95 ++++++++-- ...ttpSessionExpireSessionDestroyedTests.java | 177 ++++++++++++++++++ .../data/redis/SessionMessageListener.java | 16 +- .../http/RedisHttpSessionConfiguration.java | 57 +++++- .../handler/WebSocketRegistryListener.java | 24 ++- .../redis/SessionMessageListenerTests.java | 23 ++- ...KeyspaceNotificationsInitializerTests.java | 151 +++++++++++++++ .../WebSocketRegistryListenerTests.java | 15 ++ 13 files changed, 617 insertions(+), 47 deletions(-) create mode 100644 spring-session/src/integration-test/java/org/springframework/session/data/redis/config/annotation/web/http/EnableRedisHttpSessionExpireSessionDestroyedTests.java create mode 100644 spring-session/src/test/java/org/springframework/session/data/redis/config/annotation/web/http/EnableRedisKeyspaceNotificationsInitializerTests.java diff --git a/samples/boot/src/main/java/sample/config/EmbeddedRedisConfiguration.java b/samples/boot/src/main/java/sample/config/EmbeddedRedisConfiguration.java index 27166516..38958db6 100644 --- a/samples/boot/src/main/java/sample/config/EmbeddedRedisConfiguration.java +++ b/samples/boot/src/main/java/sample/config/EmbeddedRedisConfiguration.java @@ -15,8 +15,12 @@ package sample.config; * the License. */ +import org.springframework.beans.BeansException; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import redis.clients.jedis.Protocol; @@ -33,11 +37,17 @@ import redis.embedded.RedisServer; public class EmbeddedRedisConfiguration { @Bean - public RedisServerBean redisServer() { + public static RedisServerBean redisServer() { return new RedisServerBean(); } - class RedisServerBean implements InitializingBean, DisposableBean { + /** + * Implements BeanDefinitionRegistryPostProcessor to ensure this Bean + * is initialized before any other Beans. Specifically, we want to ensure + * that the Redis Server is started before RedisHttpSessionConfiguration + * attempts to enable Keyspace notifications. + */ + static class RedisServerBean implements InitializingBean, DisposableBean, BeanDefinitionRegistryPostProcessor { private RedisServer redisServer; public void afterPropertiesSet() throws Exception { @@ -50,5 +60,11 @@ public class EmbeddedRedisConfiguration { redisServer.stop(); } } + + @Override + public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {} + + @Override + public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {} } -} +} \ No newline at end of file diff --git a/samples/security/src/main/java/sample/EmbeddedRedisConfiguration.java b/samples/security/src/main/java/sample/EmbeddedRedisConfiguration.java index a9b695b0..f44603eb 100644 --- a/samples/security/src/main/java/sample/EmbeddedRedisConfiguration.java +++ b/samples/security/src/main/java/sample/EmbeddedRedisConfiguration.java @@ -14,8 +14,12 @@ package sample; * License for the specific language governing permissions and limitations under * the License. */ +import org.springframework.beans.BeansException; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -33,11 +37,17 @@ import redis.embedded.RedisServer; public class EmbeddedRedisConfiguration { @Bean - public RedisServerBean redisServer() { + public static RedisServerBean redisServer() { return new RedisServerBean(); } - class RedisServerBean implements InitializingBean, DisposableBean { + /** + * Implements BeanDefinitionRegistryPostProcessor to ensure this Bean + * is initialized before any other Beans. Specifically, we want to ensure + * that the Redis Server is started before RedisHttpSessionConfiguration + * attempts to enable Keyspace notifications. + */ + static class RedisServerBean implements InitializingBean, DisposableBean, BeanDefinitionRegistryPostProcessor { private RedisServer redisServer; public void afterPropertiesSet() throws Exception { @@ -50,5 +60,11 @@ public class EmbeddedRedisConfiguration { redisServer.stop(); } } + + @Override + public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {} + + @Override + public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {} } -} +} \ No newline at end of file diff --git a/samples/users/src/main/java/sample/EmbeddedRedisConfiguration.java b/samples/users/src/main/java/sample/EmbeddedRedisConfiguration.java index ae263faa..b1db4048 100644 --- a/samples/users/src/main/java/sample/EmbeddedRedisConfiguration.java +++ b/samples/users/src/main/java/sample/EmbeddedRedisConfiguration.java @@ -14,8 +14,12 @@ package sample; * License for the specific language governing permissions and limitations under * the License. */ +import org.springframework.beans.BeansException; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -37,7 +41,13 @@ public class EmbeddedRedisConfiguration { return new RedisServerBean(); } - class RedisServerBean implements InitializingBean, DisposableBean { + /** + * Implements BeanDefinitionRegistryPostProcessor to ensure this Bean + * is initialized before any other Beans. Specifically, we want to ensure + * that the Redis Server is started before RedisHttpSessionConfiguration + * attempts to enable Keyspace notifications. + */ + class RedisServerBean implements InitializingBean, DisposableBean, BeanDefinitionRegistryPostProcessor { private RedisServer redisServer; @@ -51,5 +61,11 @@ public class EmbeddedRedisConfiguration { redisServer.stop(); } } + + @Override + public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {} + + @Override + public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {} } } diff --git a/samples/web/src/main/java/sample/EmbeddedRedisConfiguration.java b/samples/web/src/main/java/sample/EmbeddedRedisConfiguration.java index ae263faa..64df07e6 100644 --- a/samples/web/src/main/java/sample/EmbeddedRedisConfiguration.java +++ b/samples/web/src/main/java/sample/EmbeddedRedisConfiguration.java @@ -14,8 +14,12 @@ package sample; * License for the specific language governing permissions and limitations under * the License. */ +import org.springframework.beans.BeansException; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -33,11 +37,17 @@ import redis.embedded.RedisServer; public class EmbeddedRedisConfiguration { @Bean - public RedisServerBean redisServer() { + public static RedisServerBean redisServer() { return new RedisServerBean(); } - class RedisServerBean implements InitializingBean, DisposableBean { + /** + * Implements BeanDefinitionRegistryPostProcessor to ensure this Bean + * is initialized before any other Beans. Specifically, we want to ensure + * that the Redis Server is started before RedisHttpSessionConfiguration + * attempts to enable Keyspace notifications. + */ + static class RedisServerBean implements InitializingBean, DisposableBean, BeanDefinitionRegistryPostProcessor { private RedisServer redisServer; @@ -51,5 +61,11 @@ public class EmbeddedRedisConfiguration { redisServer.stop(); } } + + @Override + public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {} + + @Override + public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {} } -} +} \ No newline at end of file diff --git a/samples/websocket/src/main/java/sample/config/EmbeddedRedisConfig.java b/samples/websocket/src/main/java/sample/config/EmbeddedRedisConfig.java index 131b7030..01e05749 100644 --- a/samples/websocket/src/main/java/sample/config/EmbeddedRedisConfig.java +++ b/samples/websocket/src/main/java/sample/config/EmbeddedRedisConfig.java @@ -15,8 +15,12 @@ */ package sample.config; +import org.springframework.beans.BeansException; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -30,11 +34,17 @@ import redis.embedded.RedisServer; public class EmbeddedRedisConfig { @Bean - public RedisServerBean redisServer() { + public static RedisServerBean redisServer() { return new RedisServerBean(); } - class RedisServerBean implements InitializingBean, DisposableBean { + /** + * Implements BeanDefinitionRegistryPostProcessor to ensure this Bean + * is initialized before any other Beans. Specifically, we want to ensure + * that the Redis Server is started before RedisHttpSessionConfiguration + * attempts to enable Keyspace notifications. + */ + static class RedisServerBean implements InitializingBean, DisposableBean, BeanDefinitionRegistryPostProcessor { private RedisServer redisServer; @@ -48,5 +58,11 @@ public class EmbeddedRedisConfig { redisServer.stop(); } } + + @Override + public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {} + + @Override + public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {} } -} +} \ No newline at end of file diff --git a/spring-session/src/integration-test/java/org/springframework/session/data/redis/RedisOperationsSessionRepositoryITests.java b/spring-session/src/integration-test/java/org/springframework/session/data/redis/RedisOperationsSessionRepositoryITests.java index 4192f709..860cadb4 100644 --- a/spring-session/src/integration-test/java/org/springframework/session/data/redis/RedisOperationsSessionRepositoryITests.java +++ b/spring-session/src/integration-test/java/org/springframework/session/data/redis/RedisOperationsSessionRepositoryITests.java @@ -5,14 +5,21 @@ import static org.fest.assertions.Assertions.assertThat; import java.io.IOException; import java.net.ServerSocket; -import org.junit.After; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor; +import org.springframework.context.ApplicationListener; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; +import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.security.authentication.UsernamePasswordAuthenticationToken; import org.springframework.security.core.Authentication; import org.springframework.security.core.authority.AuthorityUtils; @@ -21,9 +28,11 @@ import org.springframework.security.core.context.SecurityContextHolder; import org.springframework.session.Session; import org.springframework.session.SessionRepository; import org.springframework.session.data.redis.config.annotation.web.http.EnableRedisHttpSession; +import org.springframework.session.events.SessionDestroyedEvent; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import redis.clients.jedis.Protocol; import redis.embedded.RedisServer; @RunWith(SpringJUnit4ClassRunner.class) @@ -34,19 +43,18 @@ public class RedisOperationsSessionRepositoryITests { @Autowired private SessionRepository repository; - @Before - public void setup() throws IOException { - redisServer = new RedisServer(getPort()); - redisServer.start(); - } + @Autowired + private SessionDestroyedEventRegistry registry; - @After - public void shutdown() throws InterruptedException { - redisServer.stop(); + private final Object lock = new Object(); + + @Before + public void setup() { + registry.setLock(lock); } @Test - public void saves() { + public void saves() throws InterruptedException { S toSave = repository.createSession(); toSave.setAttribute("a", "b"); Authentication toSaveToken = new UsernamePasswordAuthenticationToken("user","password", AuthorityUtils.createAuthorityList("ROLE_USER")); @@ -65,6 +73,10 @@ public class RedisOperationsSessionRepositoryITests { repository.delete(toSave.getId()); assertThat(repository.getSession(toSave.getId())).isNull(); + synchronized (lock) { + lock.wait(3000); + } + assertThat(registry.receivedEvent()).isTrue(); } @Test @@ -86,6 +98,27 @@ public class RedisOperationsSessionRepositoryITests { assertThat(session.getAttribute("1")).isEqualTo("2"); } + static class SessionDestroyedEventRegistry implements ApplicationListener { + private boolean receivedEvent; + private Object lock; + + @Override + public void onApplicationEvent(SessionDestroyedEvent event) { + receivedEvent = true; + synchronized (lock) { + lock.notifyAll(); + } + } + + public boolean receivedEvent() { + return receivedEvent; + } + + public void setLock(Object lock) { + this.lock = lock; + } + } + @Configuration @EnableRedisHttpSession static class Config { @@ -96,6 +129,46 @@ public class RedisOperationsSessionRepositoryITests { factory.setUsePool(false); return factory; } + + @Bean + public static RedisServerBean redisServer() { + return new RedisServerBean(); + } + + @Bean + public SessionDestroyedEventRegistry sessionDestroyedEventRegistry() { + return new SessionDestroyedEventRegistry(); + } + + /** + * Implements BeanDefinitionRegistryPostProcessor to ensure this Bean + * is initialized before any other Beans. Specifically, we want to ensure + * that the Redis Server is started before RedisHttpSessionConfiguration + * attempts to enable Keyspace notifications. + */ + static class RedisServerBean implements InitializingBean, DisposableBean, BeanDefinitionRegistryPostProcessor { + private RedisServer redisServer; + + + @Override + public void afterPropertiesSet() throws Exception { + redisServer = new RedisServer(getPort()); + redisServer.start(); + } + + @Override + public void destroy() throws Exception { + if(redisServer != null) { + redisServer.stop(); + } + } + + @Override + public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {} + + @Override + public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {} + } } private static Integer availablePort; @@ -106,6 +179,6 @@ public class RedisOperationsSessionRepositoryITests { availablePort = socket.getLocalPort(); socket.close(); } - return availablePort; + return Protocol.DEFAULT_PORT; } } \ No newline at end of file diff --git a/spring-session/src/integration-test/java/org/springframework/session/data/redis/config/annotation/web/http/EnableRedisHttpSessionExpireSessionDestroyedTests.java b/spring-session/src/integration-test/java/org/springframework/session/data/redis/config/annotation/web/http/EnableRedisHttpSessionExpireSessionDestroyedTests.java new file mode 100644 index 00000000..f7d8bdac --- /dev/null +++ b/spring-session/src/integration-test/java/org/springframework/session/data/redis/config/annotation/web/http/EnableRedisHttpSessionExpireSessionDestroyedTests.java @@ -0,0 +1,177 @@ +/* + * Copyright 2002-2014 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.session.data.redis.config.annotation.web.http; + + +import static org.fest.assertions.Assertions.assertThat; + +import java.io.IOException; +import java.net.ServerSocket; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor; +import org.springframework.context.ApplicationListener; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; +import org.springframework.scheduling.annotation.EnableScheduling; +import org.springframework.security.authentication.UsernamePasswordAuthenticationToken; +import org.springframework.security.core.Authentication; +import org.springframework.security.core.authority.AuthorityUtils; +import org.springframework.security.core.context.SecurityContext; +import org.springframework.security.core.context.SecurityContextHolder; +import org.springframework.session.Session; +import org.springframework.session.SessionRepository; +import org.springframework.session.data.redis.config.annotation.web.http.EnableRedisHttpSession; +import org.springframework.session.events.SessionDestroyedEvent; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import redis.clients.jedis.Protocol; +import redis.embedded.RedisServer; + +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class EnableRedisHttpSessionExpireSessionDestroyedTests { + private RedisServer redisServer; + + @Autowired + private SessionRepository repository; + + @Autowired + private SessionDestroyedEventRegistry registry; + + private final Object lock = new Object(); + + @Before + public void setup() { + registry.setLock(lock); + } + + @Test + public void expireFiresSessionDestroyedEvent() throws InterruptedException { + S toSave = repository.createSession(); + toSave.setAttribute("a", "b"); + Authentication toSaveToken = new UsernamePasswordAuthenticationToken("user","password", AuthorityUtils.createAuthorityList("ROLE_USER")); + SecurityContext toSaveContext = SecurityContextHolder.createEmptyContext(); + toSaveContext.setAuthentication(toSaveToken); + toSave.setAttribute("SPRING_SECURITY_CONTEXT", toSaveContext); + + repository.save(toSave); + + synchronized (lock) { + lock.wait(1100); + } + if(!registry.receivedEvent()) { + // Redis makes no guarantees on when an expired event will be fired + // we can ensure it gets fired by trying to get the session + repository.getSession(toSave.getId()); + } + assertThat(registry.receivedEvent()).isTrue(); + } + + static class SessionDestroyedEventRegistry implements ApplicationListener { + private boolean receivedEvent; + private Object lock; + + @Override + public void onApplicationEvent(SessionDestroyedEvent event) { + receivedEvent = true; + synchronized (lock) { + lock.notifyAll(); + } + } + + public boolean receivedEvent() { + return receivedEvent; + } + + public void setLock(Object lock) { + this.lock = lock; + } + } + + @Configuration + @EnableRedisHttpSession(maxInactiveIntervalInSeconds = 1) + static class Config { + @Bean + public JedisConnectionFactory connectionFactory() throws Exception { + JedisConnectionFactory factory = new JedisConnectionFactory(); + factory.setPort(getPort()); + factory.setUsePool(false); + return factory; + } + + @Bean + public static RedisServerBean redisServer() { + return new RedisServerBean(); + } + + @Bean + public SessionDestroyedEventRegistry sessionDestroyedEventRegistry() { + return new SessionDestroyedEventRegistry(); + } + + /** + * Implements BeanDefinitionRegistryPostProcessor to ensure this Bean + * is initialized before any other Beans. Specifically, we want to ensure + * that the Redis Server is started before RedisHttpSessionConfiguration + * attempts to enable Keyspace notifications. + */ + static class RedisServerBean implements InitializingBean, DisposableBean, BeanDefinitionRegistryPostProcessor { + private RedisServer redisServer; + + + @Override + public void afterPropertiesSet() throws Exception { + redisServer = new RedisServer(getPort()); + redisServer.start(); + } + + @Override + public void destroy() throws Exception { + if(redisServer != null) { + redisServer.stop(); + } + } + + @Override + public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {} + + @Override + public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {} + } + } + + private static Integer availablePort; + + private static int getPort() throws IOException { + if(availablePort == null) { + ServerSocket socket = new ServerSocket(0); + availablePort = socket.getLocalPort(); + socket.close(); + } + return Protocol.DEFAULT_PORT; + } +} diff --git a/spring-session/src/main/java/org/springframework/session/data/redis/SessionMessageListener.java b/spring-session/src/main/java/org/springframework/session/data/redis/SessionMessageListener.java index 8cb1d632..90cce2bf 100644 --- a/spring-session/src/main/java/org/springframework/session/data/redis/SessionMessageListener.java +++ b/spring-session/src/main/java/org/springframework/session/data/redis/SessionMessageListener.java @@ -47,18 +47,22 @@ public class SessionMessageListener implements MessageListener { } public void onMessage(Message message, byte[] pattern) { + byte[] messageChannel = message.getChannel(); byte[] messageBody = message.getBody(); - if(messageBody == null) { + if(messageChannel == null || messageBody == null) { + return; + } + String channel = new String(messageChannel); + if(!(channel.endsWith(":del") || channel.endsWith(":expired"))) { return; } String body = new String(messageBody); - if(!("del".equals(body) || "expired".equals(body))) { + if(!body.startsWith("spring:session:sessions:")) { return; } - String channel = new String(message.getChannel()); - int beginIndex = channel.lastIndexOf(":") + 1; - int endIndex = channel.length(); - String sessionId = channel.substring(beginIndex, endIndex); + int beginIndex = body.lastIndexOf(":") + 1; + int endIndex = body.length(); + String sessionId = body.substring(beginIndex, endIndex); publishEvent(new SessionDestroyedEvent(this, sessionId)); } diff --git a/spring-session/src/main/java/org/springframework/session/data/redis/config/annotation/web/http/RedisHttpSessionConfiguration.java b/spring-session/src/main/java/org/springframework/session/data/redis/config/annotation/web/http/RedisHttpSessionConfiguration.java index 3948c12d..0527363d 100644 --- a/spring-session/src/main/java/org/springframework/session/data/redis/config/annotation/web/http/RedisHttpSessionConfiguration.java +++ b/spring-session/src/main/java/org/springframework/session/data/redis/config/annotation/web/http/RedisHttpSessionConfiguration.java @@ -15,9 +15,12 @@ */ package org.springframework.session.data.redis.config.annotation.web.http; +import java.util.Arrays; +import java.util.List; import java.util.Map; import org.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationEventPublisher; @@ -27,11 +30,13 @@ import org.springframework.context.annotation.ImportAware; import org.springframework.core.annotation.AnnotationAttributes; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.core.type.AnnotationMetadata; +import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.listener.PatternTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.session.ExpiringSession; import org.springframework.session.SessionRepository; import org.springframework.session.data.redis.RedisOperationsSessionRepository; @@ -51,6 +56,7 @@ import org.springframework.util.ClassUtils; * @see EnableRedisHttpSession */ @Configuration +@EnableScheduling public class RedisHttpSessionConfiguration implements ImportAware, BeanClassLoaderAware { private ClassLoader beanClassLoader; @@ -68,7 +74,7 @@ public class RedisHttpSessionConfiguration implements ImportAware, BeanClassLoad RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); container.addMessageListener(redisSessionMessageListener(), - new PatternTopic("__keyspace@0__:spring:session:sessions:*")); + Arrays.asList(new PatternTopic("__keyevent@*:del"),new PatternTopic("__keyevent@*:expired"))); return container; } @@ -136,6 +142,55 @@ public class RedisHttpSessionConfiguration implements ImportAware, BeanClassLoad this.httpSessionStrategy = httpSessionStrategy; } + @Bean + public EnableRedisKeyspaceNotificationsInitializer enableRedisKeyspaceNotificationsInitializer(RedisConnectionFactory connectionFactory) { + return new EnableRedisKeyspaceNotificationsInitializer(connectionFactory); + } + + /** + * Ensures that Redis is configured to send keyspace notifications. This is important to ensure that expiration and + * deletion of sessions trigger SessionDestroyedEvents. Without the SessionDestroyedEvent resources may not get + * cleaned up properly. For example, the mapping of the Session to WebSocket connections may not get cleaned up. + */ + static class EnableRedisKeyspaceNotificationsInitializer implements InitializingBean { + static final String CONFIG_NOTIFY_KEYSPACE_EVENTS = "notify-keyspace-events"; + + private final RedisConnectionFactory connectionFactory; + + EnableRedisKeyspaceNotificationsInitializer(RedisConnectionFactory connectionFactory) { + this.connectionFactory = connectionFactory; + } + + + @Override + public void afterPropertiesSet() throws Exception { + RedisConnection connection = connectionFactory.getConnection(); + String notifyOptions = getNotifyOptions(connection); + String customizedNotifyOptions = notifyOptions; + if(!customizedNotifyOptions.contains("E")) { + customizedNotifyOptions += "E"; + } + boolean A = customizedNotifyOptions.contains("A"); + if(!(A || customizedNotifyOptions.contains("g"))) { + customizedNotifyOptions += "g"; + } + if(!(A || customizedNotifyOptions.contains("x"))) { + customizedNotifyOptions += "x"; + } + if(!notifyOptions.equals(customizedNotifyOptions)) { + connection.setConfig(CONFIG_NOTIFY_KEYSPACE_EVENTS, customizedNotifyOptions); + } + } + + private String getNotifyOptions(RedisConnection connection) { + List config = connection.getConfig(CONFIG_NOTIFY_KEYSPACE_EVENTS); + if(config.size() < 2) { + return ""; + } + return config.get(1); + } + } + /* (non-Javadoc) * @see org.springframework.beans.factory.BeanClassLoaderAware#setBeanClassLoader(java.lang.ClassLoader) diff --git a/spring-session/src/main/java/org/springframework/session/web/socket/handler/WebSocketRegistryListener.java b/spring-session/src/main/java/org/springframework/session/web/socket/handler/WebSocketRegistryListener.java index f99ad8cf..b3341785 100644 --- a/spring-session/src/main/java/org/springframework/session/web/socket/handler/WebSocketRegistryListener.java +++ b/spring-session/src/main/java/org/springframework/session/web/socket/handler/WebSocketRegistryListener.java @@ -73,8 +73,8 @@ public final class WebSocketRegistryListener implements ApplicationListener sessions = httpSessionIdToWsSessions.get(sessionId); + private void registerWsSession(String httpSessionId, WebSocketSession wsSession) { + Map sessions = httpSessionIdToWsSessions.get(httpSessionId); if(sessions == null) { sessions = new ConcurrentHashMap(); - httpSessionIdToWsSessions.putIfAbsent(sessionId, sessions); - sessions = httpSessionIdToWsSessions.get(sessionId); + httpSessionIdToWsSessions.putIfAbsent(httpSessionId, sessions); + sessions = httpSessionIdToWsSessions.get(httpSessionId); } sessions.put(wsSession.getId(), wsSession); } - private void closeWsSessions(String sessionId) { - Map sessionsToClose = httpSessionIdToWsSessions.remove(sessionId); + private void closeWsSessions(String httpSessionId) { + Map sessionsToClose = httpSessionIdToWsSessions.remove(httpSessionId); if(sessionsToClose == null) { return; } if(logger.isDebugEnabled()) { - logger.debug("Closing WebSocket connections associated to expired HTTP Session " + sessionId); + logger.debug("Closing WebSocket connections associated to expired HTTP Session " + httpSessionId); } for(WebSocketSession toClose : sessionsToClose.values()) { try { diff --git a/spring-session/src/test/java/org/springframework/session/data/redis/SessionMessageListenerTests.java b/spring-session/src/test/java/org/springframework/session/data/redis/SessionMessageListenerTests.java index a02f7ffd..a1b564d5 100644 --- a/spring-session/src/test/java/org/springframework/session/data/redis/SessionMessageListenerTests.java +++ b/spring-session/src/test/java/org/springframework/session/data/redis/SessionMessageListenerTests.java @@ -72,7 +72,7 @@ public class SessionMessageListenerTests { @Test public void onMessageDel() throws Exception { - mockMessage("del","spring:sessions:session:123"); + mockMessage("__keyevent@0__:del", "spring:session:sessions:123"); listener.onMessage(message, pattern); @@ -82,7 +82,7 @@ public class SessionMessageListenerTests { @Test public void onMessageSource() throws Exception { - mockMessage("del","spring:sessions:session:123"); + mockMessage("__keyevent@0__:del","spring:session:sessions:123"); listener.onMessage(message, pattern); @@ -92,7 +92,7 @@ public class SessionMessageListenerTests { @Test public void onMessageExpired() throws Exception { - mockMessage("expired","spring:sessions:session:543"); + mockMessage("__keyevent@0__:expired","spring:session:sessions:543"); listener.onMessage(message, pattern); @@ -102,7 +102,16 @@ public class SessionMessageListenerTests { @Test public void onMessageHset() throws Exception { - mockMessage("hset","spring:sessions:session:123"); + mockMessage("__keyevent@0__:hset","spring:session:sessions:123"); + + listener.onMessage(message, pattern); + + verifyZeroInteractions(eventPublisher); + } + + @Test + public void onMessageWrongKeyPrefix() throws Exception { + mockMessage("__keyevent@0__:del","spring:session:sessionsNo:123"); listener.onMessage(message, pattern); @@ -111,7 +120,7 @@ public class SessionMessageListenerTests { @Test public void onMessageRename() throws Exception { - mockMessage("rename","spring:sessions:session:123"); + mockMessage("__keyevent@0__:rename","spring:session:sessions:123"); listener.onMessage(message, pattern); @@ -120,7 +129,7 @@ public class SessionMessageListenerTests { @Test public void onMessageEventPublisherErrorCaught() throws Exception { - mockMessage("del","spring:sessions:session:123"); + mockMessage("__keyevent@0__:del","spring:session:sessions:123"); doThrow(new IllegalStateException("Test Exceptions are caught")).when(eventPublisher).publishEvent(any(ApplicationEvent.class)); listener.onMessage(message, pattern); @@ -128,7 +137,7 @@ public class SessionMessageListenerTests { verify(eventPublisher).publishEvent(any(ApplicationEvent.class)); } - private void mockMessage(String body, String channel) throws UnsupportedEncodingException { + private void mockMessage(String channel, String body) throws UnsupportedEncodingException { when(message.getBody()).thenReturn(bytes(body)); when(message.getChannel()).thenReturn(bytes(channel)); } diff --git a/spring-session/src/test/java/org/springframework/session/data/redis/config/annotation/web/http/EnableRedisKeyspaceNotificationsInitializerTests.java b/spring-session/src/test/java/org/springframework/session/data/redis/config/annotation/web/http/EnableRedisKeyspaceNotificationsInitializerTests.java new file mode 100644 index 00000000..7456cab0 --- /dev/null +++ b/spring-session/src/test/java/org/springframework/session/data/redis/config/annotation/web/http/EnableRedisKeyspaceNotificationsInitializerTests.java @@ -0,0 +1,151 @@ +package org.springframework.session.data.redis.config.annotation.web.http; + +import static org.fest.assertions.Assertions.assertThat; +import static org.mockito.Matchers.anyString; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.*; +import static org.springframework.session.data.redis.config.annotation.web.http.RedisHttpSessionConfiguration.*; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisConnectionFactory; + +import java.util.Arrays; + +@RunWith(MockitoJUnitRunner.class) +public class EnableRedisKeyspaceNotificationsInitializerTests { + static final String CONFIG_NOTIFY_KEYSPACE_EVENTS = "notify-keyspace-events"; + + @Mock + RedisConnectionFactory connectionFactory; + @Mock + RedisConnection connection; + @Captor + ArgumentCaptor options; + + EnableRedisKeyspaceNotificationsInitializer initializer; + + @Before + public void setup() { + when(connectionFactory.getConnection()).thenReturn(connection); + + initializer = new EnableRedisKeyspaceNotificationsInitializer(connectionFactory); + } + + @Test + public void afterPropertiesSetUnset() throws Exception { + setConfigNotification(""); + + initializer.afterPropertiesSet(); + + assertOptionsContains("E","g","x"); + } + + @Test + public void afterPropertiesSetA() throws Exception { + setConfigNotification("A"); + + initializer.afterPropertiesSet(); + + assertOptionsContains("A", "E"); + } + + @Test + public void afterPropertiesSetE() throws Exception { + setConfigNotification("E"); + + initializer.afterPropertiesSet(); + + assertOptionsContains("E", "g", "x"); + } + + @Test + public void afterPropertiesSetK() throws Exception { + setConfigNotification("K"); + + initializer.afterPropertiesSet(); + + assertOptionsContains("K", "E", "g", "x"); + } + + @Test + public void afterPropertiesSetAE() throws Exception { + setConfigNotification("AE"); + + initializer.afterPropertiesSet(); + + verify(connection, never()).setConfig(anyString(), anyString()); + } + + @Test + public void afterPropertiesSetAK() throws Exception { + setConfigNotification("AK"); + + initializer.afterPropertiesSet(); + + assertOptionsContains("A", "K", "E"); + } + + @Test + public void afterPropertiesSetEK() throws Exception { + setConfigNotification("EK"); + + initializer.afterPropertiesSet(); + + assertOptionsContains("E", "K", "g", "x"); + } + + @Test + public void afterPropertiesSetEg() throws Exception { + setConfigNotification("Eg"); + + initializer.afterPropertiesSet(); + + assertOptionsContains("E", "g", "x"); + } + + @Test + public void afterPropertiesSetE$() throws Exception { + setConfigNotification("E$"); + + initializer.afterPropertiesSet(); + + assertOptionsContains("E", "$", "g", "x"); + } + + @Test + public void afterPropertiesSetKg() throws Exception { + setConfigNotification("Kg"); + + initializer.afterPropertiesSet(); + + assertOptionsContains("K", "g", "E", "x"); + } + + @Test + public void afterPropertiesSetAEK() throws Exception { + setConfigNotification("AEK"); + + initializer.afterPropertiesSet(); + + verify(connection, never()).setConfig(anyString(), anyString()); + } + + private void assertOptionsContains(String... expectedValues) { + verify(connection).setConfig(eq(CONFIG_NOTIFY_KEYSPACE_EVENTS), options.capture()); + for(String expectedValue : expectedValues) { + assertThat(options.getValue()).contains(expectedValue); + } + assertThat(options.getValue().length()).isEqualTo(expectedValues.length); + } + + private void setConfigNotification(String value) { + when(connection.getConfig(CONFIG_NOTIFY_KEYSPACE_EVENTS)).thenReturn(Arrays.asList(CONFIG_NOTIFY_KEYSPACE_EVENTS, value)); + } +} \ No newline at end of file diff --git a/spring-session/src/test/java/org/springframework/session/web/socket/handler/WebSocketRegistryListenerTests.java b/spring-session/src/test/java/org/springframework/session/web/socket/handler/WebSocketRegistryListenerTests.java index 9f5ec705..5fdf0796 100644 --- a/spring-session/src/test/java/org/springframework/session/web/socket/handler/WebSocketRegistryListenerTests.java +++ b/spring-session/src/test/java/org/springframework/session/web/socket/handler/WebSocketRegistryListenerTests.java @@ -15,6 +15,7 @@ */ package org.springframework.session.web.socket.handler; +import static org.fest.assertions.Assertions.assertThat; import static org.mockito.Mockito.*; import java.security.Principal; @@ -25,6 +26,7 @@ import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; +import org.mockito.internal.util.reflection.Whitebox; import org.mockito.runners.MockitoJUnitRunner; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; @@ -32,6 +34,7 @@ import org.springframework.messaging.simp.SimpMessageHeaderAccessor; import org.springframework.session.events.SessionDestroyedEvent; import org.springframework.session.web.socket.events.SessionConnectEvent; import org.springframework.session.web.socket.server.SessionRepositoryMessageInterceptor; +import org.springframework.test.util.ReflectionTestUtils; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.messaging.SessionDisconnectEvent; @@ -116,6 +119,18 @@ public class WebSocketRegistryListenerTests { verify(wsSession,times(0)).close(any(CloseStatus.class)); } + // gh-76 + @Test + public void onApplicationEventConnectDisconnectCleanup() { + listener.onApplicationEvent(connect); + + listener.onApplicationEvent(disconnect); + + Map> httpSessionIdToWsSessions = + (Map>) ReflectionTestUtils.getField(listener, "httpSessionIdToWsSessions"); + assertThat(httpSessionIdToWsSessions).isEmpty(); + } + @Test public void onApplicationEventConnectDisconnectNullSession() throws Exception { listener.onApplicationEvent(connect);