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);