Add ReactiveRedisIndexedSessionRepository

Closes gh-2700

spring-session-data-redis/src/integration-test/java/org/springframework/session/data/redis/AbstractRedisITests.java

@@ -16,7 +16,7 @@
 
 package org.springframework.session.data.redis;
 
-import org.testcontainers.containers.GenericContainer;
+import com.redis.testcontainers.RedisContainer;
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
@@ -34,16 +34,17 @@ public abstract class AbstractRedisITests {
 	protected static class BaseConfig {
 
 		@Bean
-		public GenericContainer redisContainer() {
-			GenericContainer redisContainer = new GenericContainer(DOCKER_IMAGE).withExposedPorts(6379);
+		public RedisContainer redisContainer() {
+			RedisContainer redisContainer = new RedisContainer(
+					RedisContainer.DEFAULT_IMAGE_NAME.withTag(RedisContainer.DEFAULT_TAG));
 			redisContainer.start();
 			return redisContainer;
 		}
 
 		@Bean
-		public LettuceConnectionFactory redisConnectionFactory() {
-			RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration(redisContainer().getHost(),
-					redisContainer().getFirstMappedPort());
+		public LettuceConnectionFactory redisConnectionFactory(RedisContainer redisContainer) {
+			RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration(redisContainer.getHost(),
+					redisContainer.getFirstMappedPort());
 			return new LettuceConnectionFactory(configuration);
 		}
 

spring-session-data-redis/src/integration-test/java/org/springframework/session/data/redis/ReactiveRedisIndexedSessionRepositoryConfigurationITests.java

@@ -0,0 +1,184 @@
+/*
+ * Copyright 2014-2023 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
+ *
+ *      https://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;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Comparator;
+import java.util.UUID;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.data.redis.core.ReactiveRedisOperations;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
+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.ReactiveFindByIndexNameSessionRepository;
+import org.springframework.session.Session;
+import org.springframework.session.config.ReactiveSessionRepositoryCustomizer;
+import org.springframework.session.data.SessionEventRegistry;
+import org.springframework.session.data.redis.ReactiveRedisIndexedSessionRepository.RedisSession;
+import org.springframework.session.data.redis.config.ConfigureReactiveRedisAction;
+import org.springframework.session.data.redis.config.annotation.web.server.EnableRedisIndexedWebSession;
+import org.springframework.session.events.SessionCreatedEvent;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+import org.springframework.web.context.support.AnnotationConfigWebApplicationContext;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+@ExtendWith(SpringExtension.class)
+class ReactiveRedisIndexedSessionRepositoryConfigurationITests {
+
+	ReactiveRedisIndexedSessionRepository repository;
+
+	ReactiveRedisOperations<String, Object> sessionRedisOperations;
+
+	AnnotationConfigWebApplicationContext context = new AnnotationConfigWebApplicationContext();
+
+	SecurityContext securityContext;
+
+	@BeforeEach
+	void setup() {
+		this.securityContext = SecurityContextHolder.createEmptyContext();
+		this.securityContext.setAuthentication(new UsernamePasswordAuthenticationToken("username-" + UUID.randomUUID(),
+				"na", AuthorityUtils.createAuthorityList("ROLE_USER")));
+	}
+
+	@Test
+	void cleanUpTaskWhenSessionIsExpiredThenAllRelatedKeysAreDeleted() {
+		registerConfig(OneSecCleanUpIntervalConfig.class);
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute("SPRING_SECURITY_CONTEXT", this.securityContext);
+		this.repository.save(session).block();
+		await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> {
+			assertThat(this.repository.findById(session.getId()).block()).isNull();
+			Boolean hasSessionKey = this.sessionRedisOperations.hasKey("spring:session:sessions:" + session.getId())
+				.block();
+			Boolean hasSessionIndexesKey = this.sessionRedisOperations
+				.hasKey("spring:session:sessions:" + session.getId() + ":idx")
+				.block();
+			Boolean hasPrincipalIndexKey = this.sessionRedisOperations
+				.hasKey("spring:session:sessions:index:"
+						+ ReactiveFindByIndexNameSessionRepository.PRINCIPAL_NAME_INDEX_NAME + ":"
+						+ this.securityContext.getAuthentication().getName())
+				.block();
+			Long expirationsSize = this.sessionRedisOperations.opsForZSet()
+				.size("spring:session:sessions:expirations")
+				.block();
+			assertThat(hasSessionKey).isFalse();
+			assertThat(hasSessionIndexesKey).isFalse();
+			assertThat(hasPrincipalIndexKey).isFalse();
+			assertThat(expirationsSize).isZero();
+		});
+	}
+
+	@Test
+	void onSessionCreatedWhenUsingJsonSerializerThenEventDeserializedCorrectly() throws InterruptedException {
+		registerConfig(SessionEventRegistryJsonSerializerConfig.class);
+		RedisSession session = this.repository.createSession().block();
+		this.repository.save(session).block();
+		SessionEventRegistry registry = this.context.getBean(SessionEventRegistry.class);
+		SessionCreatedEvent event = registry.getEvent(session.getId());
+		Session eventSession = event.getSession();
+		assertThat(eventSession).usingRecursiveComparison()
+			.withComparatorForFields(new InstantComparator(), "cached.creationTime", "cached.lastAccessedTime")
+			.isEqualTo(session);
+	}
+
+	@Test
+	void sessionExpiredWhenNoCleanUpTaskAndNoKeyspaceEventsThenNoCleanup() {
+		registerConfig(DisableCleanupTaskAndNoKeyspaceEventsConfig.class);
+		RedisSession session = this.repository.createSession().block();
+		this.repository.save(session).block();
+		await().during(Duration.ofSeconds(3)).untilAsserted(() -> {
+			Boolean exists = this.sessionRedisOperations.hasKey("spring:session:sessions:" + session.getId()).block();
+			assertThat(exists).isTrue();
+		});
+	}
+
+	private void registerConfig(Class<?> clazz) {
+		this.context.register(clazz);
+		this.context.refresh();
+		this.repository = this.context.getBean(ReactiveRedisIndexedSessionRepository.class);
+		this.sessionRedisOperations = this.repository.getSessionRedisOperations();
+	}
+
+	static class InstantComparator implements Comparator<Instant> {
+
+		@Override
+		public int compare(Instant o1, Instant o2) {
+			return o1.truncatedTo(ChronoUnit.SECONDS).compareTo(o2.truncatedTo(ChronoUnit.SECONDS));
+		}
+
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@EnableRedisIndexedWebSession(maxInactiveIntervalInSeconds = 1)
+	@Import(AbstractRedisITests.BaseConfig.class)
+	static class OneSecCleanUpIntervalConfig {
+
+		@Bean
+		ReactiveSessionRepositoryCustomizer<ReactiveRedisIndexedSessionRepository> customizer() {
+			return (sessionRepository) -> sessionRepository.setCleanupInterval(Duration.ofSeconds(1));
+		}
+
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@EnableRedisIndexedWebSession
+	@Import(AbstractRedisITests.BaseConfig.class)
+	static class SessionEventRegistryJsonSerializerConfig {
+
+		@Bean
+		SessionEventRegistry sessionEventRegistry() {
+			return new SessionEventRegistry();
+		}
+
+		@Bean
+		RedisSerializer<Object> springSessionDefaultRedisSerializer() {
+			return RedisSerializer.json();
+		}
+
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@EnableRedisIndexedWebSession(maxInactiveIntervalInSeconds = 1)
+	@Import(AbstractRedisITests.BaseConfig.class)
+	static class DisableCleanupTaskAndNoKeyspaceEventsConfig {
+
+		@Bean
+		ReactiveSessionRepositoryCustomizer<ReactiveRedisIndexedSessionRepository> customizer() {
+			return ReactiveRedisIndexedSessionRepository::disableCleanupTask;
+		}
+
+		@Bean
+		ConfigureReactiveRedisAction configureReactiveRedisAction() {
+			return (connection) -> connection.serverCommands().setConfig("notify-keyspace-events", "").then();
+		}
+
+	}
+
+}

spring-session-data-redis/src/integration-test/java/org/springframework/session/data/redis/ReactiveRedisIndexedSessionRepositoryITests.java

@@ -0,0 +1,728 @@
+/*
+ * Copyright 2014-2023 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
+ *
+ *      https://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;
+
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.Map;
+import java.util.UUID;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.data.redis.core.ReactiveRedisOperations;
+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.ReactiveFindByIndexNameSessionRepository;
+import org.springframework.session.data.SessionEventRegistry;
+import org.springframework.session.data.redis.ReactiveRedisIndexedSessionRepository.RedisSession;
+import org.springframework.session.data.redis.config.annotation.SpringSessionRedisOperations;
+import org.springframework.session.data.redis.config.annotation.web.server.EnableRedisIndexedWebSession;
+import org.springframework.session.events.SessionCreatedEvent;
+import org.springframework.session.events.SessionDeletedEvent;
+import org.springframework.session.events.SessionExpiredEvent;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+import org.springframework.test.context.web.WebAppConfiguration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.awaitility.Awaitility.await;
+
+@ExtendWith(SpringExtension.class)
+@ContextConfiguration
+@WebAppConfiguration
+@SuppressWarnings({ "ConstantConditions" })
+class ReactiveRedisIndexedSessionRepositoryITests {
+
+	private static final String SPRING_SECURITY_CONTEXT = "SPRING_SECURITY_CONTEXT";
+
+	private static final String INDEX_NAME = ReactiveRedisIndexedSessionRepository.PRINCIPAL_NAME_INDEX_NAME;
+
+	@Autowired
+	private ReactiveRedisIndexedSessionRepository repository;
+
+	@Autowired
+	private SessionEventRegistry eventRegistry;
+
+	@SpringSessionRedisOperations
+	private ReactiveRedisOperations<Object, Object> redis;
+
+	private SecurityContext context;
+
+	private SecurityContext changedContext;
+
+	@BeforeEach
+	void setup() {
+		this.context = SecurityContextHolder.createEmptyContext();
+		this.context.setAuthentication(new UsernamePasswordAuthenticationToken("username-" + UUID.randomUUID(), "na",
+				AuthorityUtils.createAuthorityList("ROLE_USER")));
+
+		this.changedContext = SecurityContextHolder.createEmptyContext();
+		this.changedContext.setAuthentication(new UsernamePasswordAuthenticationToken(
+				"changedContext-" + UUID.randomUUID(), "na", AuthorityUtils.createAuthorityList("ROLE_USER")));
+	}
+
+	@Test
+	void findByIdWhenSavedThenFound() {
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute("foo", "bar");
+		this.repository.save(session).block();
+		RedisSession savedSession = this.repository.findById(session.getId()).block();
+		assertThat(savedSession).isNotNull();
+		assertThat(savedSession.getId()).isEqualTo(session.getId());
+		assertThat(savedSession.<String>getAttribute("foo")).isEqualTo("bar");
+	}
+
+	@Test
+	void saveWhenHasSecurityContextAttributeThenPrincipalIndexKeySaved() {
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute("foo", "bar");
+
+		String username = "user";
+		Authentication authentication = UsernamePasswordAuthenticationToken.authenticated(username, "password",
+				AuthorityUtils.createAuthorityList("ROLE_USER"));
+		SecurityContext context = SecurityContextHolder.createEmptyContext();
+		context.setAuthentication(authentication);
+		session.setAttribute(SPRING_SECURITY_CONTEXT, context);
+		this.repository.save(session).block();
+
+		String usernameSessionKey = "spring:session:sessions:index:"
+				+ ReactiveRedisIndexedSessionRepository.PRINCIPAL_NAME_INDEX_NAME + ":" + username;
+		Boolean sessionExistsOnPrincipalKey = this.redis.opsForSet()
+			.isMember(usernameSessionKey, session.getId())
+			.block();
+		assertThat(sessionExistsOnPrincipalKey).isTrue();
+	}
+
+	@Test
+	void saveWhenSuccessThenSessionCreatedEvent() throws InterruptedException {
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute("foo", "bar");
+
+		this.repository.save(session).block();
+
+		SessionCreatedEvent event = this.eventRegistry.getEvent(session.getId());
+		assertThat(event).isNotNull();
+		RedisSession eventSession = event.getSession();
+		compareSessions(session, eventSession);
+	}
+
+	@Test
+	void findByPrincipalNameWhenExistsThenReturn() {
+		RedisSession session = this.repository.createSession().block();
+		String principalName = "principal";
+		session.setAttribute(ReactiveRedisIndexedSessionRepository.PRINCIPAL_NAME_INDEX_NAME, principalName);
+
+		this.repository.save(session).block();
+
+		Map<String, RedisSession> principalSessions = this.repository
+			.findByIndexNameAndIndexValue(ReactiveFindByIndexNameSessionRepository.PRINCIPAL_NAME_INDEX_NAME,
+					principalName)
+			.block();
+
+		assertThat(principalSessions).hasSize(1);
+		assertThat(principalSessions.keySet()).containsOnly(session.getId());
+
+		this.repository.deleteById(session.getId()).block();
+
+		principalSessions = this.repository
+			.findByIndexNameAndIndexValue(ReactiveFindByIndexNameSessionRepository.PRINCIPAL_NAME_INDEX_NAME,
+					principalName)
+			.block();
+		assertThat(principalSessions).isEmpty();
+	}
+
+	@Test
+	void findByPrincipalNameWhenExpireKeyEventThenRemovesIndexAndSessionExpiredEvent() {
+		String principalName = "findByPrincipalNameExpireRemovesIndex" + UUID.randomUUID();
+		RedisSession toSave = this.repository.createSession().block();
+		toSave.setAttribute(INDEX_NAME, principalName);
+
+		this.repository.save(toSave).block();
+
+		this.eventRegistry.clear();
+		String key = "spring:session:sessions:expires:" + toSave.getId();
+		assertThat(this.redis.expire(key, Duration.ofSeconds(1)).block()).isTrue();
+
+		await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> {
+			SessionExpiredEvent event = this.eventRegistry.getEvent(toSave.getId());
+			RedisSession eventSession = event.getSession();
+			Map<String, RedisSession> findByPrincipalName = this.repository
+				.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+				.block();
+			assertThat(findByPrincipalName).hasSize(0);
+			assertThat(findByPrincipalName.keySet()).doesNotContain(toSave.getId());
+			assertThat(event).isNotNull();
+			compareSessions(toSave, eventSession);
+		});
+	}
+
+	private static void compareSessions(RedisSession session1, RedisSession session2) {
+		assertThat(session2.getCreationTime().truncatedTo(ChronoUnit.SECONDS))
+			.isEqualTo(session1.getCreationTime().truncatedTo(ChronoUnit.SECONDS));
+		assertThat(session2.getMaxInactiveInterval().truncatedTo(ChronoUnit.SECONDS))
+			.isEqualTo(session1.getMaxInactiveInterval().truncatedTo(ChronoUnit.SECONDS));
+		assertThat(session2.getId()).isEqualTo(session1.getId());
+		assertThat(session2.getAttributeNames()).isEqualTo(session1.getAttributeNames());
+	}
+
+	@Test
+	void findByPrincipalNameWhenDeletedKeyEventThenRemovesIndex() {
+		String principalName = "findByPrincipalNameExpireRemovesIndex" + UUID.randomUUID();
+		RedisSession toSave = this.repository.createSession().block();
+		toSave.setAttribute(INDEX_NAME, principalName);
+
+		this.repository.save(toSave).block();
+
+		String key = "spring:session:sessions:expires:" + toSave.getId();
+		assertThat(this.redis.delete(key).block()).isEqualTo(1);
+
+		await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> {
+			Map<String, RedisSession> findByPrincipalName = this.repository
+				.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+				.block();
+			assertThat(findByPrincipalName).hasSize(0);
+			assertThat(findByPrincipalName.keySet()).doesNotContain(toSave.getId());
+			SessionDeletedEvent event = this.eventRegistry.getEvent(toSave.getId());
+			assertThat(event).isNotNull();
+			RedisSession eventSession = event.getSession();
+			compareSessions(toSave, eventSession);
+		});
+	}
+
+	@Test
+	void findByPrincipalNameWhenNoPrincipalNameChangeThenKeepIndex() {
+		String principalName = "findByPrincipalNameNoPrincipalNameChange" + UUID.randomUUID();
+		RedisSession toSave = this.repository.createSession().block();
+		toSave.setAttribute(INDEX_NAME, principalName);
+
+		this.repository.save(toSave).block();
+
+		toSave.setAttribute("other", "value");
+		this.repository.save(toSave).block();
+
+		Map<String, RedisSession> findByPrincipalName = this.repository
+			.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+			.block();
+
+		assertThat(findByPrincipalName).hasSize(1);
+		assertThat(findByPrincipalName.keySet()).containsOnly(toSave.getId());
+	}
+
+	@Test
+	void findByPrincipalNameWhenNoPrincipalNameChangeAndFindByIdThenKeepIndex() {
+		String principalName = "findByPrincipalNameNoPrincipalNameChange" + UUID.randomUUID();
+		RedisSession toSave = this.repository.createSession().block();
+		toSave.setAttribute(INDEX_NAME, principalName);
+
+		this.repository.save(toSave).block();
+		toSave = this.repository.findById(toSave.getId()).block();
+
+		toSave.setAttribute("other", "value");
+		this.repository.save(toSave).block();
+
+		Map<String, RedisSession> findByPrincipalName = this.repository
+			.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+			.block();
+
+		assertThat(findByPrincipalName).hasSize(1);
+		assertThat(findByPrincipalName.keySet()).containsOnly(toSave.getId());
+	}
+
+	@Test
+	void findByPrincipalNameWhenDeletedPrincipalAttributeThenEmpty() {
+		String principalName = "findByDeletedPrincipalName" + UUID.randomUUID();
+		RedisSession toSave = this.repository.createSession().block();
+		toSave.setAttribute(INDEX_NAME, principalName);
+
+		this.repository.save(toSave).block();
+
+		toSave.removeAttribute(INDEX_NAME);
+		this.repository.save(toSave).block();
+
+		Map<String, RedisSession> findByPrincipalName = this.repository
+			.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+			.block();
+
+		assertThat(findByPrincipalName).isEmpty();
+	}
+
+	@Test
+	void findByPrincipalNameWhenDeletedPrincipalAttributeAndFindByIdThenEmpty() {
+		String principalName = "findByDeletedPrincipalName" + UUID.randomUUID();
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute(INDEX_NAME, principalName);
+
+		this.repository.save(session).block();
+		session = this.repository.findById(session.getId()).block();
+
+		session.removeAttribute(INDEX_NAME);
+		this.repository.save(session).block();
+
+		Map<String, RedisSession> findByPrincipalName = this.repository
+			.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+			.block();
+
+		assertThat(findByPrincipalName).isEmpty();
+	}
+
+	@Test
+	void findByPrincipalNameWhenChangedSecurityContextAttributeThenIndexMovedToNewPrincipal() {
+		String principalName = this.context.getAuthentication().getName();
+		String principalNameChanged = this.changedContext.getAuthentication().getName();
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute(SPRING_SECURITY_CONTEXT, this.context);
+
+		this.repository.save(session).block();
+
+		session.setAttribute(SPRING_SECURITY_CONTEXT, this.changedContext);
+		this.repository.save(session).block();
+
+		Map<String, RedisSession> findByPrincipalName = this.repository
+			.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+			.block();
+		assertThat(findByPrincipalName).isEmpty();
+
+		findByPrincipalName = this.repository.findByIndexNameAndIndexValue(INDEX_NAME, principalNameChanged).block();
+		assertThat(findByPrincipalName).hasSize(1);
+		assertThat(findByPrincipalName.keySet()).containsOnly(session.getId());
+	}
+
+	@Test
+	void findByPrincipalNameWhenChangedSecurityContextAttributeAndFindByIdThenIndexMovedToNewPrincipal() {
+		String principalName = this.context.getAuthentication().getName();
+		String principalNameChanged = this.changedContext.getAuthentication().getName();
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute(SPRING_SECURITY_CONTEXT, this.context);
+
+		this.repository.save(session).block();
+		session = this.repository.findById(session.getId()).block();
+
+		session.setAttribute(SPRING_SECURITY_CONTEXT, this.changedContext);
+		this.repository.save(session).block();
+
+		Map<String, RedisSession> findByPrincipalName = this.repository
+			.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+			.block();
+		assertThat(findByPrincipalName).isEmpty();
+
+		findByPrincipalName = this.repository.findByIndexNameAndIndexValue(INDEX_NAME, principalNameChanged).block();
+		assertThat(findByPrincipalName).hasSize(1);
+		assertThat(findByPrincipalName.keySet()).containsOnly(session.getId());
+	}
+
+	@Test
+	void findByPrincipalNameWhenNoSecurityContextChangeThenKeepIndex() {
+		String principalName = this.context.getAuthentication().getName();
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute(SPRING_SECURITY_CONTEXT, this.context);
+
+		this.repository.save(session).block();
+
+		session.setAttribute("other", "value");
+		this.repository.save(session).block();
+
+		Map<String, RedisSession> findByPrincipalName = this.repository
+			.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+			.block();
+
+		assertThat(findByPrincipalName).hasSize(1);
+		assertThat(findByPrincipalName.keySet()).containsOnly(session.getId());
+	}
+
+	@Test
+	void findByPrincipalNameWhenNoSecurityContextChangeAndFindByIdThenKeepIndex() {
+		String principalName = this.context.getAuthentication().getName();
+		RedisSession toSave = this.repository.createSession().block();
+		toSave.setAttribute(SPRING_SECURITY_CONTEXT, this.context);
+
+		this.repository.save(toSave).block();
+		toSave = this.repository.findById(toSave.getId()).block();
+
+		toSave.setAttribute("other", "value");
+		this.repository.save(toSave).block();
+
+		Map<String, RedisSession> findByPrincipalName = this.repository
+			.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+			.block();
+
+		assertThat(findByPrincipalName).hasSize(1);
+		assertThat(findByPrincipalName.keySet()).containsOnly(toSave.getId());
+	}
+
+	@Test
+	void findByPrincipalNameWhenDeletedSecurityContextAttributeThenEmpty() {
+		String principalName = this.context.getAuthentication().getName();
+		RedisSession toSave = this.repository.createSession().block();
+		toSave.setAttribute(SPRING_SECURITY_CONTEXT, this.context);
+
+		this.repository.save(toSave).block();
+
+		toSave.removeAttribute(SPRING_SECURITY_CONTEXT);
+		this.repository.save(toSave).block();
+
+		Map<String, RedisSession> findByPrincipalName = this.repository
+			.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+			.block();
+
+		assertThat(findByPrincipalName).isEmpty();
+	}
+
+	@Test
+	void findByPrincipalNameWhenDeletedSecurityContextAttributeAndFindByIdThenEmpty() {
+		String principalName = this.context.getAuthentication().getName();
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute(SPRING_SECURITY_CONTEXT, this.context);
+
+		this.repository.save(session).block();
+		session = this.repository.findById(session.getId()).block();
+
+		session.removeAttribute(SPRING_SECURITY_CONTEXT);
+		this.repository.save(session).block();
+
+		Map<String, RedisSession> findByPrincipalName = this.repository
+			.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+
+			.block();
+
+		assertThat(findByPrincipalName).isEmpty();
+	}
+
+	@Test
+	void findByPrincipalNameWhenChangedPrincipalAttributeThenEmpty() {
+		String principalName = this.context.getAuthentication().getName();
+		String principalNameChanged = this.changedContext.getAuthentication().getName();
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute(SPRING_SECURITY_CONTEXT, this.context);
+
+		this.repository.save(session).block();
+
+		session.setAttribute(SPRING_SECURITY_CONTEXT, this.changedContext);
+		this.repository.save(session).block();
+
+		Map<String, RedisSession> findByPrincipalName = this.repository
+			.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+			.block();
+		assertThat(findByPrincipalName).isEmpty();
+
+		findByPrincipalName = this.repository.findByIndexNameAndIndexValue(INDEX_NAME, principalNameChanged).block();
+		assertThat(findByPrincipalName).hasSize(1);
+		assertThat(findByPrincipalName.keySet()).containsOnly(session.getId());
+	}
+
+	@Test
+	void findByPrincipalNameWhenChangedPrincipalAttributeAndFindByIdThenEmpty() {
+		String principalName = this.context.getAuthentication().getName();
+		String principalNameChanged = this.changedContext.getAuthentication().getName();
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute(SPRING_SECURITY_CONTEXT, this.context);
+
+		this.repository.save(session).block();
+		session = this.repository.findById(session.getId()).block();
+
+		session.setAttribute(SPRING_SECURITY_CONTEXT, this.changedContext);
+		this.repository.save(session).block();
+
+		Map<String, RedisSession> findByPrincipalName = this.repository
+			.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+			.block();
+		assertThat(findByPrincipalName).isEmpty();
+
+		findByPrincipalName = this.repository.findByIndexNameAndIndexValue(INDEX_NAME, principalNameChanged).block();
+		assertThat(findByPrincipalName).hasSize(1);
+		assertThat(findByPrincipalName.keySet()).containsOnly(session.getId());
+	}
+
+	// gh-1791
+	@Test
+	void changeSessionIdWhenSessionExpiredThenRemovesAllPrincipalIndex() {
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute(SPRING_SECURITY_CONTEXT, this.context);
+
+		this.repository.save(session).block();
+		String usernameSessionKey = "spring:session:sessions:index:" + INDEX_NAME + ":" + getSecurityName();
+
+		RedisSession findById = this.repository.findById(session.getId()).block();
+		String originalFindById = findById.getId();
+
+		assertThat(this.redis.opsForSet().members(usernameSessionKey).collectList().block()).contains(originalFindById);
+
+		String changeSessionId = findById.changeSessionId();
+		findById.setAttribute(SPRING_SECURITY_CONTEXT, this.context);
+
+		this.repository.save(findById).block();
+
+		assertThat(this.redis.opsForSet().members(usernameSessionKey).collectList().block()).contains(changeSessionId);
+
+		String key = "spring:session:sessions:expires:" + changeSessionId;
+		assertThat(this.redis.expire(key, Duration.ofSeconds(1)).block()).isTrue(); // expire
+																					// the
+																					// key
+
+		await().atMost(Duration.ofSeconds(5))
+			.untilAsserted(() -> assertThat(this.redis.opsForSet().members(usernameSessionKey).collectList().block())
+				.isEmpty());
+	}
+
+	@Test
+	void changeSessionIdWhenSessionDeletedThenRemovesAllPrincipalIndex() {
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute(SPRING_SECURITY_CONTEXT, this.context);
+
+		this.repository.save(session).block();
+		String usernameSessionKey = "spring:session:sessions:index:" + INDEX_NAME + ":" + getSecurityName();
+
+		RedisSession findById = this.repository.findById(session.getId()).block();
+		String originalFindById = findById.getId();
+
+		assertThat(this.redis.opsForSet().members(usernameSessionKey).collectList().block()).contains(originalFindById);
+
+		String changeSessionId = findById.changeSessionId();
+		findById.setAttribute(SPRING_SECURITY_CONTEXT, this.context);
+
+		this.repository.save(findById).block();
+
+		assertThat(this.redis.opsForSet().members(usernameSessionKey).collectList().block()).contains(changeSessionId);
+
+		String key = "spring:session:sessions:expires:" + changeSessionId;
+		assertThat(this.redis.delete(key).block()).isEqualTo(1);
+
+		await().atMost(Duration.ofSeconds(5))
+			.untilAsserted(() -> assertThat(this.redis.opsForSet().members(usernameSessionKey).collectList().block())
+				.isEmpty());
+	}
+
+	@Test
+	void changeSessionIdWhenPrincipalNameChangesThenNewPrincipalMapsToNewSessionId() {
+		String principalName = "findByChangedPrincipalName" + UUID.randomUUID();
+		String principalNameChanged = "findByChangedPrincipalName" + UUID.randomUUID();
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute(INDEX_NAME, principalName);
+
+		this.repository.save(session).block();
+
+		RedisSession findById = this.repository.findById(session.getId()).block();
+		String changeSessionId = findById.changeSessionId();
+		findById.setAttribute(INDEX_NAME, principalNameChanged);
+		this.repository.save(findById).block();
+
+		Map<String, RedisSession> findByPrincipalName = this.repository
+			.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+
+			.block();
+		assertThat(findByPrincipalName).isEmpty();
+
+		findByPrincipalName = this.repository.findByIndexNameAndIndexValue(INDEX_NAME, principalNameChanged).block();
+
+		assertThat(findByPrincipalName).hasSize(1);
+		assertThat(findByPrincipalName.keySet()).containsOnly(changeSessionId);
+	}
+
+	// gh-1987
+	@Test
+	void changeSessionIdWhenPrincipalNameChangesFromNullThenIndexShouldNotBeCreated() {
+		String principalName = null;
+		String principalNameChanged = "findByChangedPrincipalName" + UUID.randomUUID();
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute(INDEX_NAME, principalName);
+
+		this.repository.save(session).block();
+
+		RedisSession findById = this.repository.findById(session.getId()).block();
+		String changeSessionId = findById.changeSessionId();
+		findById.setAttribute(INDEX_NAME, principalNameChanged);
+		this.repository.save(findById).block();
+
+		Map<String, RedisSession> findByPrincipalName = this.repository
+			.findByIndexNameAndIndexValue(INDEX_NAME, principalName)
+			.block();
+		assertThat(findByPrincipalName).isEmpty();
+
+		findByPrincipalName = this.repository.findByIndexNameAndIndexValue(INDEX_NAME, principalNameChanged).block();
+
+		assertThat(findByPrincipalName).hasSize(1);
+		assertThat(findByPrincipalName.keySet()).containsOnly(changeSessionId);
+	}
+
+	@Test
+	void changeSessionIdWhenOnlyChangeId() {
+		String attrName = "changeSessionId";
+		String attrValue = "changeSessionId-value";
+		RedisSession session = this.repository.createSession().block();
+		session.setAttribute(attrName, attrValue);
+
+		this.repository.save(session).block();
+
+		RedisSession findById = this.repository.findById(session.getId()).block();
+
+		assertThat(findById.<String>getAttribute(attrName)).isEqualTo(attrValue);
+
+		String originalFindById = findById.getId();
+		String changeSessionId = findById.changeSessionId();
+
+		this.repository.save(findById).block();
+
+		assertThat(this.repository.findById(originalFindById).block()).isNull();
+
+		RedisSession findByChangeSessionId = this.repository.findById(changeSessionId).block();
+
+		assertThat(findByChangeSessionId.<String>getAttribute(attrName)).isEqualTo(attrValue);
+	}
+
+	@Test
+	void changeSessionIdWhenChangeTwice() {
+		RedisSession session = this.repository.createSession().block();
+
+		this.repository.save(session).block();
+
+		String originalId = session.getId();
+		String changeId1 = session.changeSessionId();
+		String changeId2 = session.changeSessionId();
+
+		this.repository.save(session).block();
+
+		assertThat(this.repository.findById(originalId).block()).isNull();
+		assertThat(this.repository.findById(changeId1).block()).isNull();
+		assertThat(this.repository.findById(changeId2).block()).isNotNull();
+	}
+
+	@Test
+	void changeSessionIdWhenSetAttributeOnChangedSession() {
+		String attrName = "changeSessionId";
+		String attrValue = "changeSessionId-value";
+
+		RedisSession session = this.repository.createSession().block();
+
+		this.repository.save(session).block();
+
+		RedisSession findById = this.repository.findById(session.getId()).block();
+
+		findById.setAttribute(attrName, attrValue);
+
+		String originalFindById = findById.getId();
+		String changeSessionId = findById.changeSessionId();
+
+		this.repository.save(findById).block();
+
+		assertThat(this.repository.findById(originalFindById).block()).isNull();
+
+		RedisSession findByChangeSessionId = this.repository.findById(changeSessionId).block();
+
+		assertThat(findByChangeSessionId.<String>getAttribute(attrName)).isEqualTo(attrValue);
+	}
+
+	@Test
+	void changeSessionIdWhenHasNotSaved() {
+		RedisSession session = this.repository.createSession().block();
+		String originalId = session.getId();
+		session.changeSessionId();
+
+		this.repository.save(session).block();
+
+		assertThat(this.repository.findById(session.getId()).block()).isNotNull();
+		assertThat(this.repository.findById(originalId).block()).isNull();
+	}
+
+	// gh-962
+	@Test
+	void changeSessionIdSaveTwice() {
+		RedisSession toSave = this.repository.createSession().block();
+		String originalId = toSave.getId();
+		toSave.changeSessionId();
+
+		this.repository.save(toSave).block();
+		this.repository.save(toSave).block();
+
+		assertThat(this.repository.findById(toSave.getId()).block()).isNotNull();
+		assertThat(this.repository.findById(originalId).block()).isNull();
+	}
+
+	// gh-1137
+	@Test
+	void changeSessionIdWhenSessionIsDeleted() {
+		RedisSession toSave = this.repository.createSession().block();
+		String sessionId = toSave.getId();
+		this.repository.save(toSave).block();
+
+		this.repository.deleteById(sessionId).block();
+
+		toSave.changeSessionId();
+		this.repository.save(toSave).block();
+
+		assertThat(this.repository.findById(toSave.getId()).block()).isNull();
+		assertThat(this.repository.findById(sessionId).block()).isNull();
+	}
+
+	@Test // gh-1270
+	void changeSessionIdSaveConcurrently() {
+		RedisSession toSave = this.repository.createSession().block();
+		String originalId = toSave.getId();
+		this.repository.save(toSave).block();
+
+		RedisSession copy1 = this.repository.findById(originalId).block();
+		RedisSession copy2 = this.repository.findById(originalId).block();
+
+		copy1.changeSessionId();
+		this.repository.save(copy1).block();
+		copy2.changeSessionId();
+		this.repository.save(copy2).block();
+
+		assertThat(this.repository.findById(originalId).block()).isNull();
+		assertThat(this.repository.findById(copy1.getId()).block()).isNotNull();
+		assertThat(this.repository.findById(copy2.getId()).block()).isNull();
+	}
+
+	// gh-1743
+	@Test
+	void saveChangeSessionIdWhenFailedRenameOperationExceptionContainsMoreDetailsThenIgnoreError() {
+		RedisSession toSave = this.repository.createSession().block();
+		String sessionId = toSave.getId();
+
+		this.repository.save(toSave).block();
+		RedisSession session = this.repository.findById(sessionId).block();
+		this.repository.deleteById(sessionId).block();
+		session.changeSessionId();
+
+		assertThatNoException().isThrownBy(() -> this.repository.save(session).block());
+	}
+
+	private String getSecurityName() {
+		return this.context.getAuthentication().getName();
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@EnableRedisIndexedWebSession
+	@Import(AbstractRedisITests.BaseConfig.class)
+	static class Config {
+
+		@Bean
+		SessionEventRegistry sessionEventRegistry() {
+			return new SessionEventRegistry();
+		}
+
+	}
+
+}

spring-session-data-redis/src/main/java/org/springframework/session/data/redis/ReactiveRedisIndexedSessionRepository.java

@@ -0,0 +1,788 @@
+/*
+ * Copyright 2014-2023 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
+ *
+ *      https://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;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiFunction;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.core.NestedExceptionUtils;
+import org.springframework.data.redis.connection.ReactiveSubscription;
+import org.springframework.data.redis.core.ReactiveRedisOperations;
+import org.springframework.data.redis.core.ReactiveRedisTemplate;
+import org.springframework.session.IndexResolver;
+import org.springframework.session.MapSession;
+import org.springframework.session.ReactiveFindByIndexNameSessionRepository;
+import org.springframework.session.ReactiveSessionRepository;
+import org.springframework.session.SaveMode;
+import org.springframework.session.Session;
+import org.springframework.session.SessionIdGenerator;
+import org.springframework.session.UuidSessionIdGenerator;
+import org.springframework.session.events.SessionCreatedEvent;
+import org.springframework.session.events.SessionDeletedEvent;
+import org.springframework.session.events.SessionDestroyedEvent;
+import org.springframework.session.events.SessionExpiredEvent;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * A {@link ReactiveSessionRepository} that is implemented using Spring Data's
+ * {@link ReactiveRedisOperations}.
+ *
+ * <h2>Storage Details</h2> The sections below outline how Redis is updated for each
+ * operation. An example of creating a new session can be found below. The subsequent
+ * sections describe the details.
+ *
+ * <pre>
+ * HMSET spring:session:sessions:648377f7-c76f-4f45-b847-c0268bb48381 creationTime 1702400400000 maxInactiveInterval 1800 lastAccessedTime 1702400400000 sessionAttr:attrName someAttrValue sessionAttr:attrName2 someAttrValue2
+ * EXPIRE spring:session:sessions:648377f7-c76f-4f45-b847-c0268bb48381 2100
+ * APPEND spring:session:sessions:expires:648377f7-c76f-4f45-b847-c0268bb48381 ""
+ * EXPIRE spring:session:sessions:expires:648377f7-c76f-4f45-b847-c0268bb48381 1800
+ * ZADD spring:session:sessions:expirations "1.702402961162E12" "648377f7-c76f-4f45-b847-c0268bb48381"
+ * SADD spring:session:sessions:index:PRINCIPAL_NAME_INDEX_NAME:user "648377f7-c76f-4f45-b847-c0268bb48381"
+ * SADD spring:session:sessions:648377f7-c76f-4f45-b847-c0268bb48381:idx "spring:session:sessions:index:PRINCIPAL_NAME_INDEX_NAME:user"
+ * </pre>
+ *
+ * <h3>Saving a Session</h3>
+ *
+ * <p>
+ * Each session is stored in Redis as a
+ * <a href="https://redis.io/topics/data-types#hashes">Hash</a>. Each session is set and
+ * updated using the <a href="https://redis.io/commands/hmset">HMSET command</a>. An
+ * example of how each session is stored can be seen below.
+ * </p>
+ *
+ * <pre>
+ * HMSET spring:session:sessions:648377f7-c76f-4f45-b847-c0268bb48381 creationTime 1702400400000 maxInactiveInterval 1800 lastAccessedTime 1702400400000 sessionAttr:attrName someAttrValue sessionAttr:attrName2 someAttrValue2
+ * </pre>
+ *
+ * <p>
+ * In this example, the session following statements are true about the session:
+ * </p>
+ * <ul>
+ * <li>The session id is 648377f7-c76f-4f45-b847-c0268bb48381</li>
+ * <li>The session was created at 1702400400000 in milliseconds since midnight of 1/1/1970
+ * GMT.</li>
+ * <li>The session expires in 1800 seconds (30 minutes).</li>
+ * <li>The session was last accessed at 1702400400000 in milliseconds since midnight of
+ * 1/1/1970 GMT.</li>
+ * <li>The session has two attributes. The first is "attrName" with the value of
+ * "someAttrValue". The second session attribute is named "attrName2" with the value of
+ * "someAttrValue2".</li>
+ * </ul>
+ *
+ * <h3>Optimized Writes</h3>
+ *
+ * <p>
+ * The {@link ReactiveRedisIndexedSessionRepository.RedisSession} keeps track of the
+ * properties that have changed and only updates those. This means if an attribute is
+ * written once and read many times we only need to write that attribute once. For
+ * example, assume the session attribute "attrName2" from earlier was updated. The
+ * following would be executed upon saving:
+ * </p>
+ *
+ * <pre>
+ * HMSET spring:session:sessions:648377f7-c76f-4f45-b847-c0268bb48381 sessionAttr:attrName2 newValue
+ * </pre>
+ *
+ * <h3>SessionCreatedEvent</h3>
+ *
+ * <p>
+ * When a session is created an event is sent to Redis with the channel of
+ * "spring:session:event:0:created:648377f7-c76f-4f45-b847-c0268bb48381" such that
+ * "648377f7-c76f-4f45-b847-c0268bb48381" is the session id. The body of the event will be
+ * the session that was created.
+ * </p>
+ *
+ * <h3>SessionDeletedEvent and SessionExpiredEvent</h3> If you configured you Redis server
+ * to send keyspace events when keys are expired or deleted, either via
+ * {@link org.springframework.session.data.redis.config.annotation.ConfigureNotifyKeyspaceEventsReactiveAction}
+ * or via external configuration, then deleted and expired sessions will be published as
+ * {@link SessionDeletedEvent} and {@link SessionExpiredEvent} respectively.
+ *
+ * <h3>Expiration</h3>
+ *
+ * <p>
+ * An expiration is associated to each session using the
+ * <a href="https://redis.io/commands/expire">EXPIRE command</a> based upon the
+ * {@link ReactiveRedisIndexedSessionRepository.RedisSession#getMaxInactiveInterval()} .
+ * For example:
+ * </p>
+ *
+ * <pre>
+ * EXPIRE spring:session:sessions:648377f7-c76f-4f45-b847-c0268bb48381 2100
+ * </pre>
+ *
+ * <p>
+ * You will note that the expiration that is set is 5 minutes after the session actually
+ * expires. This is necessary so that the value of the session can be accessed when the
+ * session expires. An expiration is set on the session itself five minutes after it
+ * actually expires to ensure it is cleaned up, but only after we perform any necessary
+ * processing.
+ * </p>
+ *
+ * <p>
+ * <b>NOTE:</b> The {@link #findById(String)} method ensures that no expired sessions will
+ * be returned. This means there is no need to check the expiration before using a
+ * session.
+ * </p>
+ *
+ * <p>
+ * Spring Session relies on the expired and delete
+ * <a href="https://redis.io/docs/manual/keyspace-notifications/">keyspace
+ * notifications</a> from Redis to fire a SessionDestroyedEvent. It is the
+ * SessionDestroyedEvent that ensures resources associated with the Session are cleaned
+ * up. For example, when using Spring Session's WebSocket support the Redis expired or
+ * delete event is what triggers any WebSocket connections associated with the session to
+ * be closed.
+ * </p>
+ *
+ * <p>
+ * Expiration is not tracked directly on the session key itself since this would mean the
+ * session data would no longer be available. Instead a special session expires key is
+ * used. In our example the expires key is:
+ * </p>
+ *
+ * <pre>
+ * APPEND spring:session:sessions:expires:648377f7-c76f-4f45-b847-c0268bb48381 ""
+ * EXPIRE spring:session:sessions:expires:648377f7-c76f-4f45-b847-c0268bb48381 1800
+ * </pre>
+ *
+ * <p>
+ * When a session key is deleted or expires, the keyspace notification triggers a lookup
+ * of the actual session and a {@link SessionDestroyedEvent} is fired.
+ * </p>
+ *
+ * <p>
+ * One problem with relying on Redis expiration exclusively is that Redis makes no
+ * guarantee of when the expired event will be fired if the key has not been accessed. For
+ * additional details see <a href="https://redis.io/commands/expire/">How Redis expires
+ * keys</a> section in the Redis Expire documentation.
+ * </p>
+ *
+ * <p>
+ * To circumvent the fact that expired events are not guaranteed to happen we can ensure
+ * that each key is accessed when it is expected to expire. This means that if the TTL is
+ * expired on the key, Redis will remove the key and fire the expired event when we try to
+ * access the key.
+ * </p>
+ *
+ * <p>
+ * For this reason, each session expiration is also tracked by storing the session id in a
+ * sorted set ranked by its expiration time. This allows a background task to access the
+ * potentially expired sessions to ensure that Redis expired events are fired in a more
+ * deterministic fashion. For example:
+ * </p>
+ *
+ * <pre>
+ * ZADD spring:session:sessions:expirations "1.702402961162E12" "648377f7-c76f-4f45-b847-c0268bb48381"
+ * </pre>
+ *
+ * <p>
+ * <b>NOTE</b>: We do not explicitly delete the keys since in some instances there may be
+ * a race condition that incorrectly identifies a key as expired when it is not. Short of
+ * using distributed locks (which would kill our performance) there is no way to ensure
+ * the consistency of the expiration mapping. By simply accessing the key, we ensure that
+ * the key is only removed if the TTL on that key is expired.
+ * </p>
+ *
+ * <h3>Secondary Indexes</h3> By default, Spring Session will also index the sessions by
+ * identifying if the session contains any attribute that can be mapped to a principal
+ * using an {@link org.springframework.session.PrincipalNameIndexResolver}. All resolved
+ * indexes for a session are stored in a Redis Set, for example: <pre>
+ * SADD spring:session:sessions:index:PRINCIPAL_NAME_INDEX_NAME:user "648377f7-c76f-4f45-b847-c0268bb48381"
+ * SADD spring:session:sessions:648377f7-c76f-4f45-b847-c0268bb48381:idx "spring:session:sessions:index:PRINCIPAL_NAME_INDEX_NAME:user"
+ * </pre>
+ *
+ * Therefore, you can check all indexes for a given session by getting the members of the
+ * {@code "spring:session:sessions:648377f7-c76f-4f45-b847-c0268bb48381:idx"} Redis set.
+ *
+ * @author Marcus da Coregio
+ * @since 3.3
+ */
+public class ReactiveRedisIndexedSessionRepository
+		implements ReactiveSessionRepository<ReactiveRedisIndexedSessionRepository.RedisSession>,
+		ReactiveFindByIndexNameSessionRepository<ReactiveRedisIndexedSessionRepository.RedisSession>, DisposableBean,
+		InitializingBean {
+
+	private static final Log logger = LogFactory.getLog(ReactiveRedisIndexedSessionRepository.class);
+
+	/**
+	 * The default namespace for each key and channel in Redis used by Spring Session.
+	 */
+	public static final String DEFAULT_NAMESPACE = "spring:session";
+
+	/**
+	 * The default Redis database used by Spring Session.
+	 */
+	public static final int DEFAULT_DATABASE = 0;
+
+	private final ReactiveRedisOperations<String, Object> sessionRedisOperations;
+
+	private final ReactiveRedisTemplate<String, String> keyEventsOperations;
+
+	private SessionIdGenerator sessionIdGenerator = UuidSessionIdGenerator.getInstance();
+
+	private BiFunction<String, Map<String, Object>, Mono<MapSession>> redisSessionMapper = new RedisSessionMapperAdapter();
+
+	private Duration defaultMaxInactiveInterval = Duration.ofSeconds(MapSession.DEFAULT_MAX_INACTIVE_INTERVAL_SECONDS);
+
+	private SaveMode saveMode = SaveMode.ON_SET_ATTRIBUTE;
+
+	private ApplicationEventPublisher eventPublisher = (event) -> {
+	};
+
+	private String sessionCreatedChannelPrefix;
+
+	private String sessionDeletedChannel;
+
+	private String sessionExpiredChannel;
+
+	private String expiredKeyPrefix;
+
+	private final List<Disposable> subscriptions = new ArrayList<>();
+
+	/**
+	 * The namespace for every key used by Spring Session in Redis.
+	 */
+	private String namespace = DEFAULT_NAMESPACE + ":";
+
+	private int database = DEFAULT_DATABASE;
+
+	private ReactiveRedisSessionIndexer indexer;
+
+	private SortedSetReactiveRedisSessionExpirationStore expirationStore;
+
+	private Duration cleanupInterval = Duration.ofSeconds(60);
+
+	private Clock clock = Clock.systemUTC();
+
+	/**
+	 * Creates a new instance with the provided {@link ReactiveRedisOperations}.
+	 * @param sessionRedisOperations the {@link ReactiveRedisOperations} to use for
+	 * managing the sessions. Cannot be null.
+	 * @param keyEventsOperations the {@link ReactiveRedisTemplate} to use to subscribe to
+	 * keyspace events. Cannot be null.
+	 */
+	public ReactiveRedisIndexedSessionRepository(ReactiveRedisOperations<String, Object> sessionRedisOperations,
+			ReactiveRedisTemplate<String, String> keyEventsOperations) {
+		Assert.notNull(sessionRedisOperations, "sessionRedisOperations cannot be null");
+		Assert.notNull(keyEventsOperations, "keyEventsOperations cannot be null");
+		this.sessionRedisOperations = sessionRedisOperations;
+		this.keyEventsOperations = keyEventsOperations;
+		this.indexer = new ReactiveRedisSessionIndexer(sessionRedisOperations, this.namespace);
+		this.expirationStore = new SortedSetReactiveRedisSessionExpirationStore(sessionRedisOperations, this.namespace);
+		configureSessionChannels();
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		subscribeToRedisEvents();
+		setupCleanupTask();
+	}
+
+	private void setupCleanupTask() {
+		if (!this.cleanupInterval.isZero()) {
+			Disposable cleanupExpiredSessionsTask = Flux.interval(this.cleanupInterval, this.cleanupInterval)
+				.onBackpressureDrop((count) -> logger
+					.debug("Skipping clean-up expired sessions because the previous one is still running."))
+				.concatMap((count) -> cleanUpExpiredSessions())
+				.subscribe();
+			this.subscriptions.add(cleanupExpiredSessionsTask);
+		}
+	}
+
+	private Flux<Void> cleanUpExpiredSessions() {
+		return this.expirationStore.retrieveExpiredSessions(this.clock.instant()).flatMap(this::touch);
+	}
+
+	private Mono<Void> touch(String sessionId) {
+		return this.sessionRedisOperations.hasKey(getExpiredKey(sessionId)).then();
+	}
+
+	@Override
+	public void destroy() {
+		for (Disposable subscription : this.subscriptions) {
+			subscription.dispose();
+		}
+		this.subscriptions.clear();
+	}
+
+	@Override
+	public Mono<Map<String, RedisSession>> findByIndexNameAndIndexValue(String indexName, String indexValue) {
+		return this.indexer.getSessionIds(indexName, indexValue)
+			.flatMap(this::findById)
+			.collectMap(RedisSession::getId);
+	}
+
+	@Override
+	public Mono<RedisSession> createSession() {
+		return Mono.fromSupplier(() -> this.sessionIdGenerator.generate())
+			.subscribeOn(Schedulers.boundedElastic())
+			.publishOn(Schedulers.parallel())
+			.map(MapSession::new)
+			.doOnNext((session) -> session.setMaxInactiveInterval(this.defaultMaxInactiveInterval))
+			.map((session) -> new RedisSession(session, true));
+	}
+
+	@Override
+	public Mono<Void> save(RedisSession session) {
+		// @formatter:off
+		return session.save()
+				.then(Mono.defer(() -> this.indexer.update(session)))
+				.then(Mono.defer(() -> this.expirationStore.add(session.getId(), session.getLastAccessedTime().plus(session.getMaxInactiveInterval()))));
+		// @formatter:on
+	}
+
+	@Override
+	public Mono<RedisSession> findById(String id) {
+		return getSession(id, false);
+	}
+
+	private Mono<RedisSession> getSession(String sessionId, boolean allowExpired) {
+		// @formatter:off
+		String sessionKey = getSessionKey(sessionId);
+		return this.sessionRedisOperations.opsForHash().entries(sessionKey)
+				.collectMap((entry) -> entry.getKey().toString(), Map.Entry::getValue)
+				.filter((map) -> !map.isEmpty())
+				.flatMap((map) -> this.redisSessionMapper.apply(sessionId, map))
+				.filter((session) -> allowExpired || !session.isExpired())
+				.map((session) -> new RedisSession(session, false));
+		// @formatter:on
+	}
+
+	@Override
+	public Mono<Void> deleteById(String id) {
+		// @formatter:off
+		return getSession(id, true)
+				.flatMap((session) -> this.sessionRedisOperations.delete(getExpiredKey(session.getId()))
+						.thenReturn(session))
+				.flatMap((session) -> this.sessionRedisOperations.delete(getSessionKey(session.getId())).thenReturn(session))
+				.flatMap((session) -> this.indexer.delete(session.getId()).thenReturn(session))
+				.flatMap((session) -> this.expirationStore.remove(session.getId()));
+		// @formatter:on
+	}
+
+	/**
+	 * Subscribes to {@code __keyevent@0__:expired} and {@code __keyevent@0__:del} Redis
+	 * Keyspaces events and to {@code spring:session:event:0:created:*} Redis Channel
+	 * event in order to clean up the sessions and publish the related Spring Session
+	 * events.
+	 */
+	private void subscribeToRedisEvents() {
+		Disposable sessionCreatedSubscription = this.sessionRedisOperations
+			.listenToPattern(getSessionCreatedChannelPrefix() + "*")
+			.flatMap(this::onSessionCreatedChannelMessage)
+			.subscribe();
+		Disposable sessionDestroyedSubscription = this.keyEventsOperations
+			.listenToChannel(getSessionDeletedChannel(), getSessionExpiredChannel())
+			.flatMap(this::onKeyDestroyedMessage)
+			.subscribe();
+		this.subscriptions.addAll(Arrays.asList(sessionCreatedSubscription, sessionDestroyedSubscription));
+	}
+
+	@SuppressWarnings("unchecked")
+	private Mono<Void> onSessionCreatedChannelMessage(ReactiveSubscription.Message<String, Object> message) {
+		return Mono.just(message.getChannel())
+			.filter((channel) -> channel.startsWith(getSessionCreatedChannelPrefix()))
+			.map((channel) -> {
+				int sessionIdBeginIndex = channel.lastIndexOf(":") + 1;
+				return channel.substring(sessionIdBeginIndex);
+			})
+			.flatMap((sessionId) -> {
+				Map<String, Object> entries = (Map<String, Object>) message.getMessage();
+				return this.redisSessionMapper.apply(sessionId, entries);
+			})
+			.map((loaded) -> {
+				RedisSession session = new RedisSession(loaded, false);
+				return new SessionCreatedEvent(this, session);
+			})
+			.doOnNext(this::publishEvent)
+			.then();
+	}
+
+	private Mono<Void> onKeyDestroyedMessage(ReactiveSubscription.Message<String, String> message) {
+		return Mono.just(message.getMessage()).filter((key) -> key.startsWith(getExpiredKeyPrefix())).map((key) -> {
+			int sessionIdBeginIndex = key.lastIndexOf(":") + 1;
+			return key.substring(sessionIdBeginIndex);
+		})
+			.flatMap((sessionId) -> getSession(sessionId, true))
+			.flatMap((session) -> this.deleteById(session.getId()).thenReturn(session))
+			.map((session) -> {
+				if (message.getChannel().equals(this.sessionDeletedChannel)) {
+					return new SessionDeletedEvent(this, session);
+				}
+				return new SessionExpiredEvent(this, session);
+			})
+			.doOnNext(this::publishEvent)
+			.then();
+	}
+
+	private void publishEvent(Object event) {
+		this.eventPublisher.publishEvent(event);
+	}
+
+	/**
+	 * Sets the Redis database index used by Spring Session.
+	 * @param database the database index to use
+	 */
+	public void setDatabase(int database) {
+		this.database = database;
+		configureSessionChannels();
+	}
+
+	/**
+	 * Sets the namespace for keys used by Spring Session. Defaults to 'spring:session:'.
+	 * @param namespace the namespace to set
+	 */
+	public void setRedisKeyNamespace(String namespace) {
+		Assert.hasText(namespace, "namespace cannot be null or empty");
+		this.namespace = namespace.endsWith(":") ? namespace : namespace.trim() + ":";
+		this.indexer.setNamespace(this.namespace);
+		this.expirationStore.setNamespace(this.namespace);
+		configureSessionChannels();
+	}
+
+	/**
+	 * Sets the interval that the clean-up of expired sessions task should run. Defaults
+	 * to 60 seconds. Use {@link Duration#ZERO} to disable it.
+	 * @param cleanupInterval the interval to use
+	 */
+	public void setCleanupInterval(Duration cleanupInterval) {
+		Assert.notNull(cleanupInterval, "cleanupInterval cannot be null");
+		this.cleanupInterval = cleanupInterval;
+	}
+
+	/**
+	 * Disables the clean-up task. This is just a shortcut to invoke
+	 * {@link #setCleanupInterval(Duration)} passing {@link Duration#ZERO}
+	 */
+	public void disableCleanupTask() {
+		setCleanupInterval(Duration.ZERO);
+	}
+
+	/**
+	 * Sets the {@link Clock} to use. Defaults to {@link Clock#systemUTC()}.
+	 * @param clock the clock to use
+	 */
+	public void setClock(Clock clock) {
+		Assert.notNull(clock, "clock cannot be null");
+		this.clock = clock;
+	}
+
+	public void setDefaultMaxInactiveInterval(Duration defaultMaxInactiveInterval) {
+		Assert.notNull(defaultMaxInactiveInterval, "defaultMaxInactiveInterval must not be null");
+		this.defaultMaxInactiveInterval = defaultMaxInactiveInterval;
+	}
+
+	public void setSessionIdGenerator(SessionIdGenerator sessionIdGenerator) {
+		Assert.notNull(sessionIdGenerator, "sessionIdGenerator cannot be null");
+		this.sessionIdGenerator = sessionIdGenerator;
+	}
+
+	public void setRedisSessionMapper(BiFunction<String, Map<String, Object>, Mono<MapSession>> redisSessionMapper) {
+		Assert.notNull(redisSessionMapper, "redisSessionMapper cannot be null");
+		this.redisSessionMapper = redisSessionMapper;
+	}
+
+	public void setSaveMode(SaveMode saveMode) {
+		Assert.notNull(saveMode, "saveMode cannot be null");
+		this.saveMode = saveMode;
+	}
+
+	public ReactiveRedisOperations<String, Object> getSessionRedisOperations() {
+		return this.sessionRedisOperations;
+	}
+
+	public void setEventPublisher(ApplicationEventPublisher eventPublisher) {
+		Assert.notNull(eventPublisher, "eventPublisher cannot be null");
+		this.eventPublisher = eventPublisher;
+	}
+
+	public void setIndexResolver(IndexResolver<Session> indexResolver) {
+		Assert.notNull(indexResolver, "indexResolver cannot be null");
+		this.indexer.setIndexResolver(indexResolver);
+	}
+
+	private static String getAttributeNameWithPrefix(String attributeName) {
+		return RedisSessionMapper.ATTRIBUTE_PREFIX + attributeName;
+	}
+
+	private String getSessionKey(String sessionId) {
+		return this.namespace + "sessions:" + sessionId;
+	}
+
+	private String getExpiredKey(String sessionId) {
+		return getExpiredKeyPrefix() + sessionId;
+	}
+
+	private String getExpiredKeyPrefix() {
+		return this.expiredKeyPrefix;
+	}
+
+	private void configureSessionChannels() {
+		this.sessionCreatedChannelPrefix = this.namespace + "event:" + this.database + ":created:";
+		this.sessionDeletedChannel = "__keyevent@" + this.database + "__:del";
+		this.sessionExpiredChannel = "__keyevent@" + this.database + "__:expired";
+		this.expiredKeyPrefix = this.namespace + "sessions:expires:";
+	}
+
+	public String getSessionCreatedChannel(String sessionId) {
+		return getSessionCreatedChannelPrefix() + sessionId;
+	}
+
+	public String getSessionCreatedChannelPrefix() {
+		return this.sessionCreatedChannelPrefix;
+	}
+
+	public String getSessionDeletedChannel() {
+		return this.sessionDeletedChannel;
+	}
+
+	public String getSessionExpiredChannel() {
+		return this.sessionExpiredChannel;
+	}
+
+	public final class RedisSession implements Session {
+
+		private final MapSession cached;
+
+		private Map<String, Object> delta = new HashMap<>();
+
+		private boolean isNew;
+
+		private String originalSessionId;
+
+		private Map<String, String> indexes = new HashMap<>();
+
+		public RedisSession(MapSession cached, boolean isNew) {
+			this.cached = cached;
+			this.isNew = isNew;
+			this.originalSessionId = cached.getId();
+			if (this.isNew) {
+				this.delta.put(RedisSessionMapper.CREATION_TIME_KEY, cached.getCreationTime().toEpochMilli());
+				this.delta.put(RedisSessionMapper.MAX_INACTIVE_INTERVAL_KEY,
+						(int) cached.getMaxInactiveInterval().getSeconds());
+				this.delta.put(RedisSessionMapper.LAST_ACCESSED_TIME_KEY, cached.getLastAccessedTime().toEpochMilli());
+			}
+			if (this.isNew || (ReactiveRedisIndexedSessionRepository.this.saveMode == SaveMode.ALWAYS)) {
+				getAttributeNames().forEach((attributeName) -> this.delta.put(getAttributeNameWithPrefix(attributeName),
+						cached.getAttribute(attributeName)));
+			}
+		}
+
+		@Override
+		public String getId() {
+			return this.cached.getId();
+		}
+
+		@Override
+		public String changeSessionId() {
+			String newSessionId = ReactiveRedisIndexedSessionRepository.this.sessionIdGenerator.generate();
+			this.cached.setId(newSessionId);
+			return newSessionId;
+		}
+
+		@Override
+		public <T> T getAttribute(String attributeName) {
+			T attributeValue = this.cached.getAttribute(attributeName);
+			if (attributeValue != null
+					&& ReactiveRedisIndexedSessionRepository.this.saveMode.equals(SaveMode.ON_GET_ATTRIBUTE)) {
+				this.delta.put(getAttributeNameWithPrefix(attributeName), attributeValue);
+			}
+			return attributeValue;
+		}
+
+		@Override
+		public Set<String> getAttributeNames() {
+			return this.cached.getAttributeNames();
+		}
+
+		@Override
+		public void setAttribute(String attributeName, Object attributeValue) {
+			this.cached.setAttribute(attributeName, attributeValue);
+			this.delta.put(getAttributeNameWithPrefix(attributeName), attributeValue);
+		}
+
+		@Override
+		public void removeAttribute(String attributeName) {
+			this.cached.removeAttribute(attributeName);
+			this.delta.put(getAttributeNameWithPrefix(attributeName), null);
+		}
+
+		@Override
+		public Instant getCreationTime() {
+			return this.cached.getCreationTime();
+		}
+
+		@Override
+		public void setLastAccessedTime(Instant lastAccessedTime) {
+			this.cached.setLastAccessedTime(lastAccessedTime);
+			this.delta.put(RedisSessionMapper.LAST_ACCESSED_TIME_KEY, getLastAccessedTime().toEpochMilli());
+		}
+
+		@Override
+		public Instant getLastAccessedTime() {
+			return this.cached.getLastAccessedTime();
+		}
+
+		@Override
+		public void setMaxInactiveInterval(Duration interval) {
+			this.cached.setMaxInactiveInterval(interval);
+			this.delta.put(RedisSessionMapper.MAX_INACTIVE_INTERVAL_KEY, (int) getMaxInactiveInterval().getSeconds());
+		}
+
+		@Override
+		public Duration getMaxInactiveInterval() {
+			return this.cached.getMaxInactiveInterval();
+		}
+
+		@Override
+		public boolean isExpired() {
+			return this.cached.isExpired();
+		}
+
+		public Map<String, String> getIndexes() {
+			return Collections.unmodifiableMap(this.indexes);
+		}
+
+		private boolean hasChangedSessionId() {
+			return !getId().equals(this.originalSessionId);
+		}
+
+		private Mono<Void> save() {
+			return Mono
+				.defer(() -> saveChangeSessionId().then(saveDelta()).doOnSuccess((unused) -> this.isNew = false));
+		}
+
+		private Mono<Void> saveDelta() {
+			if (this.delta.isEmpty()) {
+				return Mono.empty();
+			}
+
+			String sessionKey = getSessionKey(getId());
+			Mono<Boolean> update = ReactiveRedisIndexedSessionRepository.this.sessionRedisOperations.opsForHash()
+				.putAll(sessionKey, new HashMap<>(this.delta));
+
+			String expiredKey = getExpiredKey(getId());
+			Mono<Boolean> setTtl;
+			Mono<Boolean> updateExpireKey = ReactiveRedisIndexedSessionRepository.this.sessionRedisOperations
+				.opsForValue()
+				.append(expiredKey, "")
+				.hasElement();
+			if (getMaxInactiveInterval().getSeconds() >= 0) {
+				Duration fiveMinutesFromActualExpiration = getMaxInactiveInterval().plus(Duration.ofMinutes(5));
+				setTtl = ReactiveRedisIndexedSessionRepository.this.sessionRedisOperations.expire(sessionKey,
+						fiveMinutesFromActualExpiration);
+				updateExpireKey = updateExpireKey
+					.flatMap((length) -> ReactiveRedisIndexedSessionRepository.this.sessionRedisOperations
+						.expire(expiredKey, getMaxInactiveInterval()));
+			}
+			else {
+				setTtl = ReactiveRedisIndexedSessionRepository.this.sessionRedisOperations.persist(sessionKey);
+				updateExpireKey = ReactiveRedisIndexedSessionRepository.this.sessionRedisOperations.delete(expiredKey)
+					.hasElement();
+			}
+
+			Mono<Void> publishCreated = Mono.empty();
+			if (this.isNew) {
+				String sessionCreatedChannelKey = getSessionCreatedChannel(getId());
+				publishCreated = ReactiveRedisIndexedSessionRepository.this.sessionRedisOperations
+					.convertAndSend(sessionCreatedChannelKey, this.delta)
+					.then();
+			}
+
+			return update.flatMap((updated) -> setTtl)
+				.then(updateExpireKey)
+				.then(publishCreated)
+				.then(Mono.fromRunnable(() -> this.delta = new HashMap<>(this.delta.size())))
+				.then();
+		}
+
+		private Mono<Void> saveChangeSessionId() {
+			if (!hasChangedSessionId()) {
+				return Mono.empty();
+			}
+
+			String sessionId = getId();
+
+			Mono<Void> replaceSessionId = Mono.fromRunnable(() -> this.originalSessionId = sessionId).then();
+
+			if (this.isNew) {
+				return Mono.from(replaceSessionId);
+			}
+			else {
+				String originalSessionKey = getSessionKey(this.originalSessionId);
+				String sessionKey = getSessionKey(sessionId);
+				String originalExpiredKey = getExpiredKey(this.originalSessionId);
+				String expiredKey = getExpiredKey(sessionId);
+
+				return renameKey(originalSessionKey, sessionKey)
+					.then(Mono.defer(() -> renameKey(originalExpiredKey, expiredKey)))
+					.then(Mono.defer(this::replaceSessionIdOnIndexes))
+					.then(Mono.defer(() -> replaceSessionId));
+			}
+
+		}
+
+		private Mono<Void> replaceSessionIdOnIndexes() {
+			return ReactiveRedisIndexedSessionRepository.this.indexer.delete(this.originalSessionId)
+				.then(ReactiveRedisIndexedSessionRepository.this.indexer.update(this));
+		}
+
+		private Mono<Void> renameKey(String oldKey, String newKey) {
+			return ReactiveRedisIndexedSessionRepository.this.sessionRedisOperations.rename(oldKey, newKey)
+				.onErrorResume((ex) -> {
+					String message = NestedExceptionUtils.getMostSpecificCause(ex).getMessage();
+					return StringUtils.startsWithIgnoreCase(message, "ERR no such key");
+				}, (ex) -> Mono.empty())
+				.then();
+		}
+
+	}
+
+	private static final class RedisSessionMapperAdapter
+			implements BiFunction<String, Map<String, Object>, Mono<MapSession>> {
+
+		private final RedisSessionMapper mapper = new RedisSessionMapper();
+
+		@Override
+		public Mono<MapSession> apply(String sessionId, Map<String, Object> map) {
+			return Mono.fromSupplier(() -> this.mapper.apply(sessionId, map));
+		}
+
+	}
+
+}

spring-session-data-redis/src/main/java/org/springframework/session/data/redis/ReactiveRedisSessionIndexer.java

@@ -0,0 +1,152 @@
+/*
+ * Copyright 2014-2023 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
+ *
+ *      https://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;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.util.function.Tuples;
+
+import org.springframework.data.redis.core.ReactiveRedisOperations;
+import org.springframework.session.DelegatingIndexResolver;
+import org.springframework.session.IndexResolver;
+import org.springframework.session.PrincipalNameIndexResolver;
+import org.springframework.session.Session;
+import org.springframework.session.data.redis.ReactiveRedisIndexedSessionRepository.RedisSession;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * Uses an {@link IndexResolver} to keep track of the indexes for a
+ * {@link ReactiveRedisIndexedSessionRepository.RedisSession}. Only updates indexes that
+ * have changed.
+ *
+ * @author Marcus da Coregio
+ */
+final class ReactiveRedisSessionIndexer {
+
+	private final ReactiveRedisOperations<String, Object> sessionRedisOperations;
+
+	private String namespace;
+
+	private IndexResolver<Session> indexResolver = new DelegatingIndexResolver<>(
+			new PrincipalNameIndexResolver<>(ReactiveRedisIndexedSessionRepository.PRINCIPAL_NAME_INDEX_NAME));
+
+	private String indexKeyPrefix;
+
+	ReactiveRedisSessionIndexer(ReactiveRedisOperations<String, Object> sessionRedisOperations, String namespace) {
+		Assert.notNull(sessionRedisOperations, "sessionRedisOperations cannot be null");
+		Assert.hasText(namespace, "namespace cannot be empty");
+		this.sessionRedisOperations = sessionRedisOperations;
+		this.namespace = namespace;
+		updateIndexKeyPrefix();
+	}
+
+	Mono<Void> update(RedisSession redisSession) {
+		return getIndexes(redisSession.getId()).map((originalIndexes) -> {
+			Map<String, String> indexes = this.indexResolver.resolveIndexesFor(redisSession);
+			Map<String, String> indexToDelete = new HashMap<>();
+			Map<String, String> indexToAdd = new HashMap<>();
+			for (Map.Entry<String, String> entry : indexes.entrySet()) {
+				if (!originalIndexes.containsKey(entry.getKey())) {
+					indexToAdd.put(entry.getKey(), entry.getValue());
+					continue;
+				}
+				if (!originalIndexes.get(entry.getKey()).equals(entry.getValue())) {
+					indexToDelete.put(entry.getKey(), originalIndexes.get(entry.getKey()));
+					indexToAdd.put(entry.getKey(), entry.getValue());
+				}
+			}
+			if (CollectionUtils.isEmpty(indexes) && !CollectionUtils.isEmpty(originalIndexes)) {
+				indexToDelete.putAll(originalIndexes);
+			}
+			return Tuples.of(indexToDelete, indexToAdd);
+		}).flatMap((indexes) -> updateIndexes(indexes.getT1(), indexes.getT2(), redisSession.getId()));
+	}
+
+	private Mono<Void> updateIndexes(Map<String, String> indexToDelete, Map<String, String> indexToAdd,
+			String sessionId) {
+		// @formatter:off
+		return Flux.fromIterable(indexToDelete.entrySet())
+			.flatMap((entry) -> {
+				String indexKey = getIndexKey(entry.getKey(), entry.getValue());
+				return removeSessionFromIndex(indexKey, sessionId).thenReturn(indexKey);
+			})
+			.flatMap((indexKey) -> this.sessionRedisOperations.opsForSet().remove(getSessionIndexesKey(sessionId), indexKey))
+			.thenMany(Flux.fromIterable(indexToAdd.entrySet()))
+			.flatMap((entry) -> {
+				String indexKey = getIndexKey(entry.getKey(), entry.getValue());
+				return this.sessionRedisOperations.opsForSet().add(indexKey, sessionId).thenReturn(indexKey);
+			})
+			.flatMap((indexKey) -> this.sessionRedisOperations.opsForSet().add(getSessionIndexesKey(sessionId), indexKey))
+			.then();
+		// @formatter:on
+	}
+
+	Mono<Void> delete(String sessionId) {
+		String sessionIndexesKey = getSessionIndexesKey(sessionId);
+		return this.sessionRedisOperations.opsForSet()
+			.members(sessionIndexesKey)
+			.flatMap((indexKey) -> removeSessionFromIndex((String) indexKey, sessionId))
+			.then(this.sessionRedisOperations.delete(sessionIndexesKey))
+			.then();
+	}
+
+	private Mono<Void> removeSessionFromIndex(String indexKey, String sessionId) {
+		return this.sessionRedisOperations.opsForSet().remove(indexKey, sessionId).then();
+	}
+
+	Mono<Map<String, String>> getIndexes(String sessionId) {
+		String sessionIndexesKey = getSessionIndexesKey(sessionId);
+		return this.sessionRedisOperations.opsForSet()
+			.members(sessionIndexesKey)
+			.cast(String.class)
+			.collectMap((indexKey) -> indexKey.substring(this.indexKeyPrefix.length()).split(":")[0],
+					(indexKey) -> indexKey.substring(this.indexKeyPrefix.length()).split(":")[1]);
+	}
+
+	Flux<String> getSessionIds(String indexName, String indexValue) {
+		String indexKey = getIndexKey(indexName, indexValue);
+		return this.sessionRedisOperations.opsForSet().members(indexKey).cast(String.class);
+	}
+
+	private void updateIndexKeyPrefix() {
+		this.indexKeyPrefix = this.namespace + "sessions:index:";
+	}
+
+	private String getSessionIndexesKey(String sessionId) {
+		return this.namespace + "sessions:" + sessionId + ":idx";
+	}
+
+	private String getIndexKey(String indexName, String indexValue) {
+		return this.indexKeyPrefix + indexName + ":" + indexValue;
+	}
+
+	void setNamespace(String namespace) {
+		Assert.hasText(namespace, "namespace cannot be empty");
+		this.namespace = namespace;
+		updateIndexKeyPrefix();
+	}
+
+	void setIndexResolver(IndexResolver<Session> indexResolver) {
+		Assert.notNull(indexResolver, "indexResolver cannot be null");
+		this.indexResolver = indexResolver;
+	}
+
+}

spring-session-data-redis/src/main/java/org/springframework/session/data/redis/SortedSetReactiveRedisSessionExpirationStore.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2014-2023 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
+ *
+ *      https://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;
+
+import java.time.Instant;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.data.domain.Range;
+import org.springframework.data.redis.connection.Limit;
+import org.springframework.data.redis.core.ReactiveRedisOperations;
+import org.springframework.util.Assert;
+
+/**
+ * Uses a sorted set to store the expiration times for sessions. The score of each entry
+ * is the expiration time of the session. The value is the session id.
+ *
+ * @author Marcus da Coregio
+ */
+final class SortedSetReactiveRedisSessionExpirationStore {
+
+	private final ReactiveRedisOperations<String, Object> sessionRedisOperations;
+
+	private String namespace;
+
+	private int retrieveCount = 100;
+
+	SortedSetReactiveRedisSessionExpirationStore(ReactiveRedisOperations<String, Object> sessionRedisOperations,
+			String namespace) {
+		Assert.notNull(sessionRedisOperations, "sessionRedisOperations cannot be null");
+		Assert.hasText(namespace, "namespace cannot be null or empty");
+		this.sessionRedisOperations = sessionRedisOperations;
+		this.namespace = namespace;
+	}
+
+	/**
+	 * Add the session id associated with the expiration time into the sorted set.
+	 * @param sessionId the session id
+	 * @param expiration the expiration time
+	 * @return a {@link Mono} that completes when the operation completes
+	 */
+	Mono<Void> add(String sessionId, Instant expiration) {
+		long expirationInMillis = expiration.toEpochMilli();
+		return this.sessionRedisOperations.opsForZSet().add(getExpirationsKey(), sessionId, expirationInMillis).then();
+	}
+
+	/**
+	 * Remove the session id from the sorted set.
+	 * @param sessionId the session id
+	 * @return a {@link Mono} that completes when the operation completes
+	 */
+	Mono<Void> remove(String sessionId) {
+		return this.sessionRedisOperations.opsForZSet().remove(getExpirationsKey(), sessionId).then();
+	}
+
+	/**
+	 * Retrieve the session ids that have the expiration time less than the value passed
+	 * in {@code expiredBefore}.
+	 * @param expiredBefore the expiration time
+	 * @return a {@link Flux} that emits the session ids
+	 */
+	Flux<String> retrieveExpiredSessions(Instant expiredBefore) {
+		Range<Double> range = Range.closed(0D, (double) expiredBefore.toEpochMilli());
+		Limit limit = Limit.limit().count(this.retrieveCount);
+		return this.sessionRedisOperations.opsForZSet()
+			.reverseRangeByScore(getExpirationsKey(), range, limit)
+			.cast(String.class);
+	}
+
+	private String getExpirationsKey() {
+		return this.namespace + "sessions:expirations";
+	}
+
+	/**
+	 * Set the namespace for the keys used by this class.
+	 * @param namespace the namespace
+	 */
+	void setNamespace(String namespace) {
+		Assert.hasText(namespace, "namespace cannot be null or empty");
+		this.namespace = namespace;
+	}
+
+}

spring-session-data-redis/src/main/java/org/springframework/session/data/redis/config/ConfigureReactiveRedisAction.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2014-2023 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
+ *
+ *      https://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;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.data.redis.connection.ReactiveRedisConnection;
+
+/**
+ * Allows specifying a strategy for configuring and validating Redis using a Reactive
+ * connection.
+ *
+ * @author Marcus da Coregio
+ * @since 3.3
+ */
+public interface ConfigureReactiveRedisAction {
+
+	Mono<Void> configure(ReactiveRedisConnection connection);
+
+	/**
+	 * An implementation of {@link ConfigureReactiveRedisAction} that does nothing.
+	 */
+	ConfigureReactiveRedisAction NO_OP = (connection) -> Mono.empty();
+
+}

spring-session-data-redis/src/main/java/org/springframework/session/data/redis/config/annotation/ConfigureNotifyKeyspaceEventsReactiveAction.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2014-2023 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
+ *
+ *      https://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;
+
+import java.util.Properties;
+import java.util.function.Predicate;
+
+import reactor.core.publisher.Mono;
+import reactor.util.function.Tuples;
+
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.redis.connection.ReactiveRedisConnection;
+import org.springframework.session.data.redis.config.ConfigureReactiveRedisAction;
+
+/**
+ * <p>
+ * Ensures that Redis Keyspace events for Generic commands and Expired events are enabled.
+ * For example, it might set the following:
+ * </p>
+ *
+ * <pre>
+ * config set notify-keyspace-events Egx
+ * </pre>
+ *
+ * <p>
+ * This strategy will not work if the Redis instance has been properly secured. Instead,
+ * the Redis instance should be configured externally and a Bean of type
+ * {@link ConfigureReactiveRedisAction#NO_OP} should be exposed.
+ * </p>
+ *
+ * @author Rob Winch
+ * @author Mark Paluch
+ * @author Marcus da Coregio
+ * @since 3.3
+ */
+public class ConfigureNotifyKeyspaceEventsReactiveAction implements ConfigureReactiveRedisAction {
+
+	static final String CONFIG_NOTIFY_KEYSPACE_EVENTS = "notify-keyspace-events";
+
+	@Override
+	public Mono<Void> configure(ReactiveRedisConnection connection) {
+		return getNotifyOptions(connection).map((notifyOptions) -> {
+			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";
+			}
+			return Tuples.of(notifyOptions, customizedNotifyOptions);
+		})
+			.filter((optionsTuple) -> !optionsTuple.getT1().equals(optionsTuple.getT2()))
+			.flatMap((optionsTuple) -> connection.serverCommands()
+				.setConfig(CONFIG_NOTIFY_KEYSPACE_EVENTS, optionsTuple.getT2()))
+			.filter("OK"::equals)
+			.doFinally((unused) -> connection.close())
+			.then();
+	}
+
+	private Mono<String> getNotifyOptions(ReactiveRedisConnection connection) {
+		return connection.serverCommands()
+			.getConfig(CONFIG_NOTIFY_KEYSPACE_EVENTS)
+			.filter(Predicate.not(Properties::isEmpty))
+			.map((config) -> config.getProperty(config.stringPropertyNames().iterator().next()))
+			.onErrorMap(InvalidDataAccessApiUsageException.class,
+					(ex) -> new IllegalStateException("Unable to configure Reactive Redis to keyspace notifications",
+							ex));
+	}
+
+}

spring-session-data-redis/src/main/java/org/springframework/session/data/redis/config/annotation/web/server/AbstractRedisWebSessionConfiguration.java

@@ -0,0 +1,148 @@
+/*
+ * Copyright 2014-2023 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
+ *
+ *      https://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.server;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
+import org.springframework.data.redis.core.ReactiveRedisTemplate;
+import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
+import org.springframework.data.redis.serializer.RedisSerializationContext;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.session.MapSession;
+import org.springframework.session.ReactiveSessionRepository;
+import org.springframework.session.SaveMode;
+import org.springframework.session.Session;
+import org.springframework.session.SessionIdGenerator;
+import org.springframework.session.UuidSessionIdGenerator;
+import org.springframework.session.config.ReactiveSessionRepositoryCustomizer;
+import org.springframework.session.config.annotation.web.server.SpringWebSessionConfiguration;
+import org.springframework.session.data.redis.ReactiveRedisSessionRepository;
+import org.springframework.session.data.redis.config.annotation.SpringSessionRedisConnectionFactory;
+import org.springframework.util.Assert;
+
+@Configuration(proxyBeanMethods = false)
+@Import(SpringWebSessionConfiguration.class)
+public abstract class AbstractRedisWebSessionConfiguration<T extends ReactiveSessionRepository<? extends Session>> {
+
+	private Duration maxInactiveInterval = MapSession.DEFAULT_MAX_INACTIVE_INTERVAL;
+
+	private String redisNamespace = ReactiveRedisSessionRepository.DEFAULT_NAMESPACE;
+
+	private SaveMode saveMode = SaveMode.ON_SET_ATTRIBUTE;
+
+	private ReactiveRedisConnectionFactory redisConnectionFactory;
+
+	private RedisSerializer<Object> defaultRedisSerializer = new JdkSerializationRedisSerializer();
+
+	private List<ReactiveSessionRepositoryCustomizer<T>> sessionRepositoryCustomizers;
+
+	private SessionIdGenerator sessionIdGenerator = UuidSessionIdGenerator.getInstance();
+
+	public abstract T sessionRepository();
+
+	public void setMaxInactiveInterval(Duration maxInactiveInterval) {
+		this.maxInactiveInterval = maxInactiveInterval;
+	}
+
+	public void setRedisNamespace(String namespace) {
+		Assert.hasText(namespace, "namespace cannot be empty or null");
+		this.redisNamespace = namespace;
+	}
+
+	public void setSaveMode(SaveMode saveMode) {
+		Assert.notNull(saveMode, "saveMode cannot be null");
+		this.saveMode = saveMode;
+	}
+
+	public Duration getMaxInactiveInterval() {
+		return this.maxInactiveInterval;
+	}
+
+	public String getRedisNamespace() {
+		return this.redisNamespace;
+	}
+
+	public SaveMode getSaveMode() {
+		return this.saveMode;
+	}
+
+	public SessionIdGenerator getSessionIdGenerator() {
+		return this.sessionIdGenerator;
+	}
+
+	public RedisSerializer<Object> getDefaultRedisSerializer() {
+		return this.defaultRedisSerializer;
+	}
+
+	@Autowired
+	public void setRedisConnectionFactory(
+			@SpringSessionRedisConnectionFactory ObjectProvider<ReactiveRedisConnectionFactory> springSessionRedisConnectionFactory,
+			ObjectProvider<ReactiveRedisConnectionFactory> redisConnectionFactory) {
+		ReactiveRedisConnectionFactory redisConnectionFactoryToUse = springSessionRedisConnectionFactory
+			.getIfAvailable();
+		if (redisConnectionFactoryToUse == null) {
+			redisConnectionFactoryToUse = redisConnectionFactory.getObject();
+		}
+		this.redisConnectionFactory = redisConnectionFactoryToUse;
+	}
+
+	@Autowired(required = false)
+	@Qualifier("springSessionDefaultRedisSerializer")
+	public void setDefaultRedisSerializer(RedisSerializer<Object> defaultRedisSerializer) {
+		this.defaultRedisSerializer = defaultRedisSerializer;
+	}
+
+	@Autowired(required = false)
+	public void setSessionRepositoryCustomizer(
+			ObjectProvider<ReactiveSessionRepositoryCustomizer<T>> sessionRepositoryCustomizers) {
+		this.sessionRepositoryCustomizers = sessionRepositoryCustomizers.orderedStream().collect(Collectors.toList());
+	}
+
+	protected List<ReactiveSessionRepositoryCustomizer<T>> getSessionRepositoryCustomizers() {
+		return this.sessionRepositoryCustomizers;
+	}
+
+	protected ReactiveRedisTemplate<String, Object> createReactiveRedisTemplate() {
+		RedisSerializer<String> keySerializer = RedisSerializer.string();
+		RedisSerializer<Object> defaultSerializer = (this.defaultRedisSerializer != null) ? this.defaultRedisSerializer
+				: new JdkSerializationRedisSerializer();
+		RedisSerializationContext<String, Object> serializationContext = RedisSerializationContext
+			.<String, Object>newSerializationContext(defaultSerializer)
+			.key(keySerializer)
+			.hashKey(keySerializer)
+			.build();
+		return new ReactiveRedisTemplate<>(this.redisConnectionFactory, serializationContext);
+	}
+
+	public ReactiveRedisConnectionFactory getRedisConnectionFactory() {
+		return this.redisConnectionFactory;
+	}
+
+	@Autowired(required = false)
+	public void setSessionIdGenerator(SessionIdGenerator sessionIdGenerator) {
+		this.sessionIdGenerator = sessionIdGenerator;
+	}
+
+}

spring-session-data-redis/src/main/java/org/springframework/session/data/redis/config/annotation/web/server/EnableRedisIndexedWebSession.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2014-2023 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
+ *
+ *      https://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.server;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.context.annotation.Import;
+import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
+import org.springframework.session.MapSession;
+import org.springframework.session.SaveMode;
+import org.springframework.session.data.redis.ReactiveRedisIndexedSessionRepository;
+import org.springframework.web.server.session.WebSessionManager;
+
+/**
+ * Add this annotation to an {@link org.springframework.context.annotation.Configuration}
+ * class to expose the {@link WebSessionManager} as a bean named {@code webSessionManager}
+ * and backed by Reactive Redis. In order to leverage the annotation, a single
+ * {@link ReactiveRedisConnectionFactory} must be provided. For example:
+ *
+ * <pre class="code">
+ * &#064;Configuration(proxyBeanMethods = false)
+ * &#064;EnableRedisIndexedWebSession
+ * public class RedisIndexedWebSessionConfig {
+ *
+ *     &#064;Bean
+ *     public LettuceConnectionFactory redisConnectionFactory() {
+ *         return new LettuceConnectionFactory();
+ *     }
+ *
+ * }
+ * </pre>
+ *
+ * More advanced configurations can extend {@link RedisIndexedWebSessionConfiguration}
+ * instead.
+ *
+ * @author Marcus da Coregio
+ * @since 3.3
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+@Documented
+@Import(RedisIndexedWebSessionConfiguration.class)
+public @interface EnableRedisIndexedWebSession {
+
+	/**
+	 * The session timeout in seconds. By default, it is set to 1800 seconds (30 minutes).
+	 * A negative number means permanently valid.
+	 * @return the seconds a session can be inactive before expiring
+	 */
+	int maxInactiveIntervalInSeconds() default MapSession.DEFAULT_MAX_INACTIVE_INTERVAL_SECONDS;
+
+	/**
+	 * Defines a unique namespace for keys. The value is used to isolate sessions by
+	 * changing the prefix from default {@code spring:session:} to
+	 * {@code <redisNamespace>:}.
+	 * <p>
+	 * For example, if you had an application named "Application A" that needed to keep
+	 * the sessions isolated from "Application B" you could set two different values for
+	 * the applications and they could function within the same Redis instance.
+	 * @return the unique namespace for keys
+	 */
+	String redisNamespace() default ReactiveRedisIndexedSessionRepository.DEFAULT_NAMESPACE;
+
+	/**
+	 * Save mode for the session. The default is {@link SaveMode#ON_SET_ATTRIBUTE}, which
+	 * only saves changes made to session.
+	 * @return the save mode
+	 */
+	SaveMode saveMode() default SaveMode.ON_SET_ATTRIBUTE;
+
+}

spring-session-data-redis/src/main/java/org/springframework/session/data/redis/config/annotation/web/server/RedisIndexedWebSessionConfiguration.java

@@ -0,0 +1,205 @@
+/*
+ * Copyright 2014-2022 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
+ *
+ *      https://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.server;
+
+import java.time.Duration;
+import java.util.Map;
+
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.EmbeddedValueResolverAware;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.ImportAware;
+import org.springframework.core.annotation.AnnotationAttributes;
+import org.springframework.core.type.AnnotationMetadata;
+import org.springframework.data.redis.connection.ReactiveRedisConnection;
+import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.core.ReactiveRedisTemplate;
+import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
+import org.springframework.session.IndexResolver;
+import org.springframework.session.Session;
+import org.springframework.session.data.redis.ReactiveRedisIndexedSessionRepository;
+import org.springframework.session.data.redis.config.ConfigureReactiveRedisAction;
+import org.springframework.session.data.redis.config.annotation.ConfigureNotifyKeyspaceEventsReactiveAction;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.StringUtils;
+import org.springframework.util.StringValueResolver;
+import org.springframework.web.server.session.WebSessionManager;
+
+/**
+ * Exposes the {@link WebSessionManager} as a bean named {@code webSessionManager} backed
+ * by {@link ReactiveRedisIndexedSessionRepository}. In order to use this a single
+ * {@link ReactiveRedisConnectionFactory} must be exposed as a Bean.
+ *
+ * @author Marcus da Coregio
+ * @since 3.3
+ * @see EnableRedisIndexedWebSession
+ */
+@Configuration(proxyBeanMethods = false)
+public class RedisIndexedWebSessionConfiguration
+		extends AbstractRedisWebSessionConfiguration<ReactiveRedisIndexedSessionRepository>
+		implements EmbeddedValueResolverAware, ImportAware {
+
+	private static final boolean lettucePresent;
+
+	private static final boolean jedisPresent;
+
+	private ConfigureReactiveRedisAction configureRedisAction = new ConfigureNotifyKeyspaceEventsReactiveAction();
+
+	private StringValueResolver embeddedValueResolver;
+
+	private ApplicationEventPublisher eventPublisher;
+
+	private IndexResolver<Session> indexResolver;
+
+	static {
+		ClassLoader classLoader = RedisIndexedWebSessionConfiguration.class.getClassLoader();
+		lettucePresent = ClassUtils.isPresent("io.lettuce.core.RedisClient", classLoader);
+		jedisPresent = ClassUtils.isPresent("redis.clients.jedis.Jedis", classLoader);
+	}
+
+	@Override
+	@Bean
+	public ReactiveRedisIndexedSessionRepository sessionRepository() {
+		ReactiveRedisTemplate<String, Object> reactiveRedisTemplate = createReactiveRedisTemplate();
+		ReactiveRedisIndexedSessionRepository sessionRepository = new ReactiveRedisIndexedSessionRepository(
+				reactiveRedisTemplate, createReactiveStringRedisTemplate());
+		sessionRepository.setDefaultMaxInactiveInterval(getMaxInactiveInterval());
+		sessionRepository.setEventPublisher(this.eventPublisher);
+		if (this.indexResolver != null) {
+			sessionRepository.setIndexResolver(this.indexResolver);
+		}
+		if (StringUtils.hasText(getRedisNamespace())) {
+			sessionRepository.setRedisKeyNamespace(getRedisNamespace());
+		}
+		int database = resolveDatabase();
+		sessionRepository.setDatabase(database);
+		sessionRepository.setSaveMode(getSaveMode());
+		sessionRepository.setSessionIdGenerator(getSessionIdGenerator());
+		if (getSessionRepositoryCustomizers() != null) {
+			getSessionRepositoryCustomizers().forEach((customizer) -> customizer.customize(sessionRepository));
+		}
+		return sessionRepository;
+	}
+
+	private ReactiveStringRedisTemplate createReactiveStringRedisTemplate() {
+		return new ReactiveStringRedisTemplate(getRedisConnectionFactory());
+	}
+
+	@Bean
+	public InitializingBean enableRedisKeyspaceNotificationsInitializer() {
+		return new EnableRedisKeyspaceNotificationsInitializer(getRedisConnectionFactory(), this.configureRedisAction);
+	}
+
+	@Autowired
+	public void setEventPublisher(ApplicationEventPublisher eventPublisher) {
+		this.eventPublisher = eventPublisher;
+	}
+
+	/**
+	 * Sets the action to perform for configuring Redis.
+	 * @param configureRedisAction the configuration to apply to Redis. The default is
+	 * {@link ConfigureNotifyKeyspaceEventsReactiveAction}
+	 */
+	@Autowired(required = false)
+	public void setConfigureRedisAction(ConfigureReactiveRedisAction configureRedisAction) {
+		this.configureRedisAction = configureRedisAction;
+	}
+
+	@Autowired(required = false)
+	public void setIndexResolver(IndexResolver<Session> indexResolver) {
+		this.indexResolver = indexResolver;
+	}
+
+	@Override
+	public void setEmbeddedValueResolver(StringValueResolver resolver) {
+		this.embeddedValueResolver = resolver;
+	}
+
+	@Override
+	public void setImportMetadata(AnnotationMetadata importMetadata) {
+		Map<String, Object> attributeMap = importMetadata
+			.getAnnotationAttributes(EnableRedisIndexedWebSession.class.getName());
+		AnnotationAttributes attributes = AnnotationAttributes.fromMap(attributeMap);
+		if (attributes == null) {
+			return;
+		}
+		setMaxInactiveInterval(Duration.ofSeconds(attributes.<Integer>getNumber("maxInactiveIntervalInSeconds")));
+		String redisNamespaceValue = attributes.getString("redisNamespace");
+		if (StringUtils.hasText(redisNamespaceValue)) {
+			setRedisNamespace(this.embeddedValueResolver.resolveStringValue(redisNamespaceValue));
+		}
+		setSaveMode(attributes.getEnum("saveMode"));
+	}
+
+	private int resolveDatabase() {
+		if (lettucePresent && getRedisConnectionFactory() instanceof LettuceConnectionFactory lettuce) {
+			return lettuce.getDatabase();
+		}
+		if (jedisPresent && getRedisConnectionFactory() instanceof JedisConnectionFactory jedis) {
+			return jedis.getDatabase();
+		}
+		return ReactiveRedisIndexedSessionRepository.DEFAULT_DATABASE;
+	}
+
+	/**
+	 * 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 {
+
+		private final ReactiveRedisConnectionFactory connectionFactory;
+
+		private final ConfigureReactiveRedisAction configure;
+
+		EnableRedisKeyspaceNotificationsInitializer(ReactiveRedisConnectionFactory connectionFactory,
+				ConfigureReactiveRedisAction configure) {
+			this.connectionFactory = connectionFactory;
+			this.configure = configure;
+		}
+
+		@Override
+		public void afterPropertiesSet() {
+			if (this.configure == ConfigureReactiveRedisAction.NO_OP) {
+				return;
+			}
+			ReactiveRedisConnection connection = this.connectionFactory.getReactiveConnection();
+			try {
+				this.configure.configure(connection).block();
+			}
+			finally {
+				try {
+					connection.close();
+				}
+				catch (Exception ex) {
+					LogFactory.getLog(getClass()).error("Error closing RedisConnection", ex);
+				}
+			}
+		}
+
+	}
+
+}

spring-session-data-redis/src/test/java/org/springframework/session/data/redis/ReactiveRedisSessionIndexerTests.java

@@ -0,0 +1,197 @@
+/*
+ * Copyright 2014-2023 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
+ *
+ *      https://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;
+
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Answers;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.data.redis.core.ReactiveRedisOperations;
+import org.springframework.session.DelegatingIndexResolver;
+import org.springframework.session.IndexResolver;
+import org.springframework.session.PrincipalNameIndexResolver;
+import org.springframework.session.ReactiveFindByIndexNameSessionRepository;
+import org.springframework.session.Session;
+import org.springframework.session.SingleIndexResolver;
+import org.springframework.session.data.redis.ReactiveRedisIndexedSessionRepository.RedisSession;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Tests for {@link ReactiveRedisSessionIndexer}.
+ *
+ * @author Marcus da Coregio
+ */
+class ReactiveRedisSessionIndexerTests {
+
+	ReactiveRedisSessionIndexer indexer;
+
+	ReactiveRedisOperations<String, Object> sessionRedisOperations = mock(Answers.RETURNS_DEEP_STUBS);
+
+	String indexKeyPrefix = "spring:session:sessions:index:";
+
+	@BeforeEach
+	void setup() {
+		this.indexer = new ReactiveRedisSessionIndexer(this.sessionRedisOperations, "spring:session:");
+	}
+
+	@Test
+	void getIndexesWhenIndexKeyExistsThenReturnsIndexNameAndValue() {
+		List<Object> indexKeys = List.of(this.indexKeyPrefix + "principalName:user",
+				this.indexKeyPrefix + "index_name:index_value");
+		given(this.sessionRedisOperations.opsForSet().members(anyString())).willReturn(Flux.fromIterable(indexKeys));
+		Map<String, String> indexes = this.indexer.getIndexes("1234").block();
+		assertThat(indexes).hasSize(2)
+			.containsEntry("principalName", "user")
+			.containsEntry("index_name", "index_value");
+	}
+
+	@Test
+	void deleteWhenSessionIdHasIndexesThenRemoveSessionIdFromIndexesAndDeleteSessionIndexKey() {
+		String index1 = this.indexKeyPrefix + "principalName:user";
+		String index2 = this.indexKeyPrefix + "index_name:index_value";
+		List<Object> indexKeys = List.of(index1, index2);
+		given(this.sessionRedisOperations.opsForSet().members(anyString())).willReturn(Flux.fromIterable(indexKeys));
+		given(this.sessionRedisOperations.delete(anyString())).willReturn(Mono.just(1L));
+		given(this.sessionRedisOperations.opsForSet().remove(anyString(), anyString())).willReturn(Mono.just(1L));
+		this.indexer.delete("1234").block();
+		verify(this.sessionRedisOperations).delete("spring:session:sessions:1234:idx");
+		verify(this.sessionRedisOperations.opsForSet()).remove(index1, "1234");
+		verify(this.sessionRedisOperations.opsForSet()).remove(index2, "1234");
+	}
+
+	@Test
+	void updateWhenSessionHasNoIndexesSavedThenUpdates() {
+		RedisSession session = mock();
+		given(session.getAttribute(ReactiveFindByIndexNameSessionRepository.PRINCIPAL_NAME_INDEX_NAME))
+			.willReturn("user");
+		given(session.getId()).willReturn("1234");
+		given(this.sessionRedisOperations.opsForSet().members(anyString())).willReturn(Flux.empty());
+		given(this.sessionRedisOperations.opsForSet().add(anyString(), anyString())).willReturn(Mono.just(1L));
+		this.indexer.update(session).block();
+		verify(this.sessionRedisOperations.opsForSet()).add(this.indexKeyPrefix + "PRINCIPAL_NAME_INDEX_NAME:user",
+				"1234");
+	}
+
+	@Test
+	void updateWhenSessionIndexesSavedWithSameValueThenDoesNotUpdate() {
+		String indexKey = this.indexKeyPrefix + ReactiveFindByIndexNameSessionRepository.PRINCIPAL_NAME_INDEX_NAME
+				+ ":user";
+		RedisSession session = mock();
+		given(session.getAttribute(ReactiveFindByIndexNameSessionRepository.PRINCIPAL_NAME_INDEX_NAME))
+			.willReturn("user");
+		given(session.getId()).willReturn("1234");
+		given(this.sessionRedisOperations.opsForSet().members(anyString()))
+			.willReturn(Flux.fromIterable(List.of(indexKey)));
+		this.indexer.update(session).block();
+		verify(this.sessionRedisOperations.opsForSet(), never()).add(anyString(), anyString());
+		verify(this.sessionRedisOperations.opsForSet(), never()).remove(anyString(), anyString());
+	}
+
+	@Test
+	void updateWhenSessionIndexesSavedWithDifferentValueThenUpdates() {
+		String indexKey = this.indexKeyPrefix + ReactiveFindByIndexNameSessionRepository.PRINCIPAL_NAME_INDEX_NAME
+				+ ":user";
+		RedisSession session = mock();
+		given(session.getAttribute(ReactiveFindByIndexNameSessionRepository.PRINCIPAL_NAME_INDEX_NAME))
+			.willReturn("newuser");
+		given(session.getId()).willReturn("1234");
+		given(this.sessionRedisOperations.opsForSet().members(anyString()))
+			.willReturn(Flux.fromIterable(List.of(indexKey)));
+		given(this.sessionRedisOperations.opsForSet().add(anyString(), anyString())).willReturn(Mono.just(1L));
+		given(this.sessionRedisOperations.opsForSet().remove(anyString(), anyString())).willReturn(Mono.just(1L));
+		this.indexer.update(session).block();
+		verify(this.sessionRedisOperations.opsForSet()).add(
+				this.indexKeyPrefix + ReactiveFindByIndexNameSessionRepository.PRINCIPAL_NAME_INDEX_NAME + ":newuser",
+				"1234");
+		verify(this.sessionRedisOperations.opsForSet()).remove(indexKey, "1234");
+	}
+
+	@Test
+	void updateWhenMultipleIndexResolvedThenUpdated() {
+		IndexResolver<Session> indexResolver = new DelegatingIndexResolver<>(
+				new PrincipalNameIndexResolver<>(ReactiveFindByIndexNameSessionRepository.PRINCIPAL_NAME_INDEX_NAME),
+				new TestIndexResolver<>("test"));
+		this.indexer.setIndexResolver(indexResolver);
+		RedisSession session = mock();
+		given(session.getAttribute(ReactiveFindByIndexNameSessionRepository.PRINCIPAL_NAME_INDEX_NAME))
+			.willReturn("user");
+		given(session.getAttribute("test")).willReturn("testvalue");
+		given(session.getId()).willReturn("1234");
+		given(this.sessionRedisOperations.opsForSet().members(anyString())).willReturn(Flux.empty());
+		given(this.sessionRedisOperations.opsForSet().add(anyString(), anyString())).willReturn(Mono.just(1L));
+		this.indexer.update(session).block();
+		verify(this.sessionRedisOperations.opsForSet()).add(this.indexKeyPrefix + "PRINCIPAL_NAME_INDEX_NAME:user",
+				"1234");
+		verify(this.sessionRedisOperations.opsForSet()).add(this.indexKeyPrefix + "test:testvalue", "1234");
+	}
+
+	@Test
+	void setNamespaceShouldUpdateIndexKeyPrefix() {
+		String originalPrefix = (String) ReflectionTestUtils.getField(this.indexer, "indexKeyPrefix");
+		this.indexer.setNamespace("my:namespace:");
+		String updatedPrefix = (String) ReflectionTestUtils.getField(this.indexer, "indexKeyPrefix");
+		assertThat(originalPrefix).isEqualTo(this.indexKeyPrefix);
+		assertThat(updatedPrefix).isEqualTo("my:namespace:sessions:index:");
+	}
+
+	@Test
+	void constructorWhenSessionRedisOperationsNullThenException() {
+		assertThatIllegalArgumentException().isThrownBy(() -> new ReactiveRedisSessionIndexer(null, "spring:session:"))
+			.withMessage("sessionRedisOperations cannot be null");
+	}
+
+	@Test
+	void constructorWhenNamespaceNullThenException() {
+		assertThatIllegalArgumentException()
+			.isThrownBy(() -> new ReactiveRedisSessionIndexer(this.sessionRedisOperations, null))
+			.withMessage("namespace cannot be empty");
+	}
+
+	@Test
+	void constructorWhenNamespaceEmptyThenException() {
+		assertThatIllegalArgumentException()
+			.isThrownBy(() -> new ReactiveRedisSessionIndexer(this.sessionRedisOperations, ""))
+			.withMessage("namespace cannot be empty");
+	}
+
+	static class TestIndexResolver<S extends Session> extends SingleIndexResolver<S> {
+
+		protected TestIndexResolver(String indexName) {
+			super(indexName);
+		}
+
+		@Override
+		public String resolveIndexValueFor(S session) {
+			return session.getAttribute(getIndexName());
+		}
+
+	}
+
+}

spring-session-data-redis/src/test/java/org/springframework/session/data/redis/SortedSetReactiveRedisSessionExpirationStoreTests.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2014-2023 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
+ *
+ *      https://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;
+
+import java.time.Instant;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Answers;
+import org.mockito.ArgumentCaptor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import org.springframework.data.domain.Range;
+import org.springframework.data.redis.connection.Limit;
+import org.springframework.data.redis.core.ReactiveRedisOperations;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyDouble;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+@SuppressWarnings("unchecked")
+class SortedSetReactiveRedisSessionExpirationStoreTests {
+
+	SortedSetReactiveRedisSessionExpirationStore store;
+
+	ReactiveRedisOperations<String, Object> sessionRedisOperations = mock(Answers.RETURNS_DEEP_STUBS);
+
+	String namespace = "spring:session:";
+
+	@BeforeEach
+	void setup() {
+		this.store = new SortedSetReactiveRedisSessionExpirationStore(this.sessionRedisOperations, this.namespace);
+		given(this.sessionRedisOperations.opsForZSet().add(anyString(), anyString(), anyDouble()))
+			.willReturn(Mono.empty());
+		given(this.sessionRedisOperations.opsForZSet().remove(anyString(), anyString())).willReturn(Mono.empty());
+		given(this.sessionRedisOperations.opsForZSet()
+			.reverseRangeByScore(anyString(), any(Range.class), any(Limit.class))).willReturn(Flux.empty());
+	}
+
+	@Test
+	void addThenStoresSessionIdRankedByExpireAtEpochMilli() {
+		String sessionId = "1234";
+		Instant expireAt = Instant.ofEpochMilli(1702314490000L);
+		StepVerifier.create(this.store.add(sessionId, expireAt)).verifyComplete();
+		verify(this.sessionRedisOperations.opsForZSet()).add(this.namespace + "sessions:expirations", sessionId,
+				expireAt.toEpochMilli());
+	}
+
+	@Test
+	void removeThenRemovesSessionIdFromSortedSet() {
+		String sessionId = "1234";
+		StepVerifier.create(this.store.remove(sessionId)).verifyComplete();
+		verify(this.sessionRedisOperations.opsForZSet()).remove(this.namespace + "sessions:expirations", sessionId);
+	}
+
+	@Test
+	void retrieveExpiredSessionsThenUsesExpectedRangeAndLimit() {
+		Instant now = Instant.now();
+		StepVerifier.create(this.store.retrieveExpiredSessions(now)).verifyComplete();
+		ArgumentCaptor<Range<Double>> rangeCaptor = ArgumentCaptor.forClass(Range.class);
+		ArgumentCaptor<Limit> limitCaptor = ArgumentCaptor.forClass(Limit.class);
+		verify(this.sessionRedisOperations.opsForZSet()).reverseRangeByScore(
+				eq(this.namespace + "sessions:expirations"), rangeCaptor.capture(), limitCaptor.capture());
+		assertThat(rangeCaptor.getValue().getLowerBound().getValue()).hasValue(0D);
+		assertThat(rangeCaptor.getValue().getUpperBound().getValue()).hasValue((double) now.toEpochMilli());
+		assertThat(limitCaptor.getValue().getCount()).isEqualTo(100);
+		assertThat(limitCaptor.getValue().getOffset()).isEqualTo(0);
+	}
+
+}