diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveClusterScriptingCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterScriptingCommands.java new file mode 100644 index 000000000..df69ef664 --- /dev/null +++ b/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 { + +} diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java index e64dba5e9..0bf775474 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java +++ b/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}. * diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveScriptingCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveScriptingCommands.java new file mode 100644 index 000000000..aef45c341 --- /dev/null +++ b/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 Redis Documentation: SCRIPT FLUSH + */ + Mono scriptFlush(); + + /** + * Kill current lua script execution. + * + * @see Redis Documentation: SCRIPT KILL + */ + Mono scriptKill(); + + /** + * Load lua script into scripts cache, without executing it.
+ * Execute the script by calling {@link #evalSha(String, ReturnType, int, ByteBuffer...)}. + * + * @param script must not be {@literal null}. + * @return + * @see Redis Documentation: SCRIPT LOAD + */ + Mono scriptLoad(ByteBuffer script); + + /** + * Check if given {@code scriptShas} exist in script cache. + * + * @param scriptShas + * @return one entry per given scriptSha in returned list. + * @see Redis Documentation: SCRIPT EXISTS + */ + Flux 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 Redis Documentation: EVAL + */ + Flux 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 Redis Documentation: EVALSHA + */ + Flux evalSha(String scriptSha, ReturnType returnType, int numKeys, ByteBuffer... keysAndArgs); +} diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterScriptingCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterScriptingCommands.java new file mode 100644 index 000000000..39c291e6e --- /dev/null +++ b/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); + } +} diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnection.java index 0deb30ff4..c00dd596b 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnection.java +++ b/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() diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java index 8cafd4408..19fa06227 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java +++ b/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() diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommands.java new file mode 100644 index 000000000..2e62c0bf5 --- /dev/null +++ b/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 scriptFlush() { + return connection.execute(RedisScriptingReactiveCommands::scriptFlush).next(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveScriptingCommands#scriptKill() + */ + @Override + public Mono scriptKill() { + return connection.execute(RedisScriptingReactiveCommands::scriptKill).next(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveScriptingCommands#scriptLoad(java.nio.ByteBuffer) + */ + @Override + public Mono 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 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 Flux 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 Flux 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 Flux convertIfNecessary(Flux eval, ReturnType returnType) { + + if (returnType == ReturnType.MULTI) { + + return eval.flatMap(t -> { + return t instanceof Iterable ? Flux.fromIterable((Iterable) 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; + } +} diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java index 97dee9e06..724b949d5 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java +++ b/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 { */ Mono 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"). + */ + Flux execute(RedisScript script, List 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"). + */ + Flux execute(RedisScript script, SerializationPair argsSerializerPair, + SerializationPair resultSerializerPair, List keys, Object... args); + // ------------------------------------------------------------------------- // Methods to obtain specific operations interface objects. // ------------------------------------------------------------------------- diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java index 81d321aca..8fcc88338 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java +++ b/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 implements ReactiveRedisOperations serializationContext; private final boolean exposeConnection; + private final ScriptOperators scriptOperators; /** * Creates new {@link ReactiveRedisTemplate} using given {@link ReactiveRedisConnectionFactory} and @@ -84,6 +89,7 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations(connectionFactory, serializationContext); } /** @@ -130,6 +136,7 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations opsForSet() { return opsForSet(serializationContext); } @@ -145,6 +152,7 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations opsForZSet() { return opsForZSet(serializationContext); } @@ -211,6 +219,7 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations Flux execute(ReactiveRedisCallback action) { return execute(action, exposeConnection); } @@ -238,9 +247,10 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations 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 implements ReactiveRedisOperations 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 implements ReactiveRedisOperations hasKey(K key) { Assert.notNull(key, "Key must not be null!"); @@ -313,6 +324,7 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations type(K key) { Assert.notNull(key, "Key must not be null!"); @@ -323,6 +335,7 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations keys(K pattern) { Assert.notNull(pattern, "Pattern must not be null!"); @@ -335,6 +348,7 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations randomKey() { return createMono(connection -> connection.keyCommands().randomKey()).map(this::readKey); } @@ -342,6 +356,7 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations rename(K oldKey, K newKey) { Assert.notNull(oldKey, "Old key must not be null!"); @@ -353,18 +368,19 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations 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 delete(K... keys) { @@ -383,6 +399,7 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations delete(Publisher keys) { Assert.notNull(keys, "Keys must not be null!"); @@ -470,6 +487,27 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations 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 Flux execute(RedisScript script, List 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 Flux execute(RedisScript script, SerializationPair argsSerializerPair, + SerializationPair resultSerializerPair, List keys, Object... args) { + return scriptOperators.execute(script, argsSerializerPair, resultSerializerPair, keys, args); + } + // ------------------------------------------------------------------------- // Implementation hooks and helper methods // ------------------------------------------------------------------------- diff --git a/src/main/java/org/springframework/data/redis/core/script/DefaultScriptExecutor.java b/src/main/java/org/springframework/data/redis/core/script/DefaultScriptExecutor.java index 4bb513ea4..b2d52793e 100644 --- a/src/main/java/org/springframework/data/redis/core/script/DefaultScriptExecutor.java +++ b/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 The type of keys that may be passed during script execution */ public class DefaultScriptExecutor implements ScriptExecutor { @@ -78,7 +78,7 @@ public class DefaultScriptExecutor implements ScriptExecutor { 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 implements ScriptExecutor { return template.getStringSerializer().serialize(script.getScriptAsString()); } - @SuppressWarnings({ "rawtypes", "unchecked" }) protected T deserializeResult(RedisSerializer 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; - } - } diff --git a/src/main/java/org/springframework/data/redis/core/script/DefaultScriptOperators.java b/src/main/java/org/springframework/data/redis/core/script/DefaultScriptOperators.java new file mode 100644 index 000000000..a89637bb8 --- /dev/null +++ b/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 The type of keys that may be passed during script execution + * @since 2.0 + */ +public class DefaultScriptOperators implements ScriptOperators { + + private final ReactiveRedisConnectionFactory connectionFactory; + private final RedisSerializationContext 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 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 Flux execute(RedisScript script, List 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) 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 Flux execute(RedisScript script, SerializationPair argsSerializerPair, + SerializationPair resultSerializationPair, List 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) argsSerializerPair, keys, args); + int keySize = keys.size(); + + return eval(connection, script, returnType, keySize, keysAndArgs, resultSerializationPair); + + }); + } + + protected Flux eval(ReactiveRedisConnection connection, RedisScript script, ReturnType returnType, + int numKeys, ByteBuffer[] keysAndArgs, SerializationPair resultSerializer) { + + Flux 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 argsSerializer, List 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 Flux deserializeResult(SerializationPair pair, Flux result) { + return result.map(it -> ScriptUtils.deserializeResult(pair.getReader(), it)); + } + + protected SerializationPair keySerializer() { + return serializationContext.getKeySerializationPair(); + } + + /** + * Executes the given action object within a connection that is allocated eagerly and released after {@link Flux} + * termination. + * + * @param return type + * @param action callback object to execute + * @return object returned by the action + */ + private Flux execute(ReactiveRedisCallback 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; + } +} diff --git a/src/main/java/org/springframework/data/redis/core/script/ScriptOperators.java b/src/main/java/org/springframework/data/redis/core/script/ScriptOperators.java new file mode 100644 index 000000000..b389c342d --- /dev/null +++ b/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 The type of keys that may be passed during script execution + * @since 2.0 + */ +public interface ScriptOperators { + + /** + * 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") + */ + Flux execute(RedisScript script, List 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") + */ + Flux execute(RedisScript script, SerializationPair argsSerializer, SerializationPair resultSerializer, + List keys, Object... args); +} diff --git a/src/main/java/org/springframework/data/redis/core/script/ScriptUtils.java b/src/main/java/org/springframework/data/redis/core/script/ScriptUtils.java new file mode 100644 index 000000000..80fca3ea2 --- /dev/null +++ b/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 deserializeResult(RedisSerializer resultSerializer, Object result) { + + if (result instanceof byte[]) { + return resultSerializer == null ? (T) result : resultSerializer.deserialize((byte[]) result); + } + + if (result instanceof List) { + + List 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 deserializeResult(RedisElementReader reader, Object result) { + + if (result instanceof ByteBuffer) { + + return reader == null ? (T) result : reader.read((ByteBuffer) result); + } + + if (result instanceof List) { + + List 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; + } +} diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommandsTests.java new file mode 100644 index 000000000..9fc606718 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommandsTests.java @@ -0,0 +1,182 @@ +/* + * 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 static org.junit.Assert.*; +import static org.junit.Assume.*; +import static org.springframework.data.redis.SpinBarrier.*; + +import io.lettuce.core.ScriptOutputType; +import reactor.test.StepVerifier; + +import java.nio.ByteBuffer; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.Test; +import org.springframework.data.redis.RedisSystemException; +import org.springframework.data.redis.connection.ReactiveRedisClusterConnection; +import org.springframework.data.redis.connection.ReturnType; + +/** + * @author Mark Paluch + */ +public class LettuceReactiveScriptingCommandsTests extends LettuceReactiveCommandsTestsBase { + + @Test // DATAREDIS-683 + public void scriptExistsShouldReturnState() { + + assumeFalse(connection instanceof ReactiveRedisClusterConnection); + + String sha1 = nativeCommands.scriptLoad("return KEYS[1]"); + + StepVerifier.create(connection.scriptingCommands().scriptExists("foo", sha1)) // + .expectNext(false) // + .expectNext(true) // + .verifyComplete(); + } + + @Test // DATAREDIS-683 + public void scriptFlushShouldRemoveScripts() { + + assumeFalse(connection instanceof ReactiveRedisClusterConnection); + + String sha1 = nativeCommands.scriptLoad("return KEYS[1]"); + + StepVerifier.create(connection.scriptingCommands().scriptExists(sha1)) // + .expectNext(true) // + .verifyComplete(); + + StepVerifier.create(connection.scriptingCommands().scriptFlush()) // + .expectNext("OK") // + .verifyComplete(); + + StepVerifier.create(connection.scriptingCommands().scriptExists(sha1)) // + .expectNext(false) // + .verifyComplete(); + } + + @Test // DATAREDIS-683 + public void evalShaShouldReturnKey() { + + assumeFalse(connection instanceof ReactiveRedisClusterConnection); + + String sha1 = nativeCommands.scriptLoad("return KEYS[1]"); + + StepVerifier + .create(connection.scriptingCommands().evalSha(sha1, ReturnType.VALUE, 2, SAME_SLOT_KEY_1_BBUFFER.duplicate(), + SAME_SLOT_KEY_2_BBUFFER.duplicate())) // + .expectNext(SAME_SLOT_KEY_1_BBUFFER) // + .verifyComplete(); + } + + @Test // DATAREDIS-683 + public void evalShaShouldReturnMulti() { + + assumeFalse(connection instanceof ReactiveRedisClusterConnection); + + String sha1 = nativeCommands.scriptLoad("return {KEYS[1],ARGV[1]}"); + + StepVerifier + .create(connection.scriptingCommands().evalSha(sha1, ReturnType.MULTI, 1, SAME_SLOT_KEY_1_BBUFFER.duplicate(), + SAME_SLOT_KEY_2_BBUFFER.duplicate())) // + .expectNext(SAME_SLOT_KEY_1_BBUFFER) // + .expectNext(SAME_SLOT_KEY_2_BBUFFER) // + .verifyComplete(); + } + + @Test // DATAREDIS-683 + public void evalShaShouldFail() { + + assumeFalse(connection instanceof ReactiveRedisClusterConnection); + + StepVerifier + .create(connection.scriptingCommands().evalSha("foo", ReturnType.VALUE, 1, SAME_SLOT_KEY_1_BBUFFER.duplicate())) // + .expectError(RedisSystemException.class) // + .verify(); + } + + @Test // DATAREDIS-683 + public void evalShouldReturnStatus() { + + ByteBuffer script = wrap(String.format("return redis.call('set','%s','ghk')", SAME_SLOT_KEY_1)); + + StepVerifier + .create(connection.scriptingCommands().eval(script, ReturnType.STATUS, 1, SAME_SLOT_KEY_1_BBUFFER.duplicate())) // + .expectNext("OK") // + .verifyComplete(); + } + + @Test // DATAREDIS-683 + public void evalShouldReturnBooleanFalse() { + + ByteBuffer script = wrap("return false"); + + StepVerifier.create(connection.scriptingCommands().eval(script, ReturnType.BOOLEAN, 0)) // + .expectNext(false) // + .verifyComplete(); + } + + @Test // DATAREDIS-683 + public void evalShouldReturnMultiNumbers() { + + ByteBuffer script = wrap("return {1,2}"); + + StepVerifier.create(connection.scriptingCommands().eval(script, ReturnType.MULTI, 0)) // + .expectNext(1L) // + .expectNext(2L) // + .verifyComplete(); + } + + @Test // DATAREDIS-683 + public void evalShouldFailWithScriptError() { + + ByteBuffer script = wrap("return {1,2"); + + StepVerifier.create(connection.scriptingCommands().eval(script, ReturnType.MULTI, 0)) // + .expectError(RedisSystemException.class) // + .verify(); + } + + @Test // DATAREDIS-683 + public void scriptKillShouldKillScripts() throws Exception { + + assumeFalse(connection instanceof ReactiveRedisClusterConnection); + + AtomicBoolean scriptDead = new AtomicBoolean(false); + CountDownLatch sync = new CountDownLatch(1); + Thread th = new Thread(() -> { + try { + sync.countDown(); + nativeCommands.eval("local time=1 while time < 10000000000 do time=time+1 end", ScriptOutputType.BOOLEAN); + } catch (Exception e) { + scriptDead.set(true); + } + }); + th.start(); + sync.await(2, TimeUnit.SECONDS); + Thread.sleep(200); + + StepVerifier.create(connection.scriptingCommands().scriptKill()).expectNext("OK").verifyComplete(); + + assertTrue(waitFor(scriptDead::get, 3000L)); + } + + private static ByteBuffer wrap(String content) { + return ByteBuffer.wrap(content.getBytes()); + } +} diff --git a/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java index 7accbeee8..f60d2b64b 100644 --- a/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateIntegrationTests.java @@ -18,11 +18,13 @@ package org.springframework.data.redis.core; import static org.assertj.core.api.Assertions.*; import static org.junit.Assume.*; +import reactor.core.publisher.Flux; import reactor.test.StepVerifier; import java.time.Duration; import java.time.Instant; import java.util.Collection; +import java.util.Collections; import org.junit.AfterClass; import org.junit.Before; @@ -39,8 +41,11 @@ import org.springframework.data.redis.connection.DataType; import org.springframework.data.redis.connection.ReactiveRedisClusterConnection; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.script.DefaultRedisScript; +import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer; import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; import org.springframework.data.redis.serializer.RedisSerializationContext; +import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair; import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; @@ -160,6 +165,48 @@ public class ReactiveRedisTemplateIntegrationTests { .verify(); } + @Test // DATAREDIS-683 + @SuppressWarnings("unchecked") + public void execute() { + + K key = keyFactory.instance(); + V value = valueFactory.instance(); + + assumeFalse(value instanceof Long); + + StepVerifier.create(redisTemplate.opsForValue().set(key, value)).expectNext(true).verifyComplete(); + + Flux execute = redisTemplate.execute( + new DefaultRedisScript<>("return redis.call('get', KEYS[1])", (Class) value.getClass()), + Collections.singletonList(key)); + + StepVerifier.create(execute).expectNext(value).verifyComplete(); + } + + @Test // DATAREDIS-683 + public void executeWithSerializationPairs() { + + K key = keyFactory.instance(); + V value = valueFactory.instance(); + + SerializationPair json = SerializationPair.fromSerializer(new Jackson2JsonRedisSerializer<>(Person.class)); + SerializationPair string = SerializationPair.fromSerializer(new StringRedisSerializer()); + + assumeFalse(value instanceof Long); + + Person person = new Person("Walter", "White", 51); + StepVerifier.create( + redisTemplate.execute(new DefaultRedisScript<>("return redis.call('set', KEYS[1], ARGV[1])", String.class), + json, string, Collections.singletonList(key), person)) + .expectNext("OK").verifyComplete(); + + Flux execute = redisTemplate.execute( + new DefaultRedisScript<>("return redis.call('get', KEYS[1])", Person.class), json, json, + Collections.singletonList(key)); + + StepVerifier.create(execute).expectNext(person).verifyComplete(); + } + @Test // DATAREDIS-602 public void expire() { diff --git a/src/test/java/org/springframework/data/redis/core/script/DefaultScriptOperatorsUnitTests.java b/src/test/java/org/springframework/data/redis/core/script/DefaultScriptOperatorsUnitTests.java new file mode 100644 index 000000000..ca9604fc7 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/core/script/DefaultScriptOperatorsUnitTests.java @@ -0,0 +1,123 @@ +/* + * 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 static org.mockito.Mockito.*; + +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +import java.nio.ByteBuffer; +import java.util.Collections; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; +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.ReactiveScriptingCommands; +import org.springframework.data.redis.connection.ReturnType; +import org.springframework.data.redis.serializer.RedisSerializationContext; + +/** + * @author Mark Paluch + */ +@RunWith(MockitoJUnitRunner.class) +public class DefaultScriptOperatorsUnitTests { + + private final DefaultRedisScript SCRIPT = new DefaultRedisScript<>("return KEYS[0]", String.class); + + @Mock ReactiveRedisConnectionFactory connectionFactoryMock; + @Mock ReactiveRedisConnection connectionMock; + @Mock ReactiveScriptingCommands scriptingCommandsMock; + + DefaultScriptOperators executor; + + @Before + public void setUp() { + + when(connectionFactoryMock.getReactiveConnection()).thenReturn(connectionMock); + when(connectionMock.scriptingCommands()).thenReturn(scriptingCommandsMock); + + executor = new DefaultScriptOperators<>(connectionFactoryMock, RedisSerializationContext.string()); + } + + @Test // DATAREDIS-683 + public void executeCheckForPresenceOfScriptViaEvalSha1() { + + when(scriptingCommandsMock.evalSha(anyString(), any(ReturnType.class), anyInt())) + .thenReturn(Flux.just(ByteBuffer.wrap("FOO".getBytes()))); + + StepVerifier.create(executor.execute(SCRIPT, Collections.emptyList())).expectNext("FOO").verifyComplete(); + + verify(scriptingCommandsMock).evalSha(anyString(), any(ReturnType.class), anyInt()); + verify(scriptingCommandsMock, never()).eval(any(), any(ReturnType.class), anyInt()); + } + + @Test // DATAREDIS-683 + public void executeShouldUseEvalInCaseNoSha1PresentForGivenScript() { + + when(scriptingCommandsMock.evalSha(anyString(), any(ReturnType.class), anyInt())).thenReturn( + Flux.error(new RedisSystemException("NOSCRIPT No matching script. Please use EVAL.", new Exception()))); + + when(scriptingCommandsMock.eval(any(), any(ReturnType.class), anyInt())) + .thenReturn(Flux.just(ByteBuffer.wrap("FOO".getBytes()))); + + StepVerifier.create(executor.execute(SCRIPT, Collections.emptyList())).expectNext("FOO").verifyComplete(); + + verify(scriptingCommandsMock).evalSha(anyString(), any(ReturnType.class), anyInt()); + verify(scriptingCommandsMock).eval(any(), any(ReturnType.class), anyInt()); + } + + @Test // DATAREDIS-683 + public void executeShouldThrowExceptionInCaseEvalShaFailsWithOtherThanRedisSystemException() { + + when(scriptingCommandsMock.evalSha(anyString(), any(ReturnType.class), anyInt())).thenReturn(Flux + .error(new UnsupportedOperationException("NOSCRIPT No matching script. Please use EVAL.", new Exception()))); + + StepVerifier.create(executor.execute(SCRIPT, Collections.emptyList())) + .expectError(UnsupportedOperationException.class).verify(); + } + + @Test // DATAREDIS-683 + public void releasesConnectionAfterExecution() { + + when(scriptingCommandsMock.evalSha(anyString(), any(ReturnType.class), anyInt())) + .thenReturn(Flux.just(ByteBuffer.wrap("FOO".getBytes()))); + + Flux execute = executor.execute(SCRIPT, Collections.emptyList()); + + verify(connectionMock, never()).close(); + + StepVerifier.create(execute).expectNext("FOO").verifyComplete(); + + verify(connectionMock).close(); + } + + @Test // DATAREDIS-683 + public void releasesConnectionOnError() { + + when(scriptingCommandsMock.evalSha(anyString(), any(ReturnType.class), anyInt())) + .thenReturn(Flux.error(new RuntimeException())); + + StepVerifier.create(executor.execute(SCRIPT, Collections.emptyList())).expectError().verify(); + + verify(connectionMock).close(); + } +}