DATAREDIS-683 - Provide reactive Lua script execution commands.

We now support Redis Lua script execution using reactive infrastructure through the reactive connection and template API.

Flux<V> execute = reactiveTemplate.execute(new DefaultRedisScript<>("return redis.call('get', KEYS[1])", String.class), Collections.singletonList(key));

Release the reactive connection consistently after Publisher termination. Extract shared code between DefaultScriptExecutor and DefaultScriptOperators in ScriptUtils.

Original Pull Request: #268

src/main/java/org/springframework/data/redis/connection/ReactiveClusterScriptingCommands.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright 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.data.redis.connection;
+
+/**
+ * @author Mark Paluch
+ * @since 2.0
+ */
+public interface ReactiveClusterScriptingCommands extends ReactiveScriptingCommands {
+
+}

src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java

@@ -115,6 +115,13 @@ public interface ReactiveRedisConnection extends Closeable {
 	 */
 	ReactiveHyperLogLogCommands hyperLogLogCommands();
 
+	/**
+	 * Get {@link ReactiveScriptingCommands}.
+	 *
+	 * @return never {@literal null}.
+	 */
+	ReactiveScriptingCommands scriptingCommands();
+
 	/**
 	 * Get {@link ReactiveServerCommands}.
 	 *

src/main/java/org/springframework/data/redis/connection/ReactiveScriptingCommands.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 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.data.redis.connection;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Redis Scripting commands executed using reactive infrastructure.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ */
+public interface ReactiveScriptingCommands {
+
+	/**
+	 * Flush lua script cache.
+	 *
+	 * @see <a href="http://redis.io/commands/script-flush">Redis Documentation: SCRIPT FLUSH</a>
+	 */
+	Mono<String> scriptFlush();
+
+	/**
+	 * Kill current lua script execution.
+	 *
+	 * @see <a href="http://redis.io/commands/script-kill">Redis Documentation: SCRIPT KILL</a>
+	 */
+	Mono<String> scriptKill();
+
+	/**
+	 * Load lua script into scripts cache, without executing it.<br>
+	 * Execute the script by calling {@link #evalSha(String, ReturnType, int, ByteBuffer...)}.
+	 *
+	 * @param script must not be {@literal null}.
+	 * @return
+	 * @see <a href="http://redis.io/commands/script-load">Redis Documentation: SCRIPT LOAD</a>
+	 */
+	Mono<String> scriptLoad(ByteBuffer script);
+
+	/**
+	 * Check if given {@code scriptShas} exist in script cache.
+	 *
+	 * @param scriptShas
+	 * @return one entry per given scriptSha in returned list.
+	 * @see <a href="http://redis.io/commands/script-exists">Redis Documentation: SCRIPT EXISTS</a>
+	 */
+	Flux<Boolean> scriptExists(String... scriptShas);
+
+	/**
+	 * Evaluate given {@code script}.
+	 *
+	 * @param script must not be {@literal null}.
+	 * @param returnType must not be {@literal null}.
+	 * @param numKeys
+	 * @param keysAndArgs must not be {@literal null}.
+	 * @return
+	 * @see <a href="http://redis.io/commands/eval">Redis Documentation: EVAL</a>
+	 */
+	<T> Flux<T> eval(ByteBuffer script, ReturnType returnType, int numKeys, ByteBuffer... keysAndArgs);
+
+	/**
+	 * Evaluate given {@code scriptSha}.
+	 *
+	 * @param scriptSha must not be {@literal null}.
+	 * @param returnType must not be {@literal null}.
+	 * @param numKeys
+	 * @param keysAndArgs must not be {@literal null}.
+	 * @return
+	 * @see <a href="http://redis.io/commands/evalsha">Redis Documentation: EVALSHA</a>
+	 */
+	<T> Flux<T> evalSha(String scriptSha, ReturnType returnType, int numKeys, ByteBuffer... keysAndArgs);
+}

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterScriptingCommands.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 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.data.redis.connection.lettuce;
+
+import org.springframework.data.redis.connection.ReactiveClusterScriptingCommands;
+
+/**
+ * @author Mark Paluch
+ * @since 2.0
+ */
+class LettuceReactiveClusterScriptingCommands extends LettuceReactiveScriptingCommands
+		implements ReactiveClusterScriptingCommands {
+
+	/**
+	 * Create new {@link LettuceReactiveStringCommands}.
+	 *
+	 * @param connection must not be {@literal null}.
+	 */
+	LettuceReactiveClusterScriptingCommands(LettuceReactiveRedisConnection connection) {
+		super(connection);
+	}
+}

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnection.java

@@ -140,6 +140,15 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
 		return new LettuceReactiveClusterNumberCommands(this);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#scriptingCommands()
+	 */
+	@Override
+	public LettuceReactiveClusterScriptingCommands scriptingCommands() {
+		return new LettuceReactiveClusterScriptingCommands(this);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#serverCommands()

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java

@@ -24,6 +24,7 @@ import io.lettuce.core.cluster.RedisClusterClient;
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
 import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands;
 import io.lettuce.core.codec.RedisCodec;
+import org.springframework.data.redis.connection.*;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -33,17 +34,6 @@ import java.util.function.Function;
 import org.reactivestreams.Publisher;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessResourceUsageException;
-import org.springframework.data.redis.connection.ReactiveGeoCommands;
-import org.springframework.data.redis.connection.ReactiveHashCommands;
-import org.springframework.data.redis.connection.ReactiveHyperLogLogCommands;
-import org.springframework.data.redis.connection.ReactiveKeyCommands;
-import org.springframework.data.redis.connection.ReactiveListCommands;
-import org.springframework.data.redis.connection.ReactiveNumberCommands;
-import org.springframework.data.redis.connection.ReactiveRedisConnection;
-import org.springframework.data.redis.connection.ReactiveServerCommands;
-import org.springframework.data.redis.connection.ReactiveSetCommands;
-import org.springframework.data.redis.connection.ReactiveStringCommands;
-import org.springframework.data.redis.connection.ReactiveZSetCommands;
 import org.springframework.util.Assert;
 
 /**
@@ -159,6 +149,15 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 		return new LettuceReactiveHyperLogLogCommands(this);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection#scriptingCommands()
+	 */
+	@Override
+	public ReactiveScriptingCommands scriptingCommands() {
+		return new LettuceReactiveScriptingCommands(this);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection#hyperLogLogCommands()

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommands.java

@@ -0,0 +1,158 @@
+/*
+ * Copyright 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.data.redis.connection.lettuce;
+
+import io.lettuce.core.api.reactive.RedisScriptingReactiveCommands;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.Arrays;
+
+import org.springframework.data.redis.connection.ReactiveScriptingCommands;
+import org.springframework.data.redis.connection.ReturnType;
+import org.springframework.util.Assert;
+
+/**
+ * @author Mark Paluch
+ * @since 2.0
+ */
+class LettuceReactiveScriptingCommands implements ReactiveScriptingCommands {
+
+	private static final ByteBuffer[] EMPTY_BUFFER_ARRAY = new ByteBuffer[0];
+
+	private final LettuceReactiveRedisConnection connection;
+
+	/**
+	 * Create new {@link LettuceReactiveScriptingCommands}.
+	 *
+	 * @param connection must not be {@literal null}.
+	 */
+	LettuceReactiveScriptingCommands(LettuceReactiveRedisConnection connection) {
+
+		Assert.notNull(connection, "Connection must not be null!");
+
+		this.connection = connection;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveScriptingCommands#scriptFlush()
+	 */
+	@Override
+	public Mono<String> scriptFlush() {
+		return connection.execute(RedisScriptingReactiveCommands::scriptFlush).next();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveScriptingCommands#scriptKill()
+	 */
+	@Override
+	public Mono<String> scriptKill() {
+		return connection.execute(RedisScriptingReactiveCommands::scriptKill).next();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveScriptingCommands#scriptLoad(java.nio.ByteBuffer)
+	 */
+	@Override
+	public Mono<String> scriptLoad(ByteBuffer script) {
+
+		Assert.notNull(script, "Script must not be null!");
+
+		return connection.execute(cmd -> cmd.scriptLoad(script)).next();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveScriptingCommands#scriptExists(java.lang.String[])
+	 */
+	@Override
+	public Flux<Boolean> scriptExists(String... scriptShas) {
+
+		Assert.notNull(scriptShas, "Script SHAs must not be null!");
+
+		return connection.execute(cmd -> cmd.scriptExists(scriptShas));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveScriptingCommands#eval(java.nio.ByteBuffer, org.springframework.data.redis.connection.ReturnType, int, java.nio.ByteBuffer[])
+	 */
+	@Override
+	public <T> Flux<T> eval(ByteBuffer script, ReturnType returnType, int numKeys, ByteBuffer... keysAndArgs) {
+
+		Assert.notNull(script, "Script must not be null!");
+		Assert.notNull(returnType, "ReturnType must not be null!");
+		Assert.notNull(keysAndArgs, "Keys and args must not be null!");
+
+		ByteBuffer[] keys = extractScriptKeys(numKeys, keysAndArgs);
+		ByteBuffer[] args = extractScriptArgs(numKeys, keysAndArgs);
+
+		String scriptBody = Charset.defaultCharset().decode(script).toString();
+
+		return convertIfNecessary(
+				connection.execute(cmd -> cmd.eval(scriptBody, LettuceConverters.toScriptOutputType(returnType), keys, args)),
+				returnType);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveScriptingCommands#evalSha(java.lang.String, org.springframework.data.redis.connection.ReturnType, int, java.nio.ByteBuffer[])
+	 */
+	@Override
+	public <T> Flux<T> evalSha(String scriptSha, ReturnType returnType, int numKeys, ByteBuffer... keysAndArgs) {
+
+		Assert.notNull(scriptSha, "Script SHA1 must not be null!");
+		Assert.notNull(returnType, "ReturnType must not be null!");
+		Assert.notNull(keysAndArgs, "Keys and args must not be null!");
+
+		ByteBuffer[] keys = extractScriptKeys(numKeys, keysAndArgs);
+		ByteBuffer[] args = extractScriptArgs(numKeys, keysAndArgs);
+
+		return convertIfNecessary(
+				connection.execute(cmd -> cmd.evalsha(scriptSha, LettuceConverters.toScriptOutputType(returnType), keys, args)),
+				returnType);
+	}
+
+	@SuppressWarnings("unchecked")
+	private <T> Flux<T> convertIfNecessary(Flux<T> eval, ReturnType returnType) {
+
+		if (returnType == ReturnType.MULTI) {
+
+			return eval.flatMap(t -> {
+				return t instanceof Iterable ? Flux.fromIterable((Iterable<T>) t) : Flux.just(t);
+			}).flatMap(t -> {
+				return t instanceof Exception ? Flux.error(connection.translateException().apply((Exception) t)) : Flux.just(t);
+			});
+		}
+
+		return eval;
+	}
+
+	private static ByteBuffer[] extractScriptKeys(int numKeys, ByteBuffer... keysAndArgs) {
+		return numKeys > 0 ? Arrays.copyOfRange(keysAndArgs, 0, numKeys) : EMPTY_BUFFER_ARRAY;
+	}
+
+	private static ByteBuffer[] extractScriptArgs(int numKeys, ByteBuffer... keysAndArgs) {
+
+		return keysAndArgs.length > numKeys ? Arrays.copyOfRange(keysAndArgs, numKeys, keysAndArgs.length)
+				: EMPTY_BUFFER_ARRAY;
+	}
+}

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

@@ -15,15 +15,19 @@
  */
 package org.springframework.data.redis.core;
 
-import org.springframework.data.redis.serializer.RedisSerializationContext;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.List;
 
 import org.reactivestreams.Publisher;
 import org.springframework.data.redis.connection.DataType;
+import org.springframework.data.redis.core.script.RedisScript;
+import org.springframework.data.redis.serializer.RedisSerializationContext;
+import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
+import org.springframework.data.redis.serializer.RedisSerializer;
 
 /**
  * Interface that specified a basic set of Redis operations, implemented by {@link ReactiveRedisTemplate}. Not often
@@ -173,6 +177,36 @@ public interface ReactiveRedisOperations<K, V> {
 	 */
 	Mono<Duration> getExpire(K key);
 
+	// -------------------------------------------------------------------------
+	// Methods dealing with Redis Lua scripts
+	// -------------------------------------------------------------------------
+
+	/**
+	 * Executes the given {@link RedisScript}
+	 *
+	 * @param script The script to execute
+	 * @param keys keys that need to be passed to the script.
+	 * @param args args that need to be passed to the script.
+	 * @return return value of the script or raw {@link java.nio.ByteBuffer} if {@link RedisScript#getResultType()} is
+	 *         {@literal null}, likely indicating a throw-away status reply (i.e. "OK").
+	 */
+	<T> Flux<T> execute(RedisScript<T> script, List<K> keys, Object... args);
+
+	/**
+	 * Executes the given {@link RedisScript}, using the provided {@link RedisSerializer}s to serialize the script
+	 * arguments and result.
+	 *
+	 * @param script The script to execute
+	 * @param argsSerializerPair The {@link SerializationPair} to use for serializing args
+	 * @param resultSerializerPair The {@link SerializationPair} to use for serializing the script return value
+	 * @param keys keys that need to be passed to the script.
+	 * @param args args that need to be passed to the script.
+	 * @return return value of the script or raw {@link java.nio.ByteBuffer} if {@link RedisScript#getResultType()} is
+	 *         {@literal null}, likely indicating a throw-away status reply (i.e. "OK").
+	 */
+	<T> Flux<T> execute(RedisScript<T> script, SerializationPair<?> argsSerializerPair,
+			SerializationPair<T> resultSerializerPair, List<K> keys, Object... args);
+
 	// -------------------------------------------------------------------------
 	// Methods to obtain specific operations interface objects.
 	// -------------------------------------------------------------------------

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

@@ -30,7 +30,11 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
 import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
+import org.springframework.data.redis.core.script.DefaultScriptOperators;
+import org.springframework.data.redis.core.script.RedisScript;
+import org.springframework.data.redis.core.script.ScriptOperators;
 import org.springframework.data.redis.serializer.RedisSerializationContext;
+import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
@@ -54,6 +58,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 	private final ReactiveRedisConnectionFactory connectionFactory;
 	private final RedisSerializationContext<K, V> serializationContext;
 	private final boolean exposeConnection;
+	private final ScriptOperators<K> scriptOperators;
 
 	/**
 	 * Creates new {@link ReactiveRedisTemplate} using given {@link ReactiveRedisConnectionFactory} and
@@ -84,6 +89,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		this.connectionFactory = connectionFactory;
 		this.serializationContext = serializationContext;
 		this.exposeConnection = exposeConnection;
+		this.scriptOperators = new DefaultScriptOperators<>(connectionFactory, serializationContext);
 	}
 
 	/**
@@ -130,6 +136,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForSet()
 	 */
+	@Override
 	public ReactiveSetOperations<K, V> opsForSet() {
 		return opsForSet(serializationContext);
 	}
@@ -145,6 +152,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#opsForZSet()
 	 */
+	@Override
 	public ReactiveZSetOperations<K, V> opsForZSet() {
 		return opsForZSet(serializationContext);
 	}
@@ -211,6 +219,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 	// Execution methods
 	// -------------------------------------------------------------------------
 
+	@Override
 	public <T> Flux<T> execute(ReactiveRedisCallback<T> action) {
 		return execute(action, exposeConnection);
 	}
@@ -238,9 +247,10 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 			ReactiveRedisConnection connToExpose = (exposeConnection ? connToUse : createRedisConnectionProxy(connToUse));
 			Publisher<T> result = action.doInRedis(connToExpose);
 
-			return Flux.from(postProcessResult(result, connToUse, false));
-		} finally {
+			return Flux.from(postProcessResult(result, connToUse, false)).doFinally(signalType -> conn.close());
+		} catch (RuntimeException e) {
 			conn.close();
+			throw e;
 		}
 	}
 
@@ -293,7 +303,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		ReactiveRedisConnection connToExpose = (exposeConnection ? connToUse : createRedisConnectionProxy(connToUse));
 		Publisher<T> result = action.doInRedis(connToExpose);
 
-		return Flux.from(postProcessResult(result, connToUse, false)).doAfterTerminate(conn::close);
+		return Flux.from(postProcessResult(result, connToUse, false)).doFinally(signal -> conn.close());
 	}
 
 	// -------------------------------------------------------------------------
@@ -303,6 +313,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#hasKey(java.lang.Object)
 	 */
+	@Override
 	public Mono<Boolean> hasKey(K key) {
 
 		Assert.notNull(key, "Key must not be null!");
@@ -313,6 +324,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#type(java.lang.Object)
 	 */
+	@Override
 	public Mono<DataType> type(K key) {
 
 		Assert.notNull(key, "Key must not be null!");
@@ -323,6 +335,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#keys(java.lang.Object)
 	 */
+	@Override
 	public Flux<K> keys(K pattern) {
 
 		Assert.notNull(pattern, "Pattern must not be null!");
@@ -335,6 +348,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#randomKey()
 	 */
+	@Override
 	public Mono<K> randomKey() {
 		return createMono(connection -> connection.keyCommands().randomKey()).map(this::readKey);
 	}
@@ -342,6 +356,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#rename(java.lang.Object, java.lang.Object)
 	 */
+	@Override
 	public Mono<Boolean> rename(K oldKey, K newKey) {
 
 		Assert.notNull(oldKey, "Old key must not be null!");
@@ -353,18 +368,19 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#renameIfAbsent(java.lang.Object, java.lang.Object)
 	 */
+	@Override
 	public Mono<Boolean> renameIfAbsent(K oldKey, K newKey) {
 
 		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)));
-
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#delete(java.lang.Object[])
 	 */
+	@Override
 	@SafeVarargs
 	public final Mono<Long> delete(K... keys) {
 
@@ -383,6 +399,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#delete(org.reactivestreams.Publisher)
 	 */
+	@Override
 	public Mono<Long> delete(Publisher<K> keys) {
 
 		Assert.notNull(keys, "Keys must not be null!");
@@ -470,6 +487,27 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		return createMono(connection -> connection.keyCommands().move(rawKey(key), dbIndex));
 	}
 
+	// -------------------------------------------------------------------------
+	// Methods dealing with Redis Lua scripts
+	// -------------------------------------------------------------------------
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#execute(org.springframework.data.redis.core.script.RedisScript, java.util.List, java.lang.Object[])
+	 */
+	@Override
+	public <T> Flux<T> execute(RedisScript<T> script, List<K> keys, Object... args) {
+		return scriptOperators.execute(script, keys, args);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#execute(org.springframework.data.redis.core.script.RedisScript, org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair, org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair, java.util.List, java.lang.Object[])
+	 */
+	@Override
+	public <T> Flux<T> execute(RedisScript<T> script, SerializationPair<?> argsSerializerPair,
+			SerializationPair<T> resultSerializerPair, List<K> keys, Object... args) {
+		return scriptOperators.execute(script, argsSerializerPair, resultSerializerPair, keys, args);
+	}
+
 	// -------------------------------------------------------------------------
 	// Implementation hooks and helper methods
 	// -------------------------------------------------------------------------

src/main/java/org/springframework/data/redis/core/script/DefaultScriptExecutor.java

@@ -15,7 +15,6 @@
  */
 package org.springframework.data.redis.core.script;
 
-import java.util.ArrayList;
 import java.util.List;
 
 import org.springframework.dao.NonTransientDataAccessException;
@@ -34,6 +33,7 @@ import org.springframework.data.redis.serializer.RedisSerializer;
  * @author Jennifer Hickey
  * @author Christoph Strobl
  * @author Thomas Darimont
+ * @author Mark Paluch
  * @param <K> The type of keys that may be passed during script execution
  */
 public class DefaultScriptExecutor<K> implements ScriptExecutor<K> {
@@ -78,7 +78,7 @@ public class DefaultScriptExecutor<K> implements ScriptExecutor<K> {
 			result = connection.evalSha(script.getSha1(), returnType, numKeys, keysAndArgs);
 		} catch (Exception e) {
 
-			if (!exceptionContainsNoScriptError(e)) {
+			if (!ScriptUtils.exceptionContainsNoScriptError(e)) {
 				throw e instanceof RuntimeException ? (RuntimeException) e : new RedisSystemException(e.getMessage(), e);
 			}
 
@@ -120,47 +120,12 @@ public class DefaultScriptExecutor<K> implements ScriptExecutor<K> {
 		return template.getStringSerializer().serialize(script.getScriptAsString());
 	}
 
-	@SuppressWarnings({ "rawtypes", "unchecked" })
 	protected <T> T deserializeResult(RedisSerializer<T> resultSerializer, Object result) {
-		if (result instanceof byte[]) {
-			if (resultSerializer == null) {
-				return (T) result;
-			}
-			return resultSerializer.deserialize((byte[]) result);
-		}
-		if (result instanceof List) {
-			List results = new ArrayList();
-			for (Object obj : (List) result) {
-				results.add(deserializeResult(resultSerializer, obj));
-			}
-			return (T) results;
-		}
-		return (T) result;
+		return ScriptUtils.deserializeResult(resultSerializer, result);
 	}
 
 	@SuppressWarnings("rawtypes")
 	protected RedisSerializer keySerializer() {
 		return template.getKeySerializer();
 	}
-
-	private boolean exceptionContainsNoScriptError(Exception e) {
-
-		if (!(e instanceof NonTransientDataAccessException)) {
-			return false;
-		}
-
-		Throwable current = e;
-		while (current != null) {
-
-			String exMessage = current.getMessage();
-			if (exMessage != null && exMessage.contains("NOSCRIPT")) {
-				return true;
-			}
-
-			current = current.getCause();
-		}
-
-		return false;
-	}
-
 }

src/main/java/org/springframework/data/redis/core/script/DefaultScriptOperators.java

@@ -0,0 +1,185 @@
+/*
+ * Copyright 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.data.redis.core.script;
+
+import reactor.core.publisher.Flux;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import org.springframework.data.redis.RedisSystemException;
+import org.springframework.data.redis.connection.ReactiveRedisConnection;
+import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
+import org.springframework.data.redis.connection.ReturnType;
+import org.springframework.data.redis.core.ReactiveRedisCallback;
+import org.springframework.data.redis.serializer.RedisSerializationContext;
+import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
+import org.springframework.util.Assert;
+
+/**
+ * Default implementation of {@link ScriptOperators}. Optimizes performance by attempting to execute script first using
+ * evalsha, then falling back to eval if Redis has not yet cached the script.
+ *
+ * @author Mark Paluch
+ * @param <K> The type of keys that may be passed during script execution
+ * @since 2.0
+ */
+public class DefaultScriptOperators<K> implements ScriptOperators<K> {
+
+	private final ReactiveRedisConnectionFactory connectionFactory;
+	private final RedisSerializationContext<K, ?> serializationContext;
+
+	/**
+	 * Creates a new {@link DefaultScriptOperators} given {@link ReactiveRedisConnectionFactory} and
+	 * {@link RedisSerializationContext}.
+	 *
+	 * @param connectionFactory must not be {@literal null}.
+	 * @param serializationContext must not be {@literal null}.
+	 */
+	public DefaultScriptOperators(ReactiveRedisConnectionFactory connectionFactory,
+			RedisSerializationContext<K, ?> serializationContext) {
+
+		Assert.notNull(connectionFactory, "ReactiveRedisConnectionFactory must not be null!");
+		Assert.notNull(serializationContext, "RedisSerializationContext must not be null!");
+
+		this.connectionFactory = connectionFactory;
+		this.serializationContext = serializationContext;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.redis.core.script.ScriptOperators#execute(org.springframework.data.redis.core.script.RedisScript, java.util.List, java.lang.Object[])
+	 */
+	@Override
+	@SuppressWarnings("unchecked")
+	public <T> Flux<T> execute(RedisScript<T> script, List<K> keys, Object... args) {
+
+		Assert.notNull(script, "RedisScript must not be null!");
+		Assert.notNull(keys, "Keys must not be null!");
+		Assert.notNull(args, "Args must not be null!");
+
+		// use the Template's value serializer for args and result
+		return execute(script, serializationContext.getKeySerializationPair(),
+				(SerializationPair<T>) serializationContext.getValueSerializationPair(), keys, args);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.redis.core.script.ScriptOperators#execute(org.springframework.data.redis.core.script.RedisScript, org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair, org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair, java.util.List, java.lang.Object[])
+	 */
+	@Override
+	@SuppressWarnings("unchecked")
+	public <T> Flux<T> execute(RedisScript<T> script, SerializationPair<?> argsSerializerPair,
+			SerializationPair<T> resultSerializationPair, List<K> keys, Object... args) {
+
+		Assert.notNull(script, "RedisScript must not be null!");
+		Assert.notNull(argsSerializerPair, "Argument SerializationPair must not be null!");
+		Assert.notNull(resultSerializationPair, "Result SerializationPair must not be null!");
+		Assert.notNull(keys, "Keys must not be null!");
+		Assert.notNull(args, "Args must not be null!");
+
+		return execute(connection -> {
+
+			ReturnType returnType = ReturnType.fromJavaType(script.getResultType());
+			ByteBuffer[] keysAndArgs = keysAndArgs((SerializationPair<Object>) argsSerializerPair, keys, args);
+			int keySize = keys.size();
+
+			return eval(connection, script, returnType, keySize, keysAndArgs, resultSerializationPair);
+
+		});
+	}
+
+	protected <T> Flux<T> eval(ReactiveRedisConnection connection, RedisScript<T> script, ReturnType returnType,
+			int numKeys, ByteBuffer[] keysAndArgs, SerializationPair<T> resultSerializer) {
+
+		Flux<T> result = connection.scriptingCommands().evalSha(script.getSha1(), returnType, numKeys, keysAndArgs);
+
+		result = result.onErrorResume(e -> {
+
+			if (ScriptUtils.exceptionContainsNoScriptError(e)) {
+				return connection.scriptingCommands().eval(scriptBytes(script), returnType, numKeys, keysAndArgs);
+			}
+
+			return Flux
+					.error(e instanceof RuntimeException ? (RuntimeException) e : new RedisSystemException(e.getMessage(), e));
+		});
+
+		return script.getResultType() == null ? result : deserializeResult(resultSerializer, result);
+	}
+
+	protected ByteBuffer[] keysAndArgs(SerializationPair<Object> argsSerializer, List<K> keys, Object[] args) {
+
+		int keySize = keys != null ? keys.size() : 0;
+		ByteBuffer[] keysAndArgs = new ByteBuffer[args.length + keySize];
+		int i = 0;
+
+		if (keys != null) {
+			for (K key : keys) {
+				if (key instanceof ByteBuffer) {
+					keysAndArgs[i++] = (ByteBuffer) key;
+				} else {
+					keysAndArgs[i++] = keySerializer().getWriter().write(key);
+				}
+			}
+		}
+
+		for (Object arg : args) {
+			if (arg instanceof ByteBuffer) {
+				keysAndArgs[i++] = (ByteBuffer) arg;
+			} else {
+				keysAndArgs[i++] = argsSerializer.getWriter().write(arg);
+			}
+		}
+		return keysAndArgs;
+	}
+
+	protected ByteBuffer scriptBytes(RedisScript<?> script) {
+		return serializationContext.getStringSerializationPair().getWriter().write(script.getScriptAsString());
+	}
+
+	protected <T> Flux<T> deserializeResult(SerializationPair<T> pair, Flux<T> result) {
+		return result.map(it -> ScriptUtils.deserializeResult(pair.getReader(), it));
+	}
+
+	protected SerializationPair<K> keySerializer() {
+		return serializationContext.getKeySerializationPair();
+	}
+
+	/**
+	 * Executes the given action object within a connection that is allocated eagerly and released after {@link Flux}
+	 * termination.
+	 *
+	 * @param <T> return type
+	 * @param action callback object to execute
+	 * @return object returned by the action
+	 */
+	private <T> Flux<T> execute(ReactiveRedisCallback<T> action) {
+
+		Assert.notNull(action, "Callback object must not be null");
+
+		ReactiveRedisConnectionFactory factory = getConnectionFactory();
+		ReactiveRedisConnection conn = factory.getReactiveConnection();
+
+		try {
+			return Flux.defer(() -> action.doInRedis(conn)).doFinally(signal -> conn.close());
+		} catch (RuntimeException e) {
+			conn.close();
+			throw e;
+		}
+	}
+
+	public ReactiveRedisConnectionFactory getConnectionFactory() {
+		return connectionFactory;
+	}
+}

src/main/java/org/springframework/data/redis/core/script/ScriptOperators.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 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.data.redis.core.script;
+
+import reactor.core.publisher.Flux;
+
+import java.util.List;
+
+import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
+import org.springframework.data.redis.serializer.RedisSerializer;
+
+/**
+ * Executes {@link RedisScript}s using reactive infrastructure.
+ *
+ * @author Mark Paluch
+ * @param <K> The type of keys that may be passed during script execution
+ * @since 2.0
+ */
+public interface ScriptOperators<K> {
+
+	/**
+	 * Executes the given {@link RedisScript}
+	 *
+	 * @param script The script to execute
+	 * @param keys Any keys that need to be passed to the script
+	 * @param args Any args that need to be passed to the script
+	 * @return The return value of the script or null if {@link RedisScript#getResultType()} is null, likely indicating a
+	 *         throw-away status reply (i.e. "OK")
+	 */
+	<T> Flux<T> execute(RedisScript<T> script, List<K> keys, Object... args);
+
+	/**
+	 * Executes the given {@link RedisScript}, using the provided {@link RedisSerializer}s to serialize the script
+	 * arguments and result.
+	 *
+	 * @param script The script to execute
+	 * @param argsSerializer The {@link SerializationPair} to use for serializing args
+	 * @param resultSerializer The {@link SerializationPair} to use for serializing the script return value
+	 * @param keys Any keys that need to be passed to the script
+	 * @param args Any args that need to be passed to the script
+	 * @return The return value of the script or null if {@link RedisScript#getResultType()} is null, likely indicating a
+	 *         throw-away status reply (i.e. "OK")
+	 */
+	<T> Flux<T> execute(RedisScript<T> script, SerializationPair<?> argsSerializer, SerializationPair<T> resultSerializer,
+			List<K> keys, Object... args);
+}

src/main/java/org/springframework/data/redis/core/script/ScriptUtils.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright 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.data.redis.core.script;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.dao.NonTransientDataAccessException;
+import org.springframework.data.redis.serializer.RedisElementReader;
+import org.springframework.data.redis.serializer.RedisSerializer;
+
+/**
+ * Utilities for Lua script execution and result deserialization.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ */
+class ScriptUtils {
+
+	private ScriptUtils() {}
+
+	/**
+	 * Deserialize {@code result} using {@link RedisSerializer} to the serializer type. Collection types and intermediate
+	 * collection elements are deserialized recursivly.
+	 *
+	 * @param resultSerializer must not be {@literal null}.
+	 * @param result must not be {@literal null}.
+	 * @return the deserialized result.
+	 */
+	@SuppressWarnings({ "unchecked" })
+	static <T> T deserializeResult(RedisSerializer<T> resultSerializer, Object result) {
+
+		if (result instanceof byte[]) {
+			return resultSerializer == null ? (T) result : resultSerializer.deserialize((byte[]) result);
+		}
+
+		if (result instanceof List) {
+
+			List<Object> results = new ArrayList<>(((List) result).size());
+
+			for (Object obj : (List) result) {
+				results.add(deserializeResult(resultSerializer, obj));
+			}
+
+			return (T) results;
+		}
+
+		return (T) result;
+	}
+
+	/**
+	 * Deserialize {@code result} using {@link RedisElementReader} to the reader type. Collection types and intermediate
+	 * collection elements are deserialized recursivly.
+	 *
+	 * @param resultSerializer must not be {@literal null}.
+	 * @param result must not be {@literal null}.
+	 * @return the deserialized result.
+	 */
+	@SuppressWarnings({ "unchecked" })
+	static <T> T deserializeResult(RedisElementReader<T> reader, Object result) {
+
+		if (result instanceof ByteBuffer) {
+
+			return reader == null ? (T) result : reader.read((ByteBuffer) result);
+		}
+
+		if (result instanceof List) {
+
+			List<Object> results = new ArrayList<>(((List) result).size());
+
+			for (Object obj : (List) result) {
+				results.add(deserializeResult(reader, obj));
+			}
+
+			return (T) results;
+		}
+		return (T) result;
+	}
+
+	/**
+	 * Checks whether given {@link Throwable} contains a {@code NOSCRIPT} error. {@code NOSCRIPT} is reported if a script
+	 * was attempted to execute using {@code EVALSHA}.
+	 *
+	 * @param e the exception.
+	 * @return {@literal true} if the exception or one of its causes contains a {@literal NOSCRIPT} error.
+	 */
+	static boolean exceptionContainsNoScriptError(Throwable e) {
+
+		if (!(e instanceof NonTransientDataAccessException)) {
+			return false;
+		}
+
+		Throwable current = e;
+		while (current != null) {
+
+			String exMessage = current.getMessage();
+			if (exMessage != null && exMessage.contains("NOSCRIPT")) {
+				return true;
+			}
+
+			current = current.getCause();
+		}
+
+		return false;
+	}
+}