Extract connection creation into ReactiveRedisTemplate.getConnection() method.

A new method, called getConnection, is introduced to allow overriding of connection creation.

Closes #2145
Original pull request: #2162.

Author: petromir

src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java

@@ -232,21 +232,29 @@ <T> Publisher<T> doInConnection(ReactiveRedisCallback<T> action, boolean exposeC
 
 		Assert.notNull(action, "Callback object must not be null");
 
-		return Flux.usingWhen(Mono.fromSupplier(() -> {
-
-			ReactiveRedisConnectionFactory factory = getConnectionFactory();
-			ReactiveRedisConnection conn = factory.getReactiveConnection();
-			ReactiveRedisConnection connToUse = preProcessConnection(conn, false);
-
-			return (exposeConnection ? connToUse : createRedisConnectionProxy(connToUse));
-		}), conn -> {
+		return Flux.usingWhen(getConnection(exposeConnection), conn -> {
 			Publisher<T> result = action.doInRedis(conn);
 
 			return postProcessResult(result, conn, false);
 
 		}, ReactiveRedisConnection::closeLater);
 	}
 
+	/**
+	 * Creates a Mono which generates a new connection. The successors of {@link ReactiveRedisTemplate} might override
+	 * the default behaviour.
+	 *
+	 * @param exposeConnection whether to enforce exposure of the native Redis Connection to callback code
+	 * return a {@link Mono} wrapping the {@link ReactiveRedisConnection}.
+	 */
+	protected Mono<ReactiveRedisConnection> getConnection(boolean exposeConnection) {
+		ReactiveRedisConnectionFactory factory = getConnectionFactory();
+		ReactiveRedisConnection conn = factory.getReactiveConnection();
+		ReactiveRedisConnection connToUse = preProcessConnection(conn, false);
+
+		return Mono.just(exposeConnection ? connToUse : createRedisConnectionProxy(connToUse));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#convertAndSend(java.lang.String, java.lang.Object)