adding ReactiveFindByIndexNameSessionRepository

spring-session-core/src/main/java/org/springframework/session/ReactiveFindByIndexNameSessionRepository.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2014-2018 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;
+
+import java.util.Map;
+
+import reactor.core.publisher.Flux;
+import reactor.util.function.Tuple2;
+
+/**
+ * Extends a basic {@link SessionRepository} to allow finding sessions by the specified
+ * index name and index value.
+ *
+ * @param <S> the type of Session being managed by this
+ * {@link ReactiveFindByIndexNameSessionRepository}
+ * @author Artsiom Yudovin
+ */
+public interface ReactiveFindByIndexNameSessionRepository<S extends Session>
+		extends ReactiveSessionRepository<S> {
+
+	/**
+	 * A session index that contains the current principal name (i.e. username).
+	 * <p>
+	 * It is the responsibility of the developer to ensure the index is populated since
+	 * Spring Session is not aware of the authentication mechanism being used.
+	 *
+	 * @since 1.1
+	 */
+	String PRINCIPAL_NAME_INDEX_NAME = ReactiveFindByIndexNameSessionRepository.class.getName()
+			.concat(".PRINCIPAL_NAME_INDEX_NAME");
+
+	/**
+	 * Find a {@link Map} of the session id to the {@link Session} of all sessions that
+	 * contain the specified index name index value.
+	 *
+	 * @param indexName the name of the index (i.e.
+	 * {@link ReactiveFindByIndexNameSessionRepository#PRINCIPAL_NAME_INDEX_NAME})
+	 * @param indexValue the value of the index to search for.
+	 * @return a {@code Map} (never {@code null}) of the session id to the {@code Session}
+	 * of all sessions that contain the specified index name and index value. If no
+	 * results are found, an empty {@code Map} is returned.
+	 */
+	Flux<Tuple2<String, S>> findByIndexNameAndIndexValue(String indexName, String indexValue);
+
+	/**
+	 * Find a {@link Map} of the session id to the {@link Session} of all sessions that
+	 * contain the index with the name
+	 * {@link ReactiveFindByIndexNameSessionRepository#PRINCIPAL_NAME_INDEX_NAME} and the
+	 * specified principal name.
+	 *
+	 * @param principalName the principal name
+	 * @return a {@code Map} (never {@code null}) of the session id to the {@code Session}
+	 * of all sessions that contain the specified principal name. If no results are found,
+	 * an empty {@code Map} is returned.
+	 * @since 2.1.0
+	 */
+	default Flux<Tuple2<String, S>> findByPrincipalName(String principalName) {
+
+		return findByIndexNameAndIndexValue(PRINCIPAL_NAME_INDEX_NAME, principalName);
+
+	}
+
+}

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

@@ -24,10 +24,15 @@
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
 
 import org.springframework.data.redis.core.ReactiveRedisOperations;
+import org.springframework.session.FindByIndexNameSessionRepository;
 import org.springframework.session.MapSession;
+import org.springframework.session.ReactiveFindByIndexNameSessionRepository;
 import org.springframework.session.ReactiveSessionRepository;
 import org.springframework.session.Session;
 import org.springframework.util.Assert;
@@ -37,10 +42,11 @@
  * {@link ReactiveRedisOperations}.
  *
  * @author Vedran Pavic
+ * @author Artsiom Yudovin
  * @since 2.0
  */
 public class ReactiveRedisOperationsSessionRepository implements
