Introduce doCreate… methods without connection proxying.

doCreateMono and doCreateFlux now no longer proxy the connection for commands invoked directly from the ReactiveRedisTemplate implementations as proxying isn't necessary for simple command invocation.

Original Pull Request: #2129

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

@@ -15,8 +15,6 @@
  */
 package org.springframework.data.redis.core;
 
-import org.springframework.data.redis.domain.geo.GeoReference;
-import org.springframework.data.redis.domain.geo.GeoReference.GeoMemberReference;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -28,6 +26,7 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.data.geo.Circle;
 import org.springframework.data.geo.Distance;
 import org.springframework.data.geo.GeoResult;
@@ -37,6 +36,8 @@ import org.springframework.data.redis.connection.ReactiveGeoCommands;
 import org.springframework.data.redis.connection.RedisGeoCommands;
 import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
 import org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusCommandArgs;
+import org.springframework.data.redis.domain.geo.GeoReference;
+import org.springframework.data.redis.domain.geo.GeoReference.GeoMemberReference;
 import org.springframework.data.redis.domain.geo.GeoShape;
 import org.springframework.data.redis.serializer.RedisSerializationContext;
 import org.springframework.util.Assert;
@@ -320,7 +321,7 @@ class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations<K, V>
 		Assert.notEmpty(members, "Members must not be null or empty!");
 		Assert.noNullElements(members, "Members must not contain null elements!");
 
-		return template.createMono(connection -> Flux.fromArray(members) //
+		return template.doCreateMono(connection -> Flux.fromArray(members) //
 				.map(this::rawValue) //
 				.collectList() //
 				.flatMap(serialized -> connection.zSetCommands().zRem(rawKey(key), serialized)));
@@ -335,10 +336,10 @@ class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations<K, V>
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
+		return template.doCreateMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveGeoOperations#search(K, RedisGeoCommands.GeoReference, GeoShape, GeoSearchCommandArgs)
 	 */
@@ -350,11 +351,11 @@ class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations<K, V>
 		Assert.notNull(reference, "GeoReference must not be null!");
 		GeoReference<ByteBuffer> rawReference = getGeoReference(reference);
 
-		return template.createFlux(connection -> connection.geoCommands()
+		return template.doCreateFlux(connection -> connection.geoCommands()
 				.geoSearch(rawKey(key), rawReference, geoPredicate, args).map(this::readGeoResult));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveGeoOperations#searchAndStore(K, K, RedisGeoCommands.GeoReference, GeoShape, GeoSearchStoreCommandArgs)
 	 */
@@ -366,7 +367,7 @@ class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations<K, V>
 		Assert.notNull(reference, "GeoReference must not be null!");
 		GeoReference<ByteBuffer> rawReference = getGeoReference(reference);
 
-		return template.createMono(connection -> connection.geoCommands().geoSearchStore(rawKey(destKey), rawKey(key),
+		return template.doCreateMono(connection -> connection.geoCommands().geoSearchStore(rawKey(destKey), rawKey(key),
 				rawReference, geoPredicate, args));
 	}
 
@@ -374,14 +375,14 @@ class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations<K, V>
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.geoCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.geoCommands()));
 	}
 
 	private <T> Flux<T> createFlux(Function<ReactiveGeoCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createFlux(connection -> function.apply(connection.geoCommands()));
+		return template.doCreateFlux(connection -> function.apply(connection.geoCommands()));
 	}
 
 	@SuppressWarnings("unchecked")

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

@@ -15,20 +15,20 @@
  */
 package org.springframework.data.redis.core;
 
-import org.springframework.data.redis.connection.convert.Converters;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.data.redis.connection.ReactiveHashCommands;
+import org.springframework.data.redis.connection.convert.Converters;
 import org.springframework.data.redis.serializer.RedisSerializationContext;
 import org.springframework.util.Assert;
 
@@ -125,7 +125,7 @@ class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(hashKey, "Hash key must not be null!");
 
-		return template.createMono(connection -> connection //
+		return template.doCreateMono(connection -> connection //
 				.numberCommands() //
 				.hIncrBy(rawKey(key), rawHashKey(hashKey), delta));
 	}
@@ -140,7 +140,7 @@ class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(hashKey, "Hash key must not be null!");
 
-		return template.createMono(connection -> connection //
+		return template.doCreateMono(connection -> connection //
 				.numberCommands() //
 				.hIncrBy(rawKey(key), rawHashKey(hashKey), delta));
 	}
@@ -154,7 +154,7 @@ class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection //
+		return template.doCreateMono(connection -> connection //
 				.hashCommands().hRandField(rawKey(key))).map(this::readHashKey);
 	}
 
@@ -167,7 +167,7 @@ class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection //
+		return template.doCreateMono(connection -> connection //
 				.hashCommands().hRandFieldWithValues(rawKey(key))).map(this::deserializeHashEntry);
 	}
 
@@ -180,7 +180,7 @@ class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createFlux(connection -> connection //
+		return template.doCreateFlux(connection -> connection //
 				.hashCommands().hRandField(rawKey(key), count)).map(this::readHashKey);
 	}
 
@@ -193,7 +193,7 @@ class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createFlux(connection -> connection //
+		return template.doCreateFlux(connection -> connection //
 				.hashCommands().hRandFieldWithValues(rawKey(key), count)).map(this::deserializeHashEntry);
 	}
 
@@ -314,21 +314,21 @@ class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
+		return template.doCreateMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
 	}
 
 	private <T> Mono<T> createMono(Function<ReactiveHashCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.hashCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.hashCommands()));
 	}
 
 	private <T> Flux<T> createFlux(Function<ReactiveHashCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createFlux(connection -> function.apply(connection.hashCommands()));
+		return template.doCreateFlux(connection -> function.apply(connection.hashCommands()));
 	}
 
 	private ByteBuffer rawKey(H key) {

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

@@ -103,14 +103,14 @@ class DefaultReactiveHyperLogLogOperations<K, V> implements ReactiveHyperLogLogO
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
+		return template.doCreateMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
 	}
 
 	private <T> Mono<T> createMono(Function<ReactiveHyperLogLogCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.hyperLogLogCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.hyperLogLogCommands()));
 	}
 
 	private ByteBuffer rawKey(K key) {

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

@@ -200,7 +200,7 @@ class DefaultReactiveListOperations<K, V> implements ReactiveListOperations<K, V
 		return createMono(connection -> connection.lInsert(rawKey(key), Position.AFTER, rawValue(pivot), rawValue(value)));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveListOperations#move(K, Direction, K, Direction)
 	 */
@@ -216,7 +216,7 @@ class DefaultReactiveListOperations<K, V> implements ReactiveListOperations<K, V
 				connection -> connection.lMove(rawKey(sourceKey), rawKey(destinationKey), from, to).map(this::readValue));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveListOperations#move(K, Direction, K, Direction, Duration)
 	 */
@@ -378,21 +378,21 @@ class DefaultReactiveListOperations<K, V> implements ReactiveListOperations<K, V
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
+		return template.doCreateMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
 	}
 
 	private <T> Mono<T> createMono(Function<ReactiveListCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.listCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.listCommands()));
 	}
 
 	private <T> Flux<T> createFlux(Function<ReactiveListCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createFlux(connection -> function.apply(connection.listCommands()));
+		return template.doCreateFlux(connection -> function.apply(connection.listCommands()));
 	}
 
 	private boolean isZeroOrGreater1Second(Duration timeout) {

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

@@ -508,21 +508,21 @@ class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations<K, V>
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
+		return template.doCreateMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
 	}
 
 	private <T> Mono<T> createMono(Function<ReactiveSetCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.setCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.setCommands()));
 	}
 
 	private <T> Flux<T> createFlux(Function<ReactiveSetCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createFlux(connection -> function.apply(connection.setCommands()));
+		return template.doCreateFlux(connection -> function.apply(connection.setCommands()));
 	}
 
 	private ByteBuffer rawKey(K key) {

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

@@ -15,7 +15,6 @@
  */
 package org.springframework.data.redis.core;
 
-import org.springframework.data.redis.connection.convert.Converters;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -28,10 +27,12 @@ import java.util.Map.Entry;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.core.convert.ConversionService;
 import org.springframework.data.domain.Range;
 import org.springframework.data.redis.connection.ReactiveStreamCommands;
 import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
+import org.springframework.data.redis.connection.convert.Converters;
 import org.springframework.data.redis.connection.stream.ByteBufferRecord;
 import org.springframework.data.redis.connection.stream.Consumer;
 import org.springframework.data.redis.connection.stream.MapRecord;
@@ -372,14 +373,14 @@ class DefaultReactiveStreamOperations<K, HK, HV> implements ReactiveStreamOperat
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.streamCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.streamCommands()));
 	}
 
 	private <T> Flux<T> createFlux(Function<ReactiveStreamCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createFlux(connection -> function.apply(connection.streamCommands()));
+		return template.doCreateFlux(connection -> function.apply(connection.streamCommands()));
 	}
 
 	private ByteBuffer rawKey(K key) {

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

@@ -244,7 +244,7 @@ class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K,
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.numberCommands().incr(rawKey(key)));
+		return template.doCreateMono(connection -> connection.numberCommands().incr(rawKey(key)));
 	}
 
 	/* (non-Javadoc)
@@ -255,7 +255,7 @@ class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K,
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.numberCommands().incrBy(rawKey(key), delta));
+		return template.doCreateMono(connection -> connection.numberCommands().incrBy(rawKey(key), delta));
 	}
 
 	/* (non-Javadoc)
@@ -266,7 +266,7 @@ class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K,
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.numberCommands().incrBy(rawKey(key), delta));
+		return template.doCreateMono(connection -> connection.numberCommands().incrBy(rawKey(key), delta));
 	}
 
 	/* (non-Javadoc)
@@ -277,7 +277,7 @@ class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K,
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.numberCommands().decr(rawKey(key)));
+		return template.doCreateMono(connection -> connection.numberCommands().decr(rawKey(key)));
 	}
 
 	/* (non-Javadoc)
@@ -288,7 +288,7 @@ class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K,
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.numberCommands().decrBy(rawKey(key), delta));
+		return template.doCreateMono(connection -> connection.numberCommands().decrBy(rawKey(key), delta));
 	}
 
 	/* (non-Javadoc)
@@ -380,14 +380,14 @@ class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K,
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
+		return template.doCreateMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
 	}
 
 	private <T> Mono<T> createMono(Function<ReactiveStringCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.stringCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.stringCommands()));
 	}
 
 	private ByteBuffer rawKey(K key) {

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

@@ -120,7 +120,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createMono(connection -> connection.zIncrBy(rawKey(key), delta, rawValue(value)));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#randomMember(K)
 	 */
@@ -132,7 +132,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createMono(connection -> connection.zRandMember(rawKey(key))).map(this::readValue);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#distinctRandomMembers(K, long)
 	 */
@@ -145,7 +145,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createFlux(connection -> connection.zRandMember(rawKey(key), count)).map(this::readValue);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#randomMembers(K, long)
 	 */
@@ -158,7 +158,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createFlux(connection -> connection.zRandMember(rawKey(key), -count)).map(this::readValue);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#randomMemberWithScore(K)
 	 */
@@ -170,7 +170,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createMono(connection -> connection.zRandMemberWithScore(rawKey(key))).map(this::readTypedTuple);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#distinctRandomMembersWithScore(K, long)
 	 */
@@ -183,7 +183,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createFlux(connection -> connection.zRandMemberWithScore(rawKey(key), count)).map(this::readTypedTuple);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#randomMembersWithScore(K, long)
 	 */
@@ -576,7 +576,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createMono(connection -> connection.zRemRangeByScore(rawKey(key), range));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#difference(K, Collection)
 	 */
@@ -592,7 +592,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				.flatMapMany(connection::zDiff).map(this::readValue));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#differenceWithScores(K, Collection)
 	 */
@@ -608,7 +608,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				.flatMapMany(connection::zDiffWithScores).map(this::readTypedTuple));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#differenceAndStore(K, Collection, K)
 	 */
@@ -626,7 +626,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#intersect(K, Collection)
 	 */
@@ -642,7 +642,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				.flatMapMany(connection::zInter).map(this::readValue));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#intersectWithScores(K, Collection)
 	 */
@@ -658,7 +658,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				.flatMapMany(connection::zInterWithScores).map(this::readTypedTuple));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#intersectWithScores(K, Collection, Aggregate, Weights)
 	 */
@@ -714,7 +714,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				.flatMap(serialized -> connection.zInterStore(rawKey(destKey), serialized, weights, aggregate)));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#union(K, Collection)
 	 */
@@ -730,7 +730,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				.flatMapMany(connection::zUnion).map(this::readValue));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#unionWithScores(K, Collection)
 	 */
@@ -746,7 +746,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				.flatMapMany(connection::zUnionWithScores).map(this::readTypedTuple));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#unionWithScores(K, Collection, Aggregate, Weights)
 	 */
@@ -877,21 +877,21 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
+		return template.doCreateMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
 	}
 
 	private <T> Mono<T> createMono(Function<ReactiveZSetCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.zSetCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.zSetCommands()));
 	}
 
 	private <T> Flux<T> createFlux(Function<ReactiveZSetCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createFlux(connection -> function.apply(connection.zSetCommands()));
+		return template.doCreateFlux(connection -> function.apply(connection.zSetCommands()));
 	}
 
 	private ByteBuffer rawKey(K key) {

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

@@ -173,6 +173,21 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		return Flux.from(doInConnection(callback, exposeConnection));
 	}
 
+	/**
+	 * Internal variant of {@link #createFlux(ReactiveRedisCallback)} bypassing proxy creation. Create a reusable Flux for
+	 * a {@link ReactiveRedisCallback}. Callback is executed within a connection context. The connection is released
+	 * outside the callback.
+	 *
+	 * @param callback must not be {@literal null}
+	 * @return a {@link Flux} wrapping the {@link ReactiveRedisCallback}.
+	 */
+	<T> Flux<T> doCreateFlux(ReactiveRedisCallback<T> callback) {
+
+		Assert.notNull(callback, "ReactiveRedisCallback must not be null!");
+
+		return Flux.from(doInConnection(callback, true));
+	}
+
 	/**
 	 * Create a reusable Mono for a {@link ReactiveRedisCallback}. Callback is executed within a connection context. The
 	 * connection is released outside the callback.
@@ -187,6 +202,21 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		return Mono.from(doInConnection(callback, exposeConnection));
 	}
 
+	/**
+	 * Internal variant of {@link #createMono(ReactiveRedisCallback)} bypassing proxy creation. Create a reusable Mono for
+	 * a {@link ReactiveRedisCallback}. Callback is executed within a connection context. The connection is released
+	 * outside the callback.
+	 *
+	 * @param callback must not be {@literal null}
+	 * @return a {@link Mono} wrapping the {@link ReactiveRedisCallback}.
+	 */
+	<T> Mono<T> doCreateMono(ReactiveRedisCallback<T> callback) {
+
+		Assert.notNull(callback, "ReactiveRedisCallback must not be null!");
+
+		return Mono.from(doInConnection(callback, true));
+	}
+
 	/**
 	 * Executes the given action object within a connection that can be exposed or not. Additionally, the connection can
 	 * be pipelined. Note the results of the pipeline are discarded (making it suitable for write-only scenarios).
@@ -225,7 +255,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.hasText(destination, "Destination channel must not be empty!");
 		Assert.notNull(message, "Message must not be null!");
 
-		return createMono(connection -> connection.pubSubCommands().publish(
+		return doCreateMono(connection -> connection.pubSubCommands().publish(
 				getSerializationContext().getStringSerializationPair().write(destination),
 				getSerializationContext().getValueSerializationPair().write(message)));
 	}
@@ -274,7 +304,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.notNull(sourceKey, "Source key must not be null!");
 		Assert.notNull(targetKey, "Target key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().copy(rawKey(sourceKey), rawKey(targetKey), replace));
+		return doCreateMono(connection -> connection.keyCommands().copy(rawKey(sourceKey), rawKey(targetKey), replace));
 	}
 
 	/*
@@ -286,7 +316,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().exists(rawKey(key)));
+		return doCreateMono(connection -> connection.keyCommands().exists(rawKey(key)));
 	}
 
 	/*
@@ -298,7 +328,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().type(rawKey(key)));
+		return doCreateMono(connection -> connection.keyCommands().type(rawKey(key)));
 	}
 
 	/*
@@ -310,7 +340,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(pattern, "Pattern must not be null!");
 
-		return createFlux(connection -> connection.keyCommands().keys(rawKey(pattern))) //
+		return doCreateFlux(connection -> connection.keyCommands().keys(rawKey(pattern))) //
 				.flatMap(Flux::fromIterable) //
 				.map(this::readKey);
 	}
@@ -324,7 +354,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(options, "ScanOptions must not be null!");
 
-		return createFlux(connection -> connection.keyCommands().scan(options)) //
+		return doCreateFlux(connection -> connection.keyCommands().scan(options)) //
 				.map(this::readKey);
 	}
 
@@ -334,7 +364,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 	 */
 	@Override
 	public Mono<K> randomKey() {
-		return createMono(connection -> connection.keyCommands().randomKey()).map(this::readKey);
+		return doCreateMono(connection -> connection.keyCommands().randomKey()).map(this::readKey);
 	}
 
 	/*
@@ -347,7 +377,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.notNull(oldKey, "Old key must not be null!");
 		Assert.notNull(newKey, "New Key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().rename(rawKey(oldKey), rawKey(newKey)));
+		return doCreateMono(connection -> connection.keyCommands().rename(rawKey(oldKey), rawKey(newKey)));
 	}
 
 	/*
@@ -360,7 +390,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.notNull(oldKey, "Old key must not be null!");
 		Assert.notNull(newKey, "New Key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().renameNX(rawKey(oldKey), rawKey(newKey)));
+		return doCreateMono(connection -> connection.keyCommands().renameNX(rawKey(oldKey), rawKey(newKey)));
 	}
 
 	/*
@@ -376,11 +406,11 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.noNullElements(keys, "Keys must not contain null elements!");
 
 		if (keys.length == 1) {
-			return createMono(connection -> connection.keyCommands().del(rawKey(keys[0])));
+			return doCreateMono(connection -> connection.keyCommands().del(rawKey(keys[0])));
 		}
 
 		Mono<List<ByteBuffer>> listOfKeys = Flux.fromArray(keys).map(this::rawKey).collectList();
-		return createMono(connection -> listOfKeys.flatMap(rawKeys -> connection.keyCommands().mDel(rawKeys)));
+		return doCreateMono(connection -> listOfKeys.flatMap(rawKeys -> connection.keyCommands().mDel(rawKeys)));
 	}
 
 	/*
@@ -392,7 +422,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(keys, "Keys must not be null!");
 
-		return createFlux(connection -> connection.keyCommands() //
+		return doCreateFlux(connection -> connection.keyCommands() //
 				.mDel(Flux.from(keys).map(this::rawKey).buffer(128)) //
 				.map(CommandResponse::getOutput)) //
 						.collect(Collectors.summingLong(value -> value));
@@ -411,11 +441,11 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.noNullElements(keys, "Keys must not contain null elements!");
 
 		if (keys.length == 1) {
-			return createMono(connection -> connection.keyCommands().unlink(rawKey(keys[0])));
+			return doCreateMono(connection -> connection.keyCommands().unlink(rawKey(keys[0])));
 		}
 
 		Mono<List<ByteBuffer>> listOfKeys = Flux.fromArray(keys).map(this::rawKey).collectList();
-		return createMono(connection -> listOfKeys.flatMap(rawKeys -> connection.keyCommands().mUnlink(rawKeys)));
+		return doCreateMono(connection -> listOfKeys.flatMap(rawKeys -> connection.keyCommands().mUnlink(rawKeys)));
 	}
 
 	/*
@@ -427,7 +457,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(keys, "Keys must not be null!");
 
-		return createFlux(connection -> connection.keyCommands() //
+		return doCreateFlux(connection -> connection.keyCommands() //
 				.mUnlink(Flux.from(keys).map(this::rawKey).buffer(128)) //
 				.map(CommandResponse::getOutput)) //
 						.collect(Collectors.summingLong(value -> value));
@@ -444,11 +474,11 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.notNull(timeout, "Timeout must not be null!");
 
 		if (timeout.getNano() == 0) {
-			return createMono(connection -> connection.keyCommands() //
+			return doCreateMono(connection -> connection.keyCommands() //
 					.expire(rawKey(key), timeout));
 		}
 
-		return createMono(connection -> connection.keyCommands().pExpire(rawKey(key), timeout));
+		return doCreateMono(connection -> connection.keyCommands().pExpire(rawKey(key), timeout));
 	}
 
 	/*
@@ -462,11 +492,11 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.notNull(expireAt, "Expire at must not be null!");
 
 		if (expireAt.getNano() == 0) {
-			return createMono(connection -> connection.keyCommands() //
+			return doCreateMono(connection -> connection.keyCommands() //
 					.expireAt(rawKey(key), expireAt));
 		}
 
-		return createMono(connection -> connection.keyCommands().pExpireAt(rawKey(key), expireAt));
+		return doCreateMono(connection -> connection.keyCommands().pExpireAt(rawKey(key), expireAt));
 	}
 
 	/*
@@ -478,7 +508,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().persist(rawKey(key)));
+		return doCreateMono(connection -> connection.keyCommands().persist(rawKey(key)));
 	}
 
 	/*
@@ -490,7 +520,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().pTtl(rawKey(key)).flatMap(expiry -> {
+		return doCreateMono(connection -> connection.keyCommands().pTtl(rawKey(key)).flatMap(expiry -> {
 
 			if (expiry == -1) {
 				return Mono.just(Duration.ZERO);
@@ -513,7 +543,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().move(rawKey(key), dbIndex));
+		return doCreateMono(connection -> connection.keyCommands().move(rawKey(key), dbIndex));
 	}
 
 	// -------------------------------------------------------------------------