DATAREDIS-711 - Emit Lua array responses as List.

We now emit array responses from Lua script execution as complete List instead of emitting the individual elements through the Publisher.

Lua scripts may return nil (null) elements that would be not emitted through a Publisher. Skipping null values would garble up the response – and in several cases, the response array is required as List. Emitting the whole List aligns the response to the signatures imposed by generics and aligns the behavior with the imperative API.

Original Pull Request: #282

Author: mp911de

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

@@ -82,7 +82,8 @@ default Mono<Boolean> scriptExists(String scriptSha) {
 	 * Evaluate given {@code script}.
 	 *
 	 * @param script must not be {@literal null}.
-	 * @param returnType must not be {@literal null}.
+	 * @param returnType must not be {@literal null}. Using {@link ReturnType#MULTI} emits a {@link List} as-is instead of
+	 *          emitting the individual elements from the array response.
 	 * @param numKeys
 	 * @param keysAndArgs must not be {@literal null}.
 	 * @return never {@literal null}.
@@ -94,7 +95,8 @@ default Mono<Boolean> scriptExists(String scriptSha) {
 	 * Evaluate given {@code scriptSha}.
 	 *
 	 * @param scriptSha must not be {@literal null}.
-	 * @param returnType must not be {@literal null}.
+	 * @param returnType must not be {@literal null}. Using {@link ReturnType#MULTI} emits a {@link List} as-is instead of
+	 *          emitting the individual elements from the array response.
 	 * @param numKeys
 	 * @param keysAndArgs must not be {@literal null}.
 	 * @return never {@literal null}.

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

@@ -141,8 +141,6 @@ 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);
 			});
 		}

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

@@ -75,9 +75,10 @@ public <T> Flux<T> execute(RedisScript<T> script, List<K> keys, List<?> args) {
 		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, keys, args, serializationContext.getKeySerializationPair().getWriter(),
-				(RedisElementReader<T>) serializationContext.getValueSerializationPair().getReader());
+		SerializationPair<?> serializationPair = serializationContext.getValueSerializationPair();
+
+		return execute(script, keys, args, serializationPair.getWriter(),
+				(RedisElementReader<T>) serializationPair.getReader());
 	}
 
 	/*

src/test/java/org/springframework/data/redis/SettingsUtils.java

@@ -17,8 +17,11 @@
 
 import java.util.Properties;
 
+import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
+
 /**
  * @author Costin Leau
+ * @author Mark Paluch
  */
 public abstract class SettingsUtils {
 	private final static Properties DEFAULTS = new Properties();
@@ -44,4 +47,8 @@ public static String getHost() {
 	public static int getPort() {
 		return Integer.valueOf(SETTINGS.getProperty("port"));
 	}
+
+	public static RedisStandaloneConfiguration standaloneConfiguration() {
+		return new RedisStandaloneConfiguration(getHost(), getPort());
+	}
 }

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommandsTests.java

@@ -86,7 +86,7 @@ public void evalShaShouldReturnKey() {
 				.verifyComplete();
 	}
 
-	@Test // DATAREDIS-683
+	@Test // DATAREDIS-683, DATAREDIS-711
 	public void evalShaShouldReturnMulti() {
 
 		assumeFalse(connection instanceof ReactiveRedisClusterConnection);
@@ -96,8 +96,7 @@ public void evalShaShouldReturnMulti() {
 		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) //
+				.expectNext(Arrays.asList(SAME_SLOT_KEY_1_BBUFFER, SAME_SLOT_KEY_2_BBUFFER)) //
 				.verifyComplete();
 	}
 
@@ -133,14 +132,13 @@ public void evalShouldReturnBooleanFalse() {
 				.verifyComplete();
 	}
 
-	@Test // DATAREDIS-683
+	@Test // DATAREDIS-683, DATAREDIS-711
 	public void evalShouldReturnMultiNumbers() {
 
 		ByteBuffer script = wrap("return {1,2}");
 
 		StepVerifier.create(connection.scriptingCommands().eval(script, ReturnType.MULTI, 0)) //
-				.expectNext(1L) //
-				.expectNext(2L) //
+				.expectNext(Arrays.asList(1L, 2L)) //
 				.verifyComplete();
 	}
 

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceTestClientConfiguration.java

@@ -0,0 +1,46 @@
+/*
+ * 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.resource.ClientResources;
+
+import java.time.Duration;
+
+import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceClientConfigurationBuilder;
+
+/**
+ * Creates a specific client configuration for Lettuce tests.
+ * 
+ * @author Mark Paluch
+ */
+public class LettuceTestClientConfiguration {
+
+	/**
+	 * @return an initialized {@link LettuceClientConfigurationBuilder} with shutdown timeout and {@link ClientResources}.
+	 * @see LettuceTestClientResources#getSharedClientResources()
+	 */
+	public static LettuceClientConfigurationBuilder builder() {
+		return LettuceClientConfiguration.builder().shutdownTimeout(Duration.ZERO)
+				.clientResources(LettuceTestClientResources.getSharedClientResources());
+	}
+
+	/**
+	 * @return a defaulted {@link LettuceClientConfiguration} for testing.
+	 */
+	public static LettuceClientConfiguration create() {
+		return builder().build();
+	}
+}

src/test/java/org/springframework/data/redis/core/script/DefaultReactiveScriptExecutorTests.java

@@ -0,0 +1,232 @@
+/*
+ * 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.assertj.core.api.Assertions.*;
+
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.data.redis.Person;
+import org.springframework.data.redis.SettingsUtils;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettuceTestClientConfiguration;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.serializer.GenericToStringSerializer;
+import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
+import org.springframework.data.redis.serializer.RedisElementReader;
+import org.springframework.data.redis.serializer.RedisElementWriter;
+import org.springframework.data.redis.serializer.RedisSerializationContext;
+import org.springframework.data.redis.serializer.RedisSerializationContext.RedisSerializationContextBuilder;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.scripting.support.StaticScriptSource;
+
+/**
+ * @author Mark Paluch
+ */
+public class DefaultReactiveScriptExecutorTests {
+
+	private static LettuceConnectionFactory connectionFactory;
+	private static StringRedisTemplate stringTemplate;
+	private static ReactiveScriptExecutor<String> stringScriptExecutor;
+
+	@BeforeClass
+	public static void setUp() {
+
+		connectionFactory = new LettuceConnectionFactory(SettingsUtils.standaloneConfiguration(),
+				LettuceTestClientConfiguration.create());
+		connectionFactory.afterPropertiesSet();
+
+		stringTemplate = new StringRedisTemplate(connectionFactory);
+		stringScriptExecutor = new DefaultReactiveScriptExecutor<>(connectionFactory, RedisSerializationContext.string());
+	}
+
+	@AfterClass
+	public static void cleanUp() {
+
+		if (connectionFactory != null) {
+			connectionFactory.destroy();
+		}
+	}
+
+	@Before
+	public void before() {
+
+		RedisConnection connection = connectionFactory.getConnection();
+		connection.flushDb();
+		connection.close();
+	}
+
+	protected RedisConnectionFactory getConnectionFactory() {
+		return connectionFactory;
+	}
+
+	@Test // DATAREDIS-711
+	public void shouldReturnLong() {
+
+		DefaultRedisScript<Long> script = new DefaultRedisScript<>();
+		script.setLocation(new ClassPathResource("org/springframework/data/redis/core/script/increment.lua"));
+		script.setResultType(Long.class);
+
+		StepVerifier.create(stringScriptExecutor.execute(script, Collections.singletonList("mykey"))).verifyComplete();
+
+		stringTemplate.opsForValue().set("mykey", "2");
+
+		StepVerifier.create(stringScriptExecutor.execute(script, Collections.singletonList("mykey"))).expectNext(3L)
+				.verifyComplete();
+	}
+
+	@Test // DATAREDIS-711
+	public void shouldReturnBoolean() {
+
+		RedisSerializationContextBuilder<String, Long> builder = RedisSerializationContext
+				.newSerializationContext(new StringRedisSerializer());
+		builder.value(new GenericToStringSerializer<>(Long.class));
+
+		DefaultRedisScript<Boolean> script = new DefaultRedisScript<>();
+		script.setLocation(new ClassPathResource("org/springframework/data/redis/core/script/cas.lua"));
+		script.setResultType(Boolean.class);
+
+		ReactiveScriptExecutor<String> scriptExecutor = new DefaultReactiveScriptExecutor<>(connectionFactory,
+				builder.build());
+
+		stringTemplate.opsForValue().set("counter", "0");
+
+		StepVerifier.create(scriptExecutor.execute(script, Collections.singletonList("counter"), Arrays.asList(0, 3)))
+				.expectNext(true).verifyComplete();
+
+		StepVerifier.create(scriptExecutor.execute(script, Collections.singletonList("counter"), Arrays.asList(0, 3)))
+				.expectNext(false).verifyComplete();
+	}
+
+	@Test // DATAREDIS-711
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	public void shouldApplyCustomArgsSerializer() {
+
+		DefaultRedisScript<List> script = new DefaultRedisScript<>();
+		script.setLocation(new ClassPathResource("org/springframework/data/redis/core/script/bulkpop.lua"));
+		script.setResultType(List.class);
+
+		stringTemplate.boundListOps("mylist").leftPushAll("a", "b", "c", "d");
+
+		Flux<List<String>> mylist = stringScriptExecutor.execute(script, Collections.singletonList("mylist"),
+				Collections.singletonList(1L), RedisElementWriter.from(new GenericToStringSerializer<>(Long.class)),
+				(RedisElementReader) RedisElementReader.from(new StringRedisSerializer()));
+
+		StepVerifier.create(mylist).expectNext(Collections.singletonList("a")).verifyComplete();
+	}
+
+	@Test // DATAREDIS-711
+	public void testExecuteMixedListResult() {
+
+		DefaultRedisScript<List> script = new DefaultRedisScript<>();
+		script.setLocation(new ClassPathResource("org/springframework/data/redis/core/script/popandlength.lua"));
+		script.setResultType(List.class);
+
+		StepVerifier.create(stringScriptExecutor.execute(script, Collections.singletonList("mylist")))
+				.expectNext(Arrays.asList(null, 0L)).verifyComplete();
+
+		stringTemplate.boundListOps("mylist").leftPushAll("a", "b");
+
+		StepVerifier.create(stringScriptExecutor.execute(script, Collections.singletonList("mylist")))
+				.expectNext(Arrays.asList("a", 1L)).verifyComplete();
+	}
+
+	@Test // DATAREDIS-711
+	public void shouldReturnValueResult() {
+
+		DefaultRedisScript<String> script = new DefaultRedisScript<>();
+		script.setScriptText("return redis.call('GET',KEYS[1])");
+		script.setResultType(String.class);
+
+		stringTemplate.opsForValue().set("foo", "bar");
+
+		Flux<String> foo = stringScriptExecutor.execute(script, Collections.singletonList("foo"));
+
+		StepVerifier.create(foo).expectNext("bar").expectNext();
+	}
+
+	@Test // DATAREDIS-711
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	public void shouldReturnStatusValue() {
+
+		DefaultRedisScript script = new DefaultRedisScript();
+		script.setScriptText("return redis.call('SET',KEYS[1], ARGV[1])");
+
+		RedisSerializationContextBuilder<String, Long> builder = RedisSerializationContext
+				.newSerializationContext(new StringRedisSerializer());
+		builder.value(new GenericToStringSerializer<>(Long.class));
+
+		ReactiveScriptExecutor<String> scriptExecutor = new DefaultReactiveScriptExecutor<>(connectionFactory,
+				builder.build());
+
+		StepVerifier.create(scriptExecutor.execute(script, Collections.singletonList("foo"), Collections.singletonList(3L)))
+				.expectNext("OK").verifyComplete();
+
+		assertThat(stringTemplate.opsForValue().get("foo")).isEqualTo("3");
+	}
+
+	@Test // DATAREDIS-711
+	public void shouldApplyCustomResultSerializer() {
+
+		Jackson2JsonRedisSerializer<Person> personSerializer = new Jackson2JsonRedisSerializer<>(Person.class);
+
+		RedisTemplate<String, Person> template = new RedisTemplate<>();
+		template.setKeySerializer(new StringRedisSerializer());
+		template.setValueSerializer(personSerializer);
+		template.setConnectionFactory(getConnectionFactory());
+		template.afterPropertiesSet();
+
+		DefaultRedisScript<String> script = new DefaultRedisScript<>();
+		script.setScriptSource(new StaticScriptSource("redis.call('SET',KEYS[1], ARGV[1])\nreturn 'FOO'"));
+		script.setResultType(String.class);
+
+		Person joe = new Person("Joe", "Schmoe", 23);
+		Flux<String> result = stringScriptExecutor.execute(script, Collections.singletonList("bar"),
+				Collections.singletonList(joe), RedisElementWriter.from(personSerializer),
+				RedisElementReader.from(new StringRedisSerializer()));
+
+		StepVerifier.create(result).expectNext("FOO").verifyComplete();
+
+		assertThat(template.opsForValue().get("bar")).isEqualTo(joe);
+	}
+
+	@Test // DATAREDIS-711
+	public void testExecuteCachedNullKeys() {
+
+		DefaultRedisScript<String> script = new DefaultRedisScript<>();
+		script.setScriptText("return 'HELLO'");
+		script.setResultType(String.class);
+
+		// Execute script twice, second time should be from cache
+		StepVerifier.create(stringScriptExecutor.execute(script, Collections.emptyList())).expectNext("HELLO")
+				.verifyComplete();
+		StepVerifier.create(stringScriptExecutor.execute(script, Collections.emptyList())).expectNext("HELLO")
+				.verifyComplete();
+	}
+}