-		ReactiveSessionRepository<ReactiveRedisOperationsSessionRepository.RedisSession> {
+		ReactiveFindByIndexNameSessionRepository<ReactiveRedisOperationsSessionRepository.RedisSession> {
 
 	/**
 	 * The default namespace for each key and channel in Redis used by Spring Session.
@@ -192,6 +198,24 @@ private String getSessionKey(String sessionId) {
 		return this.namespace + "sessions:" + sessionId;
 	}
 
+	private String getPrincipalKey(String principalName) {
+		return this.namespace + "index:"
+				+ FindByIndexNameSessionRepository.PRINCIPAL_NAME_INDEX_NAME + ":"
+				+ principalName;
+	}
+
+	@Override
+	public Flux<Tuple2<String, RedisSession>> findByIndexNameAndIndexValue(String indexName, String indexValue) {
+		if (!PRINCIPAL_NAME_INDEX_NAME.equals(indexName)) {
+			return Flux.empty();
+		}
+		String principalKey = getPrincipalKey(indexValue);
+		return this.sessionRedisOperations.opsForSet()
+				.scan(principalKey)
+				.cast(String.class)
+				.flatMap((id) -> this.findById(id).map((session) -> Tuples.of(id, session)));
+	}
+
 	/**
 	 * 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

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

@@ -25,6 +25,8 @@
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
+import org.springframework.data.redis.core.ReactiveSetOperations;
+import org.springframework.session.ReactiveFindByIndexNameSessionRepository;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
@@ -40,14 +42,13 @@
 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.verify;
-import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.*;
 
 /**
  * Tests for {@link ReactiveRedisOperationsSessionRepository}.
  *
  * @author Vedran Pavic
+ * @author Artsiom Yudovin
  */
 public class ReactiveRedisOperationsSessionRepositoryTests {
 
@@ -59,6 +60,10 @@ public class ReactiveRedisOperationsSessionRepositoryTests {
 	private ReactiveHashOperations<String, Object, Object> hashOperations = mock(
 			ReactiveHashOperations.class);
 
+	@SuppressWarnings("unchecked")
+	private ReactiveSetOperations<String, Object> setOperations = mock(
+			ReactiveSetOperations.class);
+
 	@SuppressWarnings("unchecked")
 	private ArgumentCaptor<Map<String, Object>> delta = ArgumentCaptor
 			.forClass(Map.class);
@@ -392,6 +397,51 @@ public void getAttributeNamesAndRemove() {
 		assertThat(session.getAttributeNames()).isEmpty();
 	}
 
+	@Test
+	@SuppressWarnings("unchecked")
+	public void findByIndexNameAndIndexValue() {
+		given(this.redisOperations.opsForHash()).willReturn(this.hashOperations);
+		String attribute1 = "attribute1";
+		String attribute2 = "attribute2";
+		MapSession expected = new MapSession("test");
+		expected.setLastAccessedTime(Instant.now().minusSeconds(60));
+		expected.setAttribute(attribute1, "test");
+		expected.setAttribute(attribute2, null);
+		Map map = map(
+				ReactiveRedisOperationsSessionRepository.ATTRIBUTE_PREFIX + attribute1,
+				expected.getAttribute(attribute1),
+				ReactiveRedisOperationsSessionRepository.ATTRIBUTE_PREFIX + attribute2,
+				expected.getAttribute(attribute2),
+				ReactiveRedisOperationsSessionRepository.CREATION_TIME_KEY,
+				expected.getCreationTime().toEpochMilli(),
+				ReactiveRedisOperationsSessionRepository.MAX_INACTIVE_INTERVAL_KEY,
+				(int) expected.getMaxInactiveInterval().getSeconds(),
+				ReactiveRedisOperationsSessionRepository.LAST_ACCESSED_TIME_KEY,
+				expected.getLastAccessedTime().toEpochMilli());
+		given(this.hashOperations.entries(anyString()))
+				.willReturn(Flux.fromIterable(map.entrySet()));
+
+		Flux<Object> ids = Flux.just("id");
+		given(this.setOperations.scan(anyString())).willReturn(ids);
+		given(this.redisOperations.opsForSet()).willReturn(this.setOperations);
+
+		StepVerifier
+				.create(this.repository
+						.findByIndexNameAndIndexValue(
+								ReactiveFindByIndexNameSessionRepository.PRINCIPAL_NAME_INDEX_NAME,
+								"attribute2"))
+				.consumeNextWith(tuple -> {
+					assertThat(tuple.getT1()).isEqualTo("id");
+					assertThat(tuple.getT2())
+							.isInstanceOf(RedisSession.class);
+				}).verifyComplete();
+
+		verify(this.redisOperations).opsForHash();
+		verify(this.hashOperations).entries(anyString());
+		verify(this.redisOperations).opsForSet();
+		verify(this.setOperations).scan(anyString());
+	}
+
 	private Map<String, Object> map(Object... objects) {
 		Map<String, Object> result = new HashMap<>();
 		if (objects == null) {