getConnection() {
- return connection.getConnection();
- }
-
- protected DataAccessException convertLettuceAccessException(Exception ex) {
- return connection.convertLettuceAccessException(ex);
+ connection.invoke().just(RedisHLLAsyncCommands::pfmerge, destinationKey, sourceKeys);
}
}
diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceInvoker.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceInvoker.java
new file mode 100644
index 000000000..83af96084
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceInvoker.java
@@ -0,0 +1,665 @@
+/*
+ * Copyright 2021 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
+ *
+ * https://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.RedisFuture;
+import io.lettuce.core.cluster.api.async.RedisClusterAsyncCommands;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Supplier;
+
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Utility for functional invocation of {@link RedisClusterAsyncCommands Lettuce methods}. Typically used to express the
+ * method call as method reference and passing method arguments through one of the {@code just} or {@code from} methods.
+ *
+ * {@code just} methods record the method call and evaluate the method result immediately. {@code from} methods allows
+ * composing a functional pipeline to transform the result using a {@link Converter}.
+ *
+ * Usage example:
+ *
+ *
+ * LettuceInvoker invoker = …;
+ *
+ * Long result = invoker.just(RedisGeoAsyncCommands::geoadd, key, point.getX(), point.getY(), member);
+ *
+ * List<byte[]> result = invoker.fromMany(RedisGeoAsyncCommands::geohash, key, members)
+ * .toList(it -> it.getValueOrElse(null));
+ *
+ *
+ * The actual translation from {@link RedisFuture} is delegated to {@link Synchronizer} which can either await
+ * completion or record the future along {@link Converter} for further processing.
+ *
+ * @author Mark Paluch
+ * @since 2.5
+ */
+class LettuceInvoker {
+
+ private final RedisClusterAsyncCommands connection;
+ private final Synchronizer synchronizer;
+
+ LettuceInvoker(RedisClusterAsyncCommands connection, Synchronizer synchronizer) {
+ this.connection = connection;
+ this.synchronizer = synchronizer;
+ }
+
+ /**
+ * Returns a {@link Converter} that always returns its input argument.
+ *
+ * @param the type of the input and output objects to the function
+ * @return a function that always returns its input argument
+ */
+ static Converter identityConverter() {
+ return t -> t;
+ }
+
+ /**
+ * Invoke the {@link ConnectionFunction0} and return its result.
+ *
+ * @param function must not be {@literal null}.
+ */
+ @Nullable
+ R just(ConnectionFunction0 function) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return synchronizer.invoke(() -> function.apply(connection), identityConverter(), () -> null);
+ }
+
+ /**
+ * Invoke the {@link ConnectionFunction1} and return its result.
+ *
+ * @param function must not be {@literal null}.
+ */
+ @Nullable
+ R just(ConnectionFunction1 function, T1 t1) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return synchronizer.invoke(() -> function.apply(connection, t1));
+ }
+
+ /**
+ * Invoke the {@link ConnectionFunction2} and return its result.
+ *
+ * @param function must not be {@literal null}.
+ */
+ @Nullable
+ R just(ConnectionFunction2 function, T1 t1, T2 t2) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return synchronizer.invoke(() -> function.apply(connection, t1, t2));
+ }
+
+ /**
+ * Invoke the {@link ConnectionFunction3} and return its result.
+ *
+ * @param function must not be {@literal null}.
+ */
+ @Nullable
+ R just(ConnectionFunction3 function, T1 t1, T2 t2, T3 t3) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return synchronizer.invoke(() -> function.apply(connection, t1, t2, t3));
+ }
+
+ /**
+ * Invoke the {@link ConnectionFunction4} and return its result.
+ *
+ * @param function must not be {@literal null}.
+ */
+ @Nullable
+ R just(ConnectionFunction4 function, T1 t1, T2 t2, T3 t3, T4 t4) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return synchronizer.invoke(() -> function.apply(connection, t1, t2, t3, t4));
+ }
+
+ /**
+ * Invoke the {@link ConnectionFunction5} and return its result.
+ *
+ * @param function must not be {@literal null}.
+ */
+ @Nullable
+ R just(ConnectionFunction5 function, T1 t1, T2 t2, T3 t3, T4 t4,
+ T5 t5) {
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return synchronizer.invoke(() -> function.apply(connection, t1, t2, t3, t4, t5));
+ }
+
+ /**
+ * Compose a invocation pipeline from the {@link ConnectionFunction0} and return a {@link SingleInvocationSpec} for
+ * further composition.
+ *
+ * @param function must not be {@literal null}.
+ */
+ SingleInvocationSpec from(ConnectionFunction0 function) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return new DefaultSingleInvocationSpec<>(() -> function.apply(connection), synchronizer);
+ }
+
+ /**
+ * Compose a invocation pipeline from the {@link ConnectionFunction1} and return a {@link SingleInvocationSpec} for
+ * further composition.
+ *
+ * @param function must not be {@literal null}.
+ */
+ SingleInvocationSpec from(ConnectionFunction1 function, T1 t1) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return from(it -> function.apply(it, t1));
+ }
+
+ /**
+ * Compose a invocation pipeline from the {@link ConnectionFunction2} and return a {@link SingleInvocationSpec} for
+ * further composition.
+ *
+ * @param function must not be {@literal null}.
+ */
+ SingleInvocationSpec from(ConnectionFunction2 function, T1 t1, T2 t2) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return from(it -> function.apply(it, t1, t2));
+ }
+
+ /**
+ * Compose a invocation pipeline from the {@link ConnectionFunction3} and return a {@link SingleInvocationSpec} for
+ * further composition.
+ *
+ * @param function must not be {@literal null}.
+ */
+ SingleInvocationSpec from(ConnectionFunction3 function, T1 t1, T2 t2, T3 t3) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return from(it -> function.apply(it, t1, t2, t3));
+ }
+
+ /**
+ * Compose a invocation pipeline from the {@link ConnectionFunction4} and return a {@link SingleInvocationSpec} for
+ * further composition.
+ *
+ * @param function must not be {@literal null}.
+ */
+ SingleInvocationSpec from(ConnectionFunction4 function, T1 t1, T2 t2, T3 t3,
+ T4 t4) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return from(it -> function.apply(it, t1, t2, t3, t4));
+ }
+
+ /**
+ * Compose a invocation pipeline from the {@link ConnectionFunction5} and return a {@link SingleInvocationSpec} for
+ * further composition.
+ *
+ * @param function must not be {@literal null}.
+ */
+ SingleInvocationSpec from(ConnectionFunction5 function, T1 t1,
+ T2 t2, T3 t3, T4 t4, T5 t5) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return from(it -> function.apply(it, t1, t2, t3, t4, t5));
+ }
+
+ /**
+ * Compose a invocation pipeline from the {@link ConnectionFunction0} that returns a {@link Collection}-like result
+ * and return a {@link ManyInvocationSpec} for further composition.
+ *
+ * @param function must not be {@literal null}.
+ */
+ , E> ManyInvocationSpec fromMany(ConnectionFunction0 function) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return new DefaultManyInvocationSpec<>(() -> function.apply(connection), synchronizer);
+ }
+
+ /**
+ * Compose a invocation pipeline from the {@link ConnectionFunction1} that returns a {@link Collection}-like result
+ * and return a {@link ManyInvocationSpec} for further composition.
+ *
+ * @param function must not be {@literal null}.
+ */
+ , E, T1> ManyInvocationSpec fromMany(ConnectionFunction1 function, T1 t1) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return fromMany(it -> function.apply(it, t1));
+ }
+
+ /**
+ * Compose a invocation pipeline from the {@link ConnectionFunction2} that returns a {@link Collection}-like result
+ * and return a {@link ManyInvocationSpec} for further composition.
+ *
+ * @param function must not be {@literal null}.
+ */
+ , E, T1, T2> ManyInvocationSpec fromMany(ConnectionFunction2 function, T1 t1,
+ T2 t2) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return fromMany(it -> function.apply(it, t1, t2));
+ }
+
+ /**
+ * Compose a invocation pipeline from the {@link ConnectionFunction3} that returns a {@link Collection}-like result
+ * and return a {@link ManyInvocationSpec} for further composition.
+ *
+ * @param function must not be {@literal null}.
+ */
+ , E, T1, T2, T3> ManyInvocationSpec fromMany(ConnectionFunction3 function,
+ T1 t1, T2 t2, T3 t3) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return fromMany(it -> function.apply(it, t1, t2, t3));
+ }
+
+ /**
+ * Compose a invocation pipeline from the {@link ConnectionFunction4} that returns a {@link Collection}-like result
+ * and return a {@link ManyInvocationSpec} for further composition.
+ *
+ * @param function must not be {@literal null}.
+ */
+ , E, T1, T2, T3, T4> ManyInvocationSpec fromMany(
+ ConnectionFunction4 function, T1 t1, T2 t2, T3 t3, T4 t4) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return fromMany(it -> function.apply(it, t1, t2, t3, t4));
+ }
+
+ /**
+ * Compose a invocation pipeline from the {@link ConnectionFunction5} that returns a {@link Collection}-like result
+ * and return a {@link ManyInvocationSpec} for further composition.
+ *
+ * @param function must not be {@literal null}.
+ */
+ , E, T1, T2, T3, T4, T5> ManyInvocationSpec fromMany(
+ ConnectionFunction5 function, T1 t1, T2 t2, T3 t3, T4 t4, T5 t5) {
+
+ Assert.notNull(function, "ConnectionFunction must not be null");
+
+ return fromMany(it -> function.apply(it, t1, t2, t3, t4, t5));
+ }
+
+ /**
+ * Represents an element in the invocation pipleline allowing consuming the result by applying a {@link Converter}.
+ *
+ * @param
+ */
+ public interface SingleInvocationSpec {
+
+ /**
+ * Materialize the pipeline by invoking the {@code ConnectionFunction} and returning the result after applying
+ * {@link Converter}.
+ *
+ * @param converter must not be {@literal null}.
+ * @param target type.
+ * @return the converted result, can be {@literal null}.
+ */
+ @Nullable
+ T get(Converter converter);
+
+ /**
+ * Materialize the pipeline by invoking the {@code ConnectionFunction} and returning the result after applying
+ * {@link Converter}.
+ *
+ * @param converter must not be {@literal null}.
+ * @param nullDefault must not be {@literal null}.
+ * @param target type.
+ * @return the converted result, can be {@literal null}.
+ */
+ @Nullable
+ T getOrElse(Converter converter, Supplier nullDefault);
+ }
+
+ /**
+ * Represents an element in the invocation pipleline for methods returning {@link Collection}-like results allowing
+ * consuming the result by applying a {@link Converter}.
+ *
+ * @param
+ */
+ public interface ManyInvocationSpec {
+
+ /**
+ * Materialize the pipeline by invoking the {@code ConnectionFunction} and returning the result.
+ *
+ * @return the result as {@link List}.
+ */
+ default List toList() {
+ return toList(identityConverter());
+ }
+
+ /**
+ * Materialize the pipeline by invoking the {@code ConnectionFunction} and returning the result after applying
+ * {@link Converter}.
+ *
+ * @param converter must not be {@literal null}.
+ * @param target type.
+ * @return the converted {@link List}.
+ */
+ List toList(Converter converter);
+
+ /**
+ * Materialize the pipeline by invoking the {@code ConnectionFunction} and returning the result.
+ *
+ * @return the result as {@link Set}.
+ */
+ default Set toSet() {
+ return toSet(identityConverter());
+ }
+
+ /**
+ * Materialize the pipeline by invoking the {@code ConnectionFunction} and returning the result after applying
+ * {@link Converter}.
+ *
+ * @param converter must not be {@literal null}.
+ * @param target type.
+ * @return the converted {@link Set}.
+ */
+ Set toSet(Converter converter);
+ }
+
+ /**
+ * A function accepting {@link RedisClusterAsyncCommands} with 0 arguments.
+ *
+ * @param
+ */
+ @FunctionalInterface
+ public interface ConnectionFunction0 {
+
+ /**
+ * Apply this function to the arguments and return a {@link RedisFuture}.
+ */
+ RedisFuture apply(RedisClusterAsyncCommands connection);
+ }
+
+ /**
+ * A function accepting {@link RedisClusterAsyncCommands} with 1 argument.
+ *
+ * @param
+ * @param
+ */
+ @FunctionalInterface
+ public interface ConnectionFunction1 {
+
+ /**
+ * Apply this function to the arguments and return a {@link RedisFuture}.
+ */
+ RedisFuture apply(RedisClusterAsyncCommands connection, T1 t1);
+ }
+
+ /**
+ * A function accepting {@link RedisClusterAsyncCommands} with 2 arguments.
+ *
+ * @param
+ * @param
+ * @param
+ */
+ @FunctionalInterface
+ public interface ConnectionFunction2 {
+
+ /**
+ * Apply this function to the arguments and return a {@link RedisFuture}.
+ */
+ RedisFuture apply(RedisClusterAsyncCommands connection, T1 t1, T2 t2);
+ }
+
+ /**
+ * A function accepting {@link RedisClusterAsyncCommands} with 3 arguments.
+ *
+ * @param
+ * @param
+ * @param
+ * @param
+ */
+ @FunctionalInterface
+ public interface ConnectionFunction3 {
+
+ /**
+ * Apply this function to the arguments and return a {@link RedisFuture}.
+ */
+ RedisFuture apply(RedisClusterAsyncCommands connection, T1 t1, T2 t2, T3 t3);
+ }
+
+ /**
+ * A function accepting {@link RedisClusterAsyncCommands} with 4 arguments.
+ *
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ */
+ @FunctionalInterface
+ public interface ConnectionFunction4 {
+
+ /**
+ * Apply this function to the arguments and return a {@link RedisFuture}.
+ */
+ RedisFuture apply(RedisClusterAsyncCommands connection, T1 t1, T2 t2, T3 t3, T4 t4);
+ }
+
+ /**
+ * A function accepting {@link RedisClusterAsyncCommands} with 5 arguments.
+ *
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ */
+ @FunctionalInterface
+ public interface ConnectionFunction5 {
+
+ /**
+ * Apply this function to the arguments and return a {@link RedisFuture}.
+ */
+ RedisFuture apply(RedisClusterAsyncCommands connection, T1 t1, T2 t2, T3 t3, T4 t4, T5 t5);
+ }
+
+ /**
+ * A function accepting {@link RedisClusterAsyncCommands} with 6 arguments.
+ *
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ */
+ @FunctionalInterface
+ public interface ConnectionFunction6 {
+
+ /**
+ * Apply this function to the arguments and return a {@link RedisFuture}.
+ */
+ RedisFuture apply(RedisClusterAsyncCommands connection, T1 t1, T2 t2, T3 t3, T4 t4, T5 t5,
+ T6 t6);
+ }
+
+ /**
+ * A function accepting {@link RedisClusterAsyncCommands} with 7 arguments.
+ *
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ */
+ @FunctionalInterface
+ public interface ConnectionFunction7 {
+
+ /**
+ * Apply this function to the arguments and return a {@link RedisFuture}.
+ */
+ RedisFuture apply(RedisClusterAsyncCommands connection, T1 t1, T2 t2, T3 t3, T4 t4, T5 t5, T6 t6,
+ T7 t7);
+ }
+
+ /**
+ * A function accepting {@link RedisClusterAsyncCommands} with 8 arguments.
+ *
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ * @param
+ */
+ @FunctionalInterface
+ public interface ConnectionFunction8 {
+
+ /**
+ * Apply this function to the arguments and return a {@link RedisFuture}.
+ */
+ RedisFuture apply(RedisClusterAsyncCommands connection, T1 t1, T2 t2, T3 t3, T4 t4, T5 t5, T6 t6,
+ T7 t7, T8 t8);
+ }
+
+ static class DefaultSingleInvocationSpec implements SingleInvocationSpec {
+
+ private final Supplier> parent;
+ private final Synchronizer synchronizer;
+
+ public DefaultSingleInvocationSpec(Supplier> parent, Synchronizer synchronizer) {
+ this.parent = parent;
+ this.synchronizer = synchronizer;
+ }
+
+ @Override
+ public T get(Converter converter) {
+
+ Assert.notNull(converter, "Converter must not be null");
+
+ return synchronizer.invoke(parent, converter, () -> null);
+ }
+
+ @Nullable
+ @Override
+ public T getOrElse(Converter converter, Supplier nullDefault) {
+
+ Assert.notNull(converter, "Converter must not be null");
+
+ return synchronizer.invoke(parent, converter, nullDefault);
+ }
+ }
+
+ static class DefaultManyInvocationSpec implements ManyInvocationSpec {
+
+ private final Supplier>> parent;
+ private final Synchronizer synchronizer;
+
+ public DefaultManyInvocationSpec(Supplier>> parent, Synchronizer synchronizer) {
+ this.parent = (Supplier) parent;
+ this.synchronizer = synchronizer;
+ }
+
+ @Override
+ public List