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

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

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<String> 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));
+    }
+}

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<String,Map<String,WebSocketSession>> httpSessionIdToWsSessions =
+                (Map<String, Map<String, WebSocketSession>>) ReflectionTestUtils.getField(listener, "httpSessionIdToWsSessions");
+        assertThat(httpSessionIdToWsSessions).isEmpty();
+    }
+
     @Test
     public void onApplicationEventConnectDisconnectNullSession() throws Exception {
         listener.onApplicationEvent(connect);