Add Redis implementation of ReactorSessionRepository

Closes gh-816

samples/javaconfig/webflux/spring-session-sample-javaconfig-webflux.gradle

@@ -1,7 +1,7 @@
 apply plugin: 'io.spring.convention.spring-sample'
 
 dependencies {
-	compile project(':spring-session-core')
+	compile project(':spring-session-data-redis')
 	compile 'io.lettuce:lettuce-core'
 	compile 'io.netty:netty-buffer'
 	compile 'io.projectreactor.ipc:reactor-netty'
@@ -14,6 +14,7 @@ dependencies {
 	compile 'org.webjars:webjars-taglib'
 	compile jstlDependencies
 	compile slf4jDependencies
+	compile 'org.testcontainers:testcontainers'
 
 	testCompile 'junit:junit'
 	testCompile 'org.assertj:assertj-core'

samples/javaconfig/webflux/src/integration-test/java/sample/AttributeTests.java

@@ -41,7 +41,7 @@ import static org.assertj.core.api.Assertions.assertThat;
  */
 @RunWith(SpringRunner.class)
 @ContextConfiguration(classes = HelloWebfluxApplication.class)
-@TestPropertySource(properties = "server.port=0")
+@TestPropertySource(properties = { "spring.profiles.active=embedded-redis", "server.port=0" })
 public class AttributeTests {
 	@Value("#{@nettyContext.address().getPort()}")
 	int port;

samples/javaconfig/webflux/src/main/java/sample/EmbeddedRedisConfig.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2014-2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package sample;
+
+import java.io.IOException;
+
+import org.testcontainers.containers.GenericContainer;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+import org.springframework.context.annotation.Profile;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+
+@Configuration
+@Profile("embedded-redis")
+public class EmbeddedRedisConfig {
+
+	private static final String REDIS_DOCKER_IMAGE = "redis:3.2.9";
+
+	@Bean(initMethod = "start")
+	public GenericContainer redisContainer() {
+		return new GenericContainer(REDIS_DOCKER_IMAGE) {
+
+			@Override
+			public void close() {
+				super.close();
+				try {
+					this.dockerClient.close();
+				}
+				catch (IOException ignored) {
+				}
+			}
+
+		}.withExposedPorts(6379);
+	}
+
+	@Bean
+	@Primary
+	public LettuceConnectionFactory redisConnectionFactory() {
+		return new LettuceConnectionFactory(redisContainer().getContainerIpAddress(),
+				redisContainer().getFirstMappedPort());
+	}
+
+}

samples/javaconfig/webflux/src/main/java/sample/HelloWebfluxApplication.java

@@ -25,7 +25,6 @@ import org.springframework.context.annotation.AnnotationConfigApplicationContext
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Profile;
 import org.springframework.http.server.reactive.HttpHandler;
 import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
 import org.springframework.web.reactive.config.EnableWebFlux;
@@ -49,7 +48,6 @@ public class HelloWebfluxApplication {
 		}
 	}
 
-	@Profile("default")
 	@Bean
 	public NettyContext nettyContext(ApplicationContext context) {
 		HttpHandler handler = WebHttpHandlerBuilder.applicationContext(context).build();

samples/javaconfig/webflux/src/main/java/sample/HelloWebfluxSessionConfig.java

@@ -16,22 +16,19 @@
 
 package sample;
 
-import java.util.concurrent.ConcurrentHashMap;
-
 import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.session.EnableSpringWebSession;
-import org.springframework.session.MapReactorSessionRepository;
-
+import org.springframework.context.annotation.Import;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.session.data.redis.config.annotation.web.reactor.EnableRedisReactorSession;
 
+@Import(EmbeddedRedisConfig.class)
 // tag::class[]
-@Configuration
-@EnableSpringWebSession
+@EnableRedisReactorSession
 public class HelloWebfluxSessionConfig {
 
 	@Bean
-	public MapReactorSessionRepository reactorSessionRepository() {
-		return new MapReactorSessionRepository(new ConcurrentHashMap<>());
+	public LettuceConnectionFactory lettuceConnectionFactory() {
+		return new LettuceConnectionFactory();
 	}
 }
 // end::class[]

spring-session-data-redis/spring-session-data-redis.gradle

@@ -9,6 +9,10 @@ dependencies {
 		exclude group: "org.slf4j", module: 'jcl-over-slf4j'
 	}
 
+	optional "io.projectreactor:reactor-core"
+	optional "org.springframework:spring-web"
+
+	testCompile "io.projectreactor:reactor-test"
 	testCompile "javax.servlet:javax.servlet-api"
 	testCompile "org.springframework:spring-web"
 	testCompile "org.springframework.security:spring-security-core"

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

@@ -0,0 +1,185 @@
+/*
+ * Copyright 2014-2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.session.data.redis;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.session.Session;
+import org.springframework.session.data.redis.config.annotation.web.reactor.EnableRedisReactorSession;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.web.WebAppConfiguration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Integration tests for {@link ReactiveRedisOperationsSessionRepository}.
+ *
+ * @author Vedran Pavic
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration
+@WebAppConfiguration
+public class ReactiveRedisOperationsSessionRepositoryITests extends AbstractRedisITests {
+
+	@Autowired
+	private ReactiveRedisOperationsSessionRepository repository;
+
+	@Test
+	public void saves() throws InterruptedException {
+		ReactiveRedisOperationsSessionRepository.RedisSession toSave = this.repository
+				.createSession().block();
+
+		String expectedAttributeName = "a";
+		String expectedAttributeValue = "b";
+
+		toSave.setAttribute(expectedAttributeName, expectedAttributeValue);
+		this.repository.save(toSave).block();
+
+		Session session = this.repository.findById(toSave.getId()).block();
+
+		assertThat(session.getId()).isEqualTo(toSave.getId());
+		assertThat(session.getAttributeNames()).isEqualTo(toSave.getAttributeNames());
+		assertThat(session.<String>getAttribute(expectedAttributeName))
+				.isEqualTo(toSave.getAttribute(expectedAttributeName));
+
+		this.repository.deleteById(toSave.getId()).block();
+
+		assertThat(this.repository.findById(toSave.getId()).block()).isNull();
+	}
+
+	@Test
+	public void putAllOnSingleAttrDoesNotRemoveOld() {
+		ReactiveRedisOperationsSessionRepository.RedisSession toSave = this.repository
+				.createSession().block();
+		toSave.setAttribute("a", "b");
+
+		this.repository.save(toSave).block();
+		toSave = this.repository.findById(toSave.getId()).block();
+
+		toSave.setAttribute("1", "2");
+
+		this.repository.save(toSave).block();
+		toSave = this.repository.findById(toSave.getId()).block();
+
+		Session session = this.repository.findById(toSave.getId()).block();
+		assertThat(session.getAttributeNames().size()).isEqualTo(2);
+		assertThat(session.<String>getAttribute("a")).isEqualTo("b");
+		assertThat(session.<String>getAttribute("1")).isEqualTo("2");
+
+		this.repository.deleteById(toSave.getId()).block();
+	}
+
+	@Test
+	public void changeSessionIdWhenOnlyChangeId() throws Exception {
+		String attrName = "changeSessionId";
+		String attrValue = "changeSessionId-value";
+		ReactiveRedisOperationsSessionRepository.RedisSession toSave = this.repository
+				.createSession().block();
+		toSave.setAttribute(attrName, attrValue);
+
+		this.repository.save(toSave).block();
+
+		ReactiveRedisOperationsSessionRepository.RedisSession findById = this.repository
+				.findById(toSave.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();
+
+		ReactiveRedisOperationsSessionRepository.RedisSession findByChangeSessionId = this.repository
+				.findById(changeSessionId).block();
+
+		assertThat(findByChangeSessionId.<String>getAttribute(attrName))
+				.isEqualTo(attrValue);
+	}
+
+	@Test
+	public void changeSessionIdWhenChangeTwice() throws Exception {
+		ReactiveRedisOperationsSessionRepository.RedisSession toSave = this.repository
+				.createSession().block();
+
+		this.repository.save(toSave).block();
+
+		String originalId = toSave.getId();
+		String changeId1 = toSave.changeSessionId();
+		String changeId2 = toSave.changeSessionId();
+
+		this.repository.save(toSave).block();
+
+		assertThat(this.repository.findById(originalId).block()).isNull();
+		assertThat(this.repository.findById(changeId1).block()).isNull();
+		assertThat(this.repository.findById(changeId2).block()).isNotNull();
+	}
+
+	@Test
+	public void changeSessionIdWhenSetAttributeOnChangedSession() throws Exception {
+		String attrName = "changeSessionId";
+		String attrValue = "changeSessionId-value";
+
+		ReactiveRedisOperationsSessionRepository.RedisSession toSave = this.repository
+				.createSession().block();
+
+		this.repository.save(toSave).block();
+
+		ReactiveRedisOperationsSessionRepository.RedisSession findById = this.repository
+				.findById(toSave.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();
+
+		ReactiveRedisOperationsSessionRepository.RedisSession findByChangeSessionId = this.repository
+				.findById(changeSessionId).block();
+
+		assertThat(findByChangeSessionId.<String>getAttribute(attrName))
+				.isEqualTo(attrValue);
+	}
+
+	@Test
+	public void changeSessionIdWhenHasNotSaved() throws Exception {
+		ReactiveRedisOperationsSessionRepository.RedisSession toSave = this.repository
+				.createSession().block();
+		String originalId = toSave.getId();
+		toSave.changeSessionId();
+
+		this.repository.save(toSave).block();
+
+		assertThat(this.repository.findById(toSave.getId()).block()).isNotNull();
+		assertThat(this.repository.findById(originalId).block()).isNull();
+	}
+
+	@Configuration
+	@EnableRedisReactorSession
+	static class Config extends BaseConfig {
+
+	}
+
+}

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

@@ -0,0 +1,358 @@
+/*
+ * Copyright 2014-2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.session.data.redis;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.data.redis.core.ReactiveRedisOperations;
+import org.springframework.session.MapSession;
+import org.springframework.session.ReactorSessionRepository;
+import org.springframework.session.Session;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link ReactorSessionRepository} that is implemented using Spring Data's
+ * {@link ReactiveRedisOperations}.
+ *
+ * @author Vedran Pavic
+ * @since 2.0
+ */
+public class ReactiveRedisOperationsSessionRepository implements
+		ReactorSessionRepository<ReactiveRedisOperationsSessionRepository.RedisSession> {
+
+	/**
+	 * The default prefix for each key and channel in Redis used by Spring Session.
+	 */
+	static final String DEFAULT_SPRING_SESSION_REDIS_PREFIX = "spring:session:";
+
+	/**
+	 * The key in the Hash representing {@link Session#getCreationTime()}.
+	 */
+	static final String CREATION_TIME_KEY = "creationTime";
+
+	/**
+	 * The key in the Hash representing {@link Session#getLastAccessedTime()}.
+	 */
+	static final String LAST_ACCESSED_TIME_KEY = "lastAccessedTime";
+
+	/**
+	 * The key in the Hash representing {@link Session#getMaxInactiveInterval()} .
+	 */
+	static final String MAX_INACTIVE_INTERVAL_KEY = "maxInactiveInterval";
+
+	/**
+	 * The prefix of the key for used for session attributes. The suffix is the name of
+	 * the session attribute. For example, if the session contained an attribute named
+	 * attributeName, then there would be an entry in the hash named
+	 * sessionAttr:attributeName that mapped to its value.
+	 */
+	static final String ATTRIBUTE_PREFIX = "attribute:";
+
+	private final ReactiveRedisOperations<String, Object> sessionRedisOperations;
+
+	/**
+	 * The prefix for every key used by Spring Session in Redis.
+	 */
+	private String keyPrefix = DEFAULT_SPRING_SESSION_REDIS_PREFIX;
+
+	/**
+	 * If non-null, this value is used to override the default value for
+	 * {@link RedisSession#setMaxInactiveInterval(Duration)}.
+	 */
+	private Integer defaultMaxInactiveInterval;
+
+	private RedisFlushMode redisFlushMode = RedisFlushMode.ON_SAVE;
+
+	public ReactiveRedisOperationsSessionRepository(
+			ReactiveRedisOperations<String, Object> sessionRedisOperations) {
+		Assert.notNull(sessionRedisOperations, "sessionRedisOperations cannot be null");
+		this.sessionRedisOperations = sessionRedisOperations;
+	}
+
+	public void setRedisKeyNamespace(String namespace) {
+		Assert.hasText(namespace, "namespace cannot be null or empty");
+		this.keyPrefix = DEFAULT_SPRING_SESSION_REDIS_PREFIX + namespace.trim() + ":";
+	}
+
+	/**
+	 * Sets the maximum inactive interval in seconds between requests before newly created
+	 * sessions will be invalidated. A negative time indicates that the session will never
+	 * timeout. The default is 1800 (30 minutes).
+	 *
+	 * @param defaultMaxInactiveInterval the number of seconds that the {@link Session}
+	 * should be kept alive between client requests.
+	 */
+	public void setDefaultMaxInactiveInterval(int defaultMaxInactiveInterval) {
+		this.defaultMaxInactiveInterval = defaultMaxInactiveInterval;
+	}
+
+	/**
+	 * Sets the redis flush mode. Default flush mode is {@link RedisFlushMode#ON_SAVE}.
+	 *
+	 * @param redisFlushMode the new redis flush mode
+	 */
+	public void setRedisFlushMode(RedisFlushMode redisFlushMode) {
+		Assert.notNull(redisFlushMode, "redisFlushMode cannot be null");
+		this.redisFlushMode = redisFlushMode;
+	}
+
+	@Override
+	public Mono<RedisSession> createSession() {
+		return Mono.defer(() -> {
+			RedisSession session = new RedisSession();
+
+			if (this.defaultMaxInactiveInterval != null) {
+				session.setMaxInactiveInterval(
+						Duration.ofSeconds(this.defaultMaxInactiveInterval));
+			}
+
+			return Mono.just(session);
+		});
+	}
+
+	@Override
+	public Mono<Void> save(RedisSession session) {
+		return session.saveDelta().and(s -> {
+			if (session.isNew) {
+				session.setNew(false);
+			}
+
+			s.onComplete();
+		});
+	}
+
+	@Override
+	public Mono<RedisSession> findById(String id) {
+		String sessionKey = getSessionKey(id);
+
+		return this.sessionRedisOperations.opsForHash().entries(sessionKey)
+				.collect(
+						Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue))
+				.filter(map -> !map.isEmpty()).map(new SessionMapper(id))
+				.filter(session -> !session.isExpired()).map(RedisSession::new)
+				.switchIfEmpty(Mono.defer(() -> deleteById(id).then(Mono.empty())));
+	}
+
+	@Override
+	public Mono<Void> deleteById(String id) {
+		String sessionKey = getSessionKey(id);
+
+		return this.sessionRedisOperations.delete(sessionKey).then();
+	}
+
+	private static String getAttributeKey(String attributeName) {
+		return ATTRIBUTE_PREFIX + attributeName;
+	}
+
+	private String getSessionKey(String sessionId) {
+		return this.keyPrefix + "sessions:" + sessionId;
+	}
+
+	/**
+	 * A custom implementation of {@link Session} that uses a {@link MapSession} as the
+	 * basis for its mapping. It keeps track of any attributes that have changed. When
+	 * {@link RedisSession#saveDelta()} is invoked all the attributes that have been
+	 * changed will be persisted.
+	 */
+	final class RedisSession implements Session {
+
+		private final MapSession cached;
+
+		private final Map<String, Object> delta = new HashMap<>();
+
+		private boolean isNew;
+
+		private String originalSessionId;
+
+		/**
+		 * Creates a new instance ensuring to mark all of the new attributes to be
+		 * persisted in the next save operation.
+		 */
+		RedisSession() {
+			this(new MapSession());
+			this.delta.put(CREATION_TIME_KEY, getCreationTime().toEpochMilli());
+			this.delta.put(MAX_INACTIVE_INTERVAL_KEY,
+					(int) getMaxInactiveInterval().getSeconds());
+			this.delta.put(LAST_ACCESSED_TIME_KEY, getLastAccessedTime().toEpochMilli());
+			this.isNew = true;
+			this.flushImmediateIfNecessary();
+		}
+
+		/**
+		 * Creates a new instance from the provided {@link MapSession}.
+		 *
+		 * @param mapSession the {@link MapSession} that represents the persisted session
+		 * that was retrieved. Cannot be null.
+		 */
+		RedisSession(MapSession mapSession) {
+			Assert.notNull(mapSession, "mapSession cannot be null");
+			this.cached = mapSession;
+			this.originalSessionId = mapSession.getId();
+		}
+
+		public String getId() {
+			return this.cached.getId();
+		}
+
+		public String changeSessionId() {
+			return this.cached.changeSessionId();
+		}
+
+		public <T> T getAttribute(String attributeName) {
+			return this.cached.getAttribute(attributeName);
+		}
+
+		public Set<String> getAttributeNames() {
+			return this.cached.getAttributeNames();
+		}
+
+		public void setAttribute(String attributeName, Object attributeValue) {
+			this.cached.setAttribute(attributeName, attributeValue);
+			putAndFlush(getAttributeKey(attributeName), attributeValue);
+		}
+
+		public void removeAttribute(String attributeName) {
+			this.cached.removeAttribute(attributeName);
+			putAndFlush(getAttributeKey(attributeName), null);
+		}
+
+		public Instant getCreationTime() {
+			return this.cached.getCreationTime();
+		}
+
+		public void setLastAccessedTime(Instant lastAccessedTime) {
+			this.cached.setLastAccessedTime(lastAccessedTime);
+			putAndFlush(LAST_ACCESSED_TIME_KEY, getLastAccessedTime().toEpochMilli());
+		}
+
+		public Instant getLastAccessedTime() {
+			return this.cached.getLastAccessedTime();
+		}
+
+		public void setMaxInactiveInterval(Duration interval) {
+			this.cached.setMaxInactiveInterval(interval);
+			putAndFlush(MAX_INACTIVE_INTERVAL_KEY,
+					(int) getMaxInactiveInterval().getSeconds());
+		}
+
+		public Duration getMaxInactiveInterval() {
+			return this.cached.getMaxInactiveInterval();
+		}
+
+		public boolean isExpired() {
+			return this.cached.isExpired();
+		}
+
+		public void setNew(boolean isNew) {
+			this.isNew = isNew;
+		}
+
+		public boolean isNew() {
+			return this.isNew;
+		}
+
+		private void flushImmediateIfNecessary() {
+			if (ReactiveRedisOperationsSessionRepository.this.redisFlushMode == RedisFlushMode.IMMEDIATE) {
+				saveDelta();
+			}
+		}
+
+		private void putAndFlush(String a, Object v) {
+			this.delta.put(a, v);
+			flushImmediateIfNecessary();
+		}
+
+		private Mono<Void> saveDelta() {
+			String sessionId = getId();
+			Mono<Void> changeSessionId = saveChangeSessionId(sessionId);
+
+			if (this.delta.isEmpty()) {
+				return changeSessionId.and(Mono.empty());
+			}
+
+			String sessionKey = getSessionKey(sessionId);
+
+			Mono<Boolean> update = ReactiveRedisOperationsSessionRepository.this.sessionRedisOperations
+					.opsForHash().putAll(sessionKey, this.delta);
+
+			Mono<Boolean> setTtl = ReactiveRedisOperationsSessionRepository.this.sessionRedisOperations
+					.expire(sessionKey, getMaxInactiveInterval());
+
+			return changeSessionId.and(update).and(setTtl).and(s -> {
+				this.delta.clear();
+				s.onComplete();
+			}).then();
+		}
+
+		private Mono<Void> saveChangeSessionId(String sessionId) {
+			if (isNew() || sessionId.equals(this.originalSessionId)) {
+				return Mono.empty();
+			}
+
+			String originalSessionKey = getSessionKey(this.originalSessionId);
+			String sessionKey = getSessionKey(sessionId);
+
+			return ReactiveRedisOperationsSessionRepository.this.sessionRedisOperations
+					.rename(originalSessionKey, sessionKey).and(s -> {
+						this.originalSessionId = sessionId;
+						s.onComplete();
+					});
+		}
+
+	}
+
+	private static final class SessionMapper
+			implements Function<Map<String, Object>, MapSession> {
+
+		private final String id;
+
+		private SessionMapper(String id) {
+			this.id = id;
+		}
+
+		@Override
+		public MapSession apply(Map<String, Object> map) {
+			MapSession session = new MapSession(this.id);
+
+			session.setCreationTime(
+					Instant.ofEpochMilli((long) map.get(CREATION_TIME_KEY)));
+			session.setLastAccessedTime(
+					Instant.ofEpochMilli((long) map.get(LAST_ACCESSED_TIME_KEY)));
+			session.setMaxInactiveInterval(
+					Duration.ofSeconds((int) map.get(MAX_INACTIVE_INTERVAL_KEY)));
+
+			map.forEach((name, value) -> {
+				if (name.startsWith(ATTRIBUTE_PREFIX)) {
+					session.setAttribute(name.substring(ATTRIBUTE_PREFIX.length()),
+							value);
+				}
+			});
+
+			return session;
+		}
+
+	}
+
+}

spring-session-data-redis/src/main/java/org/springframework/session/data/redis/config/annotation/web/reactor/EnableRedisReactorSession.java

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2014-2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.session.data.redis.config.annotation.web.reactor;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
+import org.springframework.session.EnableSpringWebSession;
+import org.springframework.session.ReactorSessionRepository;
+import org.springframework.session.Session;
+import org.springframework.session.data.redis.RedisFlushMode;
+
+/**
+ * Add this annotation to an {@code @Configuration} class to expose the
+ * {@link org.springframework.web.server.session.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
+ * &#064;EnableRedisReactorSession
+ * public class RedisReactorSessionConfig {
+ *
+ *     &#064;Bean
+ *     public LettuceConnectionFactory redisConnectionFactory() {
+ *         return new LettuceConnectionFactory();
+ *     }
+ *
+ * }
+ * </pre>
+ *
+ * More advanced configurations can extend {@link RedisReactorSessionConfiguration}
+ * instead.
+ *
+ * @author Vedran Pavic
+ * @since 2.0.0
+ * @see EnableSpringWebSession
+ */
+@Retention(java.lang.annotation.RetentionPolicy.RUNTIME)
+@Target({ java.lang.annotation.ElementType.TYPE })
+@Documented
+@Import(RedisReactorSessionConfiguration.class)
+@Configuration
+public @interface EnableRedisReactorSession {
+
+	int maxInactiveIntervalInSeconds() default 1800;
+
+	/**
+	 * <p>
+	 * Defines a unique namespace for keys. The value is used to isolate sessions by
+	 * changing the prefix from {@code spring:session:} to
+	 * {@code spring:session:<redisNamespace>:}. The default is "" such that all Redis
+	 * keys begin with {@code spring:session:}.
+	 * </p>
+	 *
+	 * <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.
+	 * </p>
+	 *
+	 * @return the unique namespace for keys
+	 */
+	String redisNamespace() default "";
+
+	/**
+	 * <p>
+	 * Sets the flush mode for the Redis sessions. The default is ON_SAVE which only
+	 * updates the backing Redis when {@link ReactorSessionRepository#save(Session)} is
+	 * invoked. In a web environment this happens just before the HTTP response is
+	 * committed.
+	 * </p>
+	 *
+	 * <p>
+	 * Setting the value to IMMEDIATE will ensure that the any updates to the Session are
+	 * immediately written to the Redis instance.
+	 * </p>
+	 *
+	 * @return the {@link RedisFlushMode} to use
+	 */
+	RedisFlushMode redisFlushMode() default RedisFlushMode.ON_SAVE;
+
+}

spring-session-data-redis/src/main/java/org/springframework/session/data/redis/config/annotation/web/reactor/RedisReactorSessionConfiguration.java

@@ -0,0 +1,136 @@
+/*
+ * Copyright 2014-2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.session.data.redis.config.annotation.web.reactor;
+
+import java.util.Map;
+
+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.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.data.redis.serializer.StringRedisSerializer;
+import org.springframework.session.SpringWebSessionConfiguration;
+import org.springframework.session.data.redis.ReactiveRedisOperationsSessionRepository;
+import org.springframework.session.data.redis.RedisFlushMode;
+import org.springframework.util.Assert;
+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}. In
+ * order to use this a single {@link ReactiveRedisConnectionFactory} must be exposed as a
+ * Bean.
+ *
+ * @author Vedran Pavic
+ * @see EnableRedisReactorSession
+ * @since 2.0.0
+ */
+@Configuration
+public class RedisReactorSessionConfiguration extends SpringWebSessionConfiguration
+		implements EmbeddedValueResolverAware, ImportAware {
+
+	private static final RedisSerializer<String> keySerializer = new StringRedisSerializer();
+
+	private static final RedisSerializer<Object> valueSerializer = new JdkSerializationRedisSerializer();
+
+	private Integer maxInactiveIntervalInSeconds = 1800;
+
+	private String redisNamespace = "";
+
+	private RedisFlushMode redisFlushMode = RedisFlushMode.ON_SAVE;
+
+	private StringValueResolver embeddedValueResolver;
+
+	@Bean
+	public ReactiveRedisOperationsSessionRepository sessionRepository(
+			ReactiveRedisConnectionFactory redisConnectionFactory) {
+		ReactiveRedisOperationsSessionRepository sessionRepository = new ReactiveRedisOperationsSessionRepository(
+				createDefaultTemplate(redisConnectionFactory));
+		sessionRepository
+				.setDefaultMaxInactiveInterval(this.maxInactiveIntervalInSeconds);
+
+		String redisNamespace = getRedisNamespace();
+
+		if (StringUtils.hasText(redisNamespace)) {
+			sessionRepository.setRedisKeyNamespace(redisNamespace);
+		}
+
+		sessionRepository.setRedisFlushMode(this.redisFlushMode);
+
+		return sessionRepository;
+	}
+
+	public void setMaxInactiveIntervalInSeconds(int maxInactiveIntervalInSeconds) {
+		this.maxInactiveIntervalInSeconds = maxInactiveIntervalInSeconds;
+	}
+
+	public void setRedisNamespace(String namespace) {
+		this.redisNamespace = namespace;
+	}
+
+	public void setRedisFlushMode(RedisFlushMode redisFlushMode) {
+		Assert.notNull(redisFlushMode, "redisFlushMode cannot be null");
+		this.redisFlushMode = redisFlushMode;
+	}
+
+	public void setEmbeddedValueResolver(StringValueResolver resolver) {
+		this.embeddedValueResolver = resolver;
+	}
+
+	public void setImportMetadata(AnnotationMetadata importMetadata) {
+		Map<String, Object> enableAttrMap = importMetadata
+				.getAnnotationAttributes(EnableRedisReactorSession.class.getName());
+		AnnotationAttributes enableAttrs = AnnotationAttributes.fromMap(enableAttrMap);
+
+		if (enableAttrs != null) {
+			this.maxInactiveIntervalInSeconds = enableAttrs
+					.getNumber("maxInactiveIntervalInSeconds");
+			String redisNamespaceValue = enableAttrs.getString("redisNamespace");
+			if (StringUtils.hasText(redisNamespaceValue)) {
+				this.redisNamespace = this.embeddedValueResolver
+						.resolveStringValue(redisNamespaceValue);
+			}
+			this.redisFlushMode = enableAttrs.getEnum("redisFlushMode");
+		}
+	}
+
+	private static ReactiveRedisTemplate<String, Object> createDefaultTemplate(
+			ReactiveRedisConnectionFactory connectionFactory) {
+		RedisSerializationContext<String, Object> serializationContext = RedisSerializationContext
+				.<String, Object>newSerializationContext(valueSerializer)
+				.key(keySerializer).hashKey(keySerializer).build();
+
+		return new ReactiveRedisTemplate<>(connectionFactory, serializationContext);
+	}
+
+	private String getRedisNamespace() {
+		if (StringUtils.hasText(this.redisNamespace)) {
+			return this.redisNamespace;
+		}
+
+		return System.getProperty("spring.session.redis.namespace", "");
+	}
+
+}

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

@@ -0,0 +1,375 @@
+/*
+ * Copyright 2014-2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.session.data.redis;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.mockito.ArgumentCaptor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import org.springframework.data.redis.core.ReactiveHashOperations;
+import org.springframework.data.redis.core.ReactiveRedisOperations;
+import org.springframework.session.MapSession;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyZeroInteractions;
+
+/**
+ * Tests for {@link ReactiveRedisOperationsSessionRepository}.
+ *
+ * @author Vedran Pavic
+ */
+public class ReactiveRedisOperationsSessionRepositoryTests {
+
+	@Rule
+	public ExpectedException thrown = ExpectedException.none();
+
+	@SuppressWarnings("unchecked")
+	private ReactiveRedisOperations<String, Object> redisOperations = mock(
+			ReactiveRedisOperations.class);
+
+	@SuppressWarnings("unchecked")
+	private ReactiveHashOperations<String, Object, Object> hashOperations = mock(
+			ReactiveHashOperations.class);
+
+	@SuppressWarnings("unchecked")
+	private ArgumentCaptor<Map<String, Object>> delta = ArgumentCaptor
+			.forClass(Map.class);
+
+	private ReactiveRedisOperationsSessionRepository repository;
+
+	@Before
+	public void setUp() throws Exception {
+		this.repository = new ReactiveRedisOperationsSessionRepository(
+				this.redisOperations);
+	}
+
+	@Test
+	public void constructorWithNullReactiveRedisOperations() {
+		this.thrown.expect(IllegalArgumentException.class);
+		this.thrown.expectMessage("sessionRedisOperations cannot be null");
+
+		new ReactiveRedisOperationsSessionRepository(null);
+	}
+
+	@Test
+	public void customRedisKeyNamespace() {
+		this.repository.setRedisKeyNamespace("test");
+
+		assertThat(ReflectionTestUtils.getField(this.repository, "keyPrefix")).isEqualTo(
+				ReactiveRedisOperationsSessionRepository.DEFAULT_SPRING_SESSION_REDIS_PREFIX
+						+ "test:");
+	}
+
+	@Test
+	public void nullRedisKeyNamespace() {
+		this.thrown.expect(IllegalArgumentException.class);
+		this.thrown.expectMessage("namespace cannot be null or empty");
+
+		this.repository.setRedisKeyNamespace(null);
+	}
+
+	@Test
+	public void emptyRedisKeyNamespace() {
+		this.thrown.expect(IllegalArgumentException.class);
+		this.thrown.expectMessage("namespace cannot be null or empty");
+
+		this.repository.setRedisKeyNamespace("");
+	}
+
+	@Test
+	public void customMaxInactiveInterval() {
+		this.repository.setDefaultMaxInactiveInterval(600);
+
+		assertThat(ReflectionTestUtils.getField(this.repository,
+				"defaultMaxInactiveInterval")).isEqualTo(600);
+	}
+
+	@Test
+	public void customRedisFlushMode() {
+		this.repository.setRedisFlushMode(RedisFlushMode.IMMEDIATE);
+
+		assertThat(ReflectionTestUtils.getField(this.repository, "redisFlushMode"))
+				.isEqualTo(RedisFlushMode.IMMEDIATE);
+	}
+
+	@Test
+	public void nullRedisFlushMode() {
+		this.thrown.expect(IllegalArgumentException.class);
+		this.thrown.expectMessage("redisFlushMode cannot be null");
+
+		this.repository.setRedisFlushMode(null);
+	}
+
+	@Test
+	public void createSessionDefaultMaxInactiveInterval() {
+		Mono<ReactiveRedisOperationsSessionRepository.RedisSession> session = this.repository
+				.createSession();
+
+		StepVerifier.create(session).expectNextMatches(predicate -> {
+			assertThat(predicate.getMaxInactiveInterval()).isEqualTo(
+					Duration.ofSeconds(MapSession.DEFAULT_MAX_INACTIVE_INTERVAL_SECONDS));
+			return true;
+		});
+	}
+
+	@Test
+	public void createSessionCustomMaxInactiveInterval() {
+		this.repository.setDefaultMaxInactiveInterval(600);
+		Mono<ReactiveRedisOperationsSessionRepository.RedisSession> session = this.repository
+				.createSession();
+
+		StepVerifier.create(session).expectNextMatches(predicate -> {
+			assertThat(predicate.getMaxInactiveInterval())
+					.isEqualTo(Duration.ofSeconds(600));
+			return true;
+		});
+	}
+
+	@Test
+	public void saveNewSession() {
+		given(this.redisOperations.opsForHash()).willReturn(this.hashOperations);
+		given(this.hashOperations.putAll(anyString(), this.delta.capture()))
+				.willReturn(Mono.just(true));
+		given(this.redisOperations.expire(anyString(), any()))
+				.willReturn(Mono.just(true));
+
+		ReactiveRedisOperationsSessionRepository.RedisSession session = this.repository.new RedisSession();
+		Mono<Void> result = this.repository.save(session);
+
+		StepVerifier.create(result).expectNextMatches(predicate -> {
+			Map<String, Object> delta = this.delta.getAllValues().get(0);
+			assertThat(delta.size()).isEqualTo(3);
+			Object creationTime = delta
+					.get(ReactiveRedisOperationsSessionRepository.CREATION_TIME_KEY);
+			assertThat(creationTime).isEqualTo(session.getCreationTime().toEpochMilli());
+			assertThat(delta.get(
+					ReactiveRedisOperationsSessionRepository.MAX_INACTIVE_INTERVAL_KEY))
+							.isEqualTo((int) Duration.ofSeconds(
+									MapSession.DEFAULT_MAX_INACTIVE_INTERVAL_SECONDS)
+									.getSeconds());
+			assertThat(delta
+					.get(ReactiveRedisOperationsSessionRepository.LAST_ACCESSED_TIME_KEY))
+							.isEqualTo(session.getCreationTime().toEpochMilli());
+			return true;
+		});
+	}
+
+	@Test
+	public void saveSessionNothingChanged() {
+		ReactiveRedisOperationsSessionRepository.RedisSession session = this.repository.new RedisSession(
+				new MapSession());
+
+		Mono<Void> result = this.repository.save(session);
+
+		StepVerifier.create(result).expectNextMatches(predicate -> {
+			verifyZeroInteractions(this.redisOperations);
+			return true;
+		});
+	}
+
+	@Test
+	public void saveLastAccessChanged() {
+		given(this.redisOperations.opsForHash()).willReturn(this.hashOperations);
+		given(this.hashOperations.putAll(anyString(), this.delta.capture()))
+				.willReturn(Mono.just(true));
+		given(this.redisOperations.expire(anyString(), any()))
+				.willReturn(Mono.just(true));
+
+		ReactiveRedisOperationsSessionRepository.RedisSession session = this.repository.new RedisSession(
+				new MapSession());
+		session.setLastAccessedTime(Instant.ofEpochMilli(12345678L));
+		Mono<Void> result = this.repository.save(session);
+
+		StepVerifier.create(result).expectNextMatches(predicate -> {
+			assertThat(this.delta.getAllValues().get(0))
+					.isEqualTo(map(RedisOperationsSessionRepository.LAST_ACCESSED_ATTR,
+							session.getLastAccessedTime().toEpochMilli()));
+			return true;
+		});
+	}
+
+	@Test
+	public void saveSetAttribute() {
+		given(this.redisOperations.opsForHash()).willReturn(this.hashOperations);
+		given(this.hashOperations.putAll(anyString(), this.delta.capture()))
+				.willReturn(Mono.just(true));
+		given(this.redisOperations.expire(anyString(), any()))
+				.willReturn(Mono.just(true));
+
+		String attrName = "attrName";
+		ReactiveRedisOperationsSessionRepository.RedisSession session = this.repository.new RedisSession(
+				new MapSession());
+		session.setAttribute(attrName, "attrValue");
+		Mono<Void> result = this.repository.save(session);
+
+		StepVerifier.create(result).expectNextMatches(predicate -> {
+			assertThat(this.delta.getAllValues().get(0)).isEqualTo(
+					map(RedisOperationsSessionRepository.getSessionAttrNameKey(attrName),
+							session.getAttribute(attrName)));
+			return true;
+		});
+	}
+
+	@Test
+	public void saveRemoveAttribute() {
+		given(this.redisOperations.opsForHash()).willReturn(this.hashOperations);
+		given(this.hashOperations.putAll(anyString(), this.delta.capture()))
+				.willReturn(Mono.just(true));
+		given(this.redisOperations.expire(anyString(), any()))
+				.willReturn(Mono.just(true));
+
+		String attrName = "attrName";
+		ReactiveRedisOperationsSessionRepository.RedisSession session = this.repository.new RedisSession(
+				new MapSession());
+		session.removeAttribute(attrName);
+		Mono<Void> result = this.repository.save(session);
+
+		StepVerifier.create(result).expectNextMatches(predicate -> {
+			assertThat(this.delta.getAllValues().get(0)).isEqualTo(
+					map(RedisOperationsSessionRepository.getSessionAttrNameKey(attrName),
+							null));
+			return true;
+		});
+	}
+
+	@Test
+	public void redisSessionGetAttributes() {
+		String attrName = "attrName";
+		ReactiveRedisOperationsSessionRepository.RedisSession session = this.repository.new RedisSession();
+		assertThat(session.getAttributeNames()).isEmpty();
+
+		session.setAttribute(attrName, "attrValue");
+		assertThat(session.getAttributeNames()).containsOnly(attrName);
+
+		session.removeAttribute(attrName);
+		assertThat(session.getAttributeNames()).isEmpty();
+	}
+
+	@Test
+	public void delete() {
+		given(this.redisOperations.delete(anyString())).willReturn(Mono.just(1L));
+
+		ReactiveRedisOperationsSessionRepository.RedisSession session = this.repository.new RedisSession(
+				new MapSession());
+		Mono<Void> result = this.repository.deleteById(session.getId());
+
+		StepVerifier.create(result).expectNextMatches(predicate -> {
+			assertThat(result).isEqualTo(1);
+			return true;
+		});
+	}
+
+	@Test
+	public void getSessionNotFound() {
+		given(this.redisOperations.opsForHash()).willReturn(this.hashOperations);
+		given(this.hashOperations.entries(anyString())).willReturn(Flux.empty());
+
+		Mono<ReactiveRedisOperationsSessionRepository.RedisSession> session = this.repository
+				.findById("test");
+
+		StepVerifier.create(session).expectNextMatches(predicate -> {
+			assertThat(predicate).isEqualTo(Mono.empty());
+			return true;
+		});
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void getSessionFound() {
+		given(this.redisOperations.opsForHash()).willReturn(this.hashOperations);
+		String attrName = "attrName";
+		MapSession expected = new MapSession();
+		expected.setLastAccessedTime(Instant.now().minusSeconds(60));
+		expected.setAttribute(attrName, "attrValue");
+		Map map = map(RedisOperationsSessionRepository.getSessionAttrNameKey(attrName),
+				expected.getAttribute(attrName),
+				RedisOperationsSessionRepository.CREATION_TIME_ATTR,
+				expected.getCreationTime().toEpochMilli(),
+				RedisOperationsSessionRepository.MAX_INACTIVE_ATTR,
+				(int) expected.getMaxInactiveInterval().getSeconds(),
+				RedisOperationsSessionRepository.LAST_ACCESSED_ATTR,
+				expected.getLastAccessedTime().toEpochMilli());
+		given(this.hashOperations.entries(anyString()))
+				.willReturn(Flux.fromIterable(map.entrySet()));
+
+		Mono<ReactiveRedisOperationsSessionRepository.RedisSession> session = this.repository
+				.findById("test");
+
+		StepVerifier.create(session).expectNextMatches(predicate -> {
+			assertThat(predicate.getId()).isEqualTo(expected.getId());
+			assertThat(predicate.getAttributeNames())
+					.isEqualTo(expected.getAttributeNames());
+			assertThat(predicate.<String>getAttribute(attrName))
+					.isEqualTo(expected.getAttribute(attrName));
+			assertThat(predicate.getCreationTime()).isEqualTo(expected.getCreationTime());
+			assertThat(predicate.getMaxInactiveInterval())
+					.isEqualTo(expected.getMaxInactiveInterval());
+			assertThat(predicate.getLastAccessedTime())
+					.isEqualTo(expected.getLastAccessedTime());
+			return true;
+		});
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void getSessionExpired() {
+		given(this.redisOperations.opsForHash()).willReturn(this.hashOperations);
+		Map map = map(RedisOperationsSessionRepository.MAX_INACTIVE_ATTR, 1,
+				RedisOperationsSessionRepository.LAST_ACCESSED_ATTR,
+				Instant.now().minus(5, ChronoUnit.MINUTES).toEpochMilli());
+		given(this.hashOperations.entries(anyString()))
+				.willReturn(Flux.fromIterable(map.entrySet()));
+
+		Mono<ReactiveRedisOperationsSessionRepository.RedisSession> session = this.repository
+				.findById("test");
+
+		StepVerifier.create(session).expectNextMatches(predicate -> {
+			assertThat(predicate).isNull();
+			return true;
+		});
+	}
+
+	// TODO
+
+	private Map<String, Object> map(Object... objects) {
+		Map<String, Object> result = new HashMap<>();
+		if (objects == null) {
+			return result;
+		}
+		for (int i = 0; i < objects.length; i += 2) {
+			result.put((String) objects[i], objects[i + 1]);
+		}
+		return result;
+	}
+
+}

spring-session-data-redis/src/test/java/org/springframework/session/data/redis/config/annotation/web/reactor/RedisReactorSessionConfigurationTests.java

@@ -0,0 +1,141 @@
+/*
+ * Copyright 2014-2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.session.data.redis.config.annotation.web.reactor;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
+import org.springframework.session.data.redis.ReactiveRedisOperationsSessionRepository;
+import org.springframework.session.data.redis.RedisFlushMode;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests for {@link RedisReactorSessionConfiguration}.
+ *
+ * @author Vedran Pavic
+ */
+public class RedisReactorSessionConfigurationTests {
+
+	private static final String REDIS_NAMESPACE = "testNamespace";
+
+	private static final int MAX_INACTIVE_INTERVAL_IN_SECONDS = 600;
+
+	private AnnotationConfigApplicationContext context;
+
+	@Before
+	public void before() {
+		this.context = new AnnotationConfigApplicationContext();
+	}
+
+	@After
+	public void after() {
+		if (this.context != null) {
+			this.context.close();
+		}
+	}
+
+	@Test
+	public void defaultConfiguration() {
+		registerAndRefresh(RedisConfiguration.class, DefaultConfiguration.class);
+
+		ReactiveRedisOperationsSessionRepository repository = this.context
+				.getBean(ReactiveRedisOperationsSessionRepository.class);
+		assertThat(repository).isNotNull();
+	}
+
+	@Test
+	public void customNamespace() {
+		registerAndRefresh(RedisConfiguration.class, CustomNamespaceConfiguration.class);
+
+		ReactiveRedisOperationsSessionRepository repository = this.context
+				.getBean(ReactiveRedisOperationsSessionRepository.class);
+		assertThat(repository).isNotNull();
+		assertThat(ReflectionTestUtils.getField(repository, "keyPrefix"))
+				.isEqualTo("spring:session:" + REDIS_NAMESPACE + ":");
+	}
+
+	@Test
+	public void customMaxInactiveInterval() {
+		registerAndRefresh(RedisConfiguration.class,
+				CustomMaxInactiveIntervalConfiguration.class);
+
+		ReactiveRedisOperationsSessionRepository repository = this.context
+				.getBean(ReactiveRedisOperationsSessionRepository.class);
+		assertThat(repository).isNotNull();
+		assertThat(ReflectionTestUtils.getField(repository, "defaultMaxInactiveInterval"))
+				.isEqualTo(MAX_INACTIVE_INTERVAL_IN_SECONDS);
+	}
+
+	@Test
+	public void customFlushMode() {
+		registerAndRefresh(RedisConfiguration.class, CustomFlushModeConfiguration.class);
+
+		ReactiveRedisOperationsSessionRepository repository = this.context
+				.getBean(ReactiveRedisOperationsSessionRepository.class);
+		assertThat(repository).isNotNull();
+		assertThat(ReflectionTestUtils.getField(repository, "redisFlushMode"))
+				.isEqualTo(RedisFlushMode.IMMEDIATE);
+	}
+
+	private void registerAndRefresh(Class<?>... annotatedClasses) {
+		this.context.register(annotatedClasses);
+		this.context.refresh();
+	}
+
+	@Configuration
+	static class RedisConfiguration {
+
+		@Bean
+		public ReactiveRedisConnectionFactory redisConnectionFactory() {
+			return mock(ReactiveRedisConnectionFactory.class);
+		}
+
+	}
+
+	@Configuration
+	@EnableRedisReactorSession
+	static class DefaultConfiguration {
+
+	}
+
+	@Configuration
+	@EnableRedisReactorSession(redisNamespace = REDIS_NAMESPACE)
+	static class CustomNamespaceConfiguration {
+
+	}
+
+	@Configuration
+	@EnableRedisReactorSession(maxInactiveIntervalInSeconds = MAX_INACTIVE_INTERVAL_IN_SECONDS)
+	static class CustomMaxInactiveIntervalConfiguration {
+
+	}
+
+	@Configuration
+	@EnableRedisReactorSession(redisFlushMode = RedisFlushMode.IMMEDIATE)
+	static class CustomFlushModeConfiguration {
+
+	}
+
+}