diff --git a/pom.xml b/pom.xml
index 82f6399a9..b22b17954 100644
--- a/pom.xml
+++ b/pom.xml
@@ -90,6 +90,14 @@
true
+
+
+ io.projectreactor
+ reactor-core
+ ${reactor}
+ true
+
+
@@ -120,6 +128,11 @@
true
+
+ io.reactivex
+ rxjava
+ 1.1.1
+
@@ -206,7 +219,6 @@
**/*Tests.java
**/*Test.java
- false
true
${basedir}/src/test/resources/trusted.keystore
@@ -232,23 +244,6 @@
-
-
- release
-
-
-
-
- org.jfrog.buildinfo
- artifactory-maven-plugin
- false
-
-
-
-
-
-
-
spring-libs-snapshot
diff --git a/src/main/java/org/springframework/data/redis/connection/ClusterSlotHashUtil.java b/src/main/java/org/springframework/data/redis/connection/ClusterSlotHashUtil.java
index 2ddc22412..cd8199daf 100644
--- a/src/main/java/org/springframework/data/redis/connection/ClusterSlotHashUtil.java
+++ b/src/main/java/org/springframework/data/redis/connection/ClusterSlotHashUtil.java
@@ -15,6 +15,10 @@
*/
package org.springframework.data.redis.connection;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collection;
+
import org.springframework.util.Assert;
/**
@@ -52,6 +56,33 @@ public final class ClusterSlotHashUtil {
}
+ /**
+ * @param keys must not be {@literal null}.
+ * @return
+ * @since 2.0
+ */
+ public static boolean isSameSlotForAllKeys(Collection keys) {
+
+ Assert.notNull(keys, "Keys must not be null!");
+
+ if (keys.size() <= 1) {
+ return true;
+ }
+
+ return isSameSlotForAllKeys((byte[][]) keys.stream().map(ByteBuffer::array).toArray(size -> new byte[size][]));
+ }
+
+ /**
+ * @param keys must not be {@literal null}.
+ * @return
+ * @since 2.0
+ */
+ public static boolean isSameSlotForAllKeys(ByteBuffer... keys) {
+
+ Assert.notNull(keys, "Keys must not be null!");
+ return isSameSlotForAllKeys(Arrays.asList(keys));
+ }
+
/**
* @param keys must not be {@literal null}.
* @return
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveClusterGeoCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterGeoCommands.java
new file mode 100644
index 000000000..b30b44940
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterGeoCommands.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright 2016. 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 Christoph Strobl
+ * @since 2.0
+ */
+public interface ReactiveClusterGeoCommands extends ReactiveGeoCommands {
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveClusterHashCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterHashCommands.java
new file mode 100644
index 000000000..2ee203a94
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterHashCommands.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright 2016. 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 Christoph Strobl
+ * @since 2.0
+ */
+public interface ReactiveClusterHashCommands extends ReactiveHashCommands {
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveClusterHyperLogLogCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterHyperLogLogCommands.java
new file mode 100644
index 000000000..290bf742c
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterHyperLogLogCommands.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright 2016. 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 Christoph Strobl
+ * @since 2.0
+ */
+public interface ReactiveClusterHyperLogLogCommands extends ReactiveHyperLogLogCommands {
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveClusterKeyCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterKeyCommands.java
new file mode 100644
index 000000000..b933ab411
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterKeyCommands.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright 2016. 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.
+ */
+
+/*
+ * Copyright 2016. 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 java.nio.ByteBuffer;
+import java.util.List;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * @author Christoph Strobl
+ * @since 2.0
+ */
+public interface ReactiveClusterKeyCommands extends ReactiveKeyCommands {
+
+ /**
+ * Retrieve all {@literal keys} for a given {@literal pattern} from {@link RedisNode}.
+ *
+ * @param node must not be {@literal null}.
+ * @param pattern must not be {@literal null}.
+ * @return
+ */
+ Mono> keys(RedisClusterNode node, ByteBuffer pattern);
+
+ /**
+ * Retrieve a random {@literal key} from {@link RedisNode}.
+ *
+ * @param node must not be {@literal null}.
+ * @return
+ */
+ Mono randomKey(RedisClusterNode node);
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveClusterListCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterListCommands.java
new file mode 100644
index 000000000..f5dbd4848
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterListCommands.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright 2016. 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 Christoph Strobl
+ * @since 2.0
+ */
+public interface ReactiveClusterListCommands extends ReactiveListCommands {
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveClusterNumberCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterNumberCommands.java
new file mode 100644
index 000000000..e923a72d1
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterNumberCommands.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright 2016. 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 Christoph Strobl
+ * @since 2.0
+ */
+public interface ReactiveClusterNumberCommands extends ReactiveNumberCommands {
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveClusterSetCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterSetCommands.java
new file mode 100644
index 000000000..9aa35dd16
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterSetCommands.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright 2016. 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 Christoph Strobl
+ * @since 2.0
+ */
+public interface ReactiveClusterSetCommands extends ReactiveSetCommands {
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveClusterStringCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterStringCommands.java
new file mode 100644
index 000000000..a36b119c3
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterStringCommands.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright 2016. 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 Christoph Strobl
+ * @since 2.0
+ */
+public interface ReactiveClusterStringCommands extends ReactiveStringCommands {
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveClusterZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterZSetCommands.java
new file mode 100644
index 000000000..f3f4e147f
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ReactiveClusterZSetCommands.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright 2016. 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 Christoph Strobl
+ * @since 2.0
+ */
+public interface ReactiveClusterZSetCommands extends ReactiveZSetCommands {
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveGeoCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveGeoCommands.java
new file mode 100644
index 000000000..2e67557dc
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ReactiveGeoCommands.java
@@ -0,0 +1,771 @@
+/*
+ * Copyright 2016 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 java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.reactivestreams.Publisher;
+import org.springframework.data.domain.Sort.Direction;
+import org.springframework.data.geo.Circle;
+import org.springframework.data.geo.Distance;
+import org.springframework.data.geo.GeoResults;
+import org.springframework.data.geo.Metric;
+import org.springframework.data.geo.Point;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiValueResponse;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
+import org.springframework.data.redis.connection.RedisGeoCommands.DistanceUnit;
+import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
+import org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusCommandArgs;
+import org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusCommandArgs.Flag;
+import org.springframework.util.Assert;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author Christoph Strobl
+ * @since 2.0
+ */
+public interface ReactiveGeoCommands {
+
+ /**
+ * @author Christoph Strobl
+ */
+ class GeoAddCommand extends KeyCommand {
+
+ private final List> geoLocations;
+
+ public GeoAddCommand(ByteBuffer key, List> geoLocations) {
+
+ super(key);
+ this.geoLocations = geoLocations;
+ }
+
+ public static GeoAddCommand location(GeoLocation geoLocation) {
+ return new GeoAddCommand(null, Collections.singletonList(geoLocation));
+ }
+
+ public static GeoAddCommand locations(List> geoLocations) {
+ return new GeoAddCommand(null, new ArrayList<>(geoLocations));
+ }
+
+ public GeoAddCommand to(ByteBuffer key) {
+ return new GeoAddCommand(key, geoLocations);
+ }
+
+ public List> getGeoLocations() {
+ return geoLocations;
+ }
+
+ }
+
+ /**
+ * Add {@link Point} with given {@literal member} to {@literal key}.
+ *
+ * @param key must not be {@literal null}.
+ * @param point must not be {@literal null}.
+ * @param member must not be {@literal null}.
+ * @return
+ */
+ default Mono geoAdd(ByteBuffer key, Point point, ByteBuffer member) {
+
+ Assert.notNull(key, "key must not be null");
+ Assert.notNull(point, "point must not be null");
+ Assert.notNull(member, "member must not be null");
+
+ return geoAdd(key, new GeoLocation<>(member, point));
+ }
+
+ /**
+ * Add {@link GeoLocation} to {@literal key}.
+ *
+ * @param key must not be {@literal null}.
+ * @param location must not be {@literal null}.
+ * @return
+ */
+ default Mono geoAdd(ByteBuffer key, GeoLocation location) {
+
+ Assert.notNull(key, "key must not be null");
+ Assert.notNull(location, "location must not be null");
+
+ return geoAdd(key, Collections.singletonList(location));
+ }
+
+ /**
+ * Add {@link GeoLocation} to {@literal key}.
+ *
+ * @param key must not be {@literal null}.
+ * @param locations must not be {@literal null}.
+ * @return
+ */
+ default Mono geoAdd(ByteBuffer key, List> locations) {
+
+ Assert.notNull(key, "key must not be null");
+ Assert.notNull(locations, "locations must not be null");
+
+ return geoAdd(Mono.just(GeoAddCommand.locations(locations).to(key))).next().map(NumericResponse::getOutput);
+ }
+
+ /**
+ * Add {@link GeoLocation}s to {@literal key}.
+ *
+ * @param commands must not be {@literal null}.
+ * @return
+ */
+ Flux> geoAdd(Publisher commands);
+
+ /**
+ * @author Christoph Strobl
+ */
+ class GeoDistCommand extends KeyCommand {
+
+ private final ByteBuffer from;
+ private final ByteBuffer to;
+ private final Metric metric;
+
+ private GeoDistCommand(ByteBuffer key, ByteBuffer from, ByteBuffer to, Metric metric) {
+ super(key);
+ this.from = from;
+ this.to = to;
+ this.metric = metric;
+ }
+
+ static GeoDistCommand units(Metric unit) {
+ return new GeoDistCommand(null, null, null, unit);
+ }
+
+ public static GeoDistCommand meters() {
+ return units(DistanceUnit.METERS);
+ }
+
+ public static GeoDistCommand kiometers() {
+ return units(DistanceUnit.KILOMETERS);
+ }
+
+ public static GeoDistCommand miles() {
+ return units(DistanceUnit.MILES);
+ }
+
+ public static GeoDistCommand feet() {
+ return units(DistanceUnit.FEET);
+ }
+
+ public GeoDistCommand between(ByteBuffer from) {
+ return new GeoDistCommand(getKey(), from, to, metric);
+ }
+
+ public GeoDistCommand and(ByteBuffer to) {
+ return new GeoDistCommand(getKey(), from, to, metric);
+ }
+
+ public GeoDistCommand forKey(ByteBuffer key) {
+ return new GeoDistCommand(key, from, to, metric);
+ }
+
+ public ByteBuffer getFrom() {
+ return from;
+ }
+
+ public ByteBuffer getTo() {
+ return to;
+ }
+
+ public Optional getMetric() {
+ return Optional.ofNullable(metric);
+ }
+ }
+
+ /**
+ * Get the {@link Distance} between {@literal from} and {@literal to}.
+ *
+ * @param key must not be {@literal null}.
+ * @param from must not be {@literal null}.
+ * @param to must not be {@literal null}.
+ * @return
+ */
+ default Mono geoDist(ByteBuffer key, ByteBuffer from, ByteBuffer to) {
+ return geoDist(key, from, to, null);
+ }
+
+ /**
+ * Get the {@link Distance} between {@literal from} and {@literal to}.
+ *
+ * @param key must not be {@literal null}.
+ * @param from must not be {@literal null}.
+ * @param to must not be {@literal null}.
+ * @param metric can be {@literal null} and defaults to {@link DistanceUnit#METERS}.
+ * @return
+ */
+ default Mono geoDist(ByteBuffer key, ByteBuffer from, ByteBuffer to, Metric metric) {
+
+ Assert.notNull(key, "key must not be null");
+ Assert.notNull(from, "from must not be null");
+ Assert.notNull(to, "to must not be null");
+
+ return geoDist(Mono.just(GeoDistCommand.units(metric).between(from).and(to).forKey(key))).next()
+ .map(CommandResponse::getOutput);
+ }
+
+ /**
+ * Get the {@link Distance} between {@literal from} and {@literal to}.
+ *
+ * @param commands must not be {@literal null}.
+ * @return
+ */
+ Flux> geoDist(Publisher commands);
+
+ /**
+ * @author Christoph Strobl
+ */
+ class GeoHashCommand extends KeyCommand {
+
+ private final List members;
+
+ private GeoHashCommand(ByteBuffer key, List members) {
+
+ super(key);
+ this.members = members;
+ }
+
+ public static GeoHashCommand member(ByteBuffer member) {
+ return new GeoHashCommand(null, Collections.singletonList(member));
+ }
+
+ public static GeoHashCommand members(List members) {
+ return new GeoHashCommand(null, new ArrayList<>(members));
+ }
+
+ public GeoHashCommand of(ByteBuffer key) {
+ return new GeoHashCommand(key, members);
+ }
+
+ public List getMembers() {
+ return members;
+ }
+ }
+
+ /**
+ * Get geohash representation of the position for the one {@literal member}.
+ *
+ * @param key must not be {@literal null}.
+ * @param member must not be {@literal null}.
+ * @return
+ */
+ default Mono geoHash(ByteBuffer key, ByteBuffer member) {
+
+ Assert.notNull(member, "member must not be null");
+
+ return geoHash(key, Collections.singletonList(member)).map(vals -> vals.isEmpty() ? null : vals.iterator().next());
+ }
+
+ /**
+ * Get geohash representation of the position for one or more {@literal member}s.
+ *
+ * @param key must not be {@literal null}.
+ * @param members must not be {@literal null}.
+ * @return
+ */
+ default Mono> geoHash(ByteBuffer key, List members) {
+
+ Assert.notNull(key, "key must not be null");
+ Assert.notNull(members, "members must not be null");
+
+ return geoHash(Mono.just(GeoHashCommand.members(members).of(key))).next().map(MultiValueResponse::getOutput);
+ }
+
+ /**
+ * Get geohash representation of the position for one or more {@literal member}s.
+ *
+ * @param commands must not be {@literal null}.
+ * @return
+ */
+ Flux> geoHash(Publisher commands);
+
+ /**
+ * @author Christoph Strobl
+ */
+ class GeoPosCommand extends KeyCommand {
+
+ private final List members;
+
+ private GeoPosCommand(ByteBuffer key, List members) {
+
+ super(key);
+ this.members = members;
+ }
+
+ public static GeoPosCommand member(ByteBuffer member) {
+ return new GeoPosCommand(null, Collections.singletonList(member));
+ }
+
+ public static GeoPosCommand members(List members) {
+ return new GeoPosCommand(null, new ArrayList<>(members));
+ }
+
+ public GeoPosCommand of(ByteBuffer key) {
+ return new GeoPosCommand(key, members);
+ }
+
+ public List getMembers() {
+ return members;
+ }
+ }
+
+ /**
+ * Get the {@link Point} representation of positions for the {@literal member}s.
+ *
+ * @param key must not be {@literal null}.
+ * @param member must not be {@literal null}.
+ * @return
+ */
+ default Mono geoPos(ByteBuffer key, ByteBuffer member) {
+
+ Assert.notNull(member, "member must not be null");
+
+ return geoPos(key, Collections.singletonList(member)).map(vals -> vals.isEmpty() ? null : vals.iterator().next());
+ }
+
+ /**
+ * Get the {@link Point} representation of positions for one or more {@literal member}s.
+ *
+ * @param key must not be {@literal null}.
+ * @param members must not be {@literal null}.
+ * @return
+ */
+ default Mono> geoPos(ByteBuffer key, List members) {
+
+ Assert.notNull(key, "key must not be null");
+ Assert.notNull(members, "members must not be null");
+
+ return geoPos(Mono.just(GeoPosCommand.members(members).of(key))).next().map(MultiValueResponse::getOutput);
+ }
+
+ /**
+ * Get the {@link Point} representation of positions for one or more {@literal member}s.
+ *
+ * @param commands must not be {@literal null}.
+ * @return
+ */
+ Flux> geoPos(Publisher commands);
+
+ /**
+ * @author Christoph Strobl
+ */
+ class GeoRadiusCommand extends KeyCommand {
+
+ private final Distance distance;
+ private final Point point;
+ private final GeoRadiusCommandArgs args;
+ private final ByteBuffer store;
+ private final ByteBuffer storeDist;
+
+ private GeoRadiusCommand(ByteBuffer key, Point point, Distance distance, GeoRadiusCommandArgs args,
+ ByteBuffer store, ByteBuffer storeDist) {
+ super(key);
+ this.distance = distance;
+ this.point = point;
+ this.args = args == null ? GeoRadiusCommandArgs.newGeoRadiusArgs() : args;
+ this.store = store;
+ this.storeDist = storeDist;
+ }
+
+ public static GeoRadiusCommand within(Distance distance) {
+ return new GeoRadiusCommand(null, null, distance, null, null, null);
+ }
+
+ public static GeoRadiusCommand withinMeters(Double distance) {
+ return within(new Distance(distance, DistanceUnit.METERS));
+ }
+
+ public static GeoRadiusCommand withinKiometers(Double distance) {
+ return within(new Distance(distance, DistanceUnit.KILOMETERS));
+ }
+
+ public static GeoRadiusCommand withinMiles(Double distance) {
+ return within(new Distance(distance, DistanceUnit.MILES));
+ }
+
+ public static GeoRadiusCommand withinFeet(Double distance) {
+ return within(new Distance(distance, DistanceUnit.FEET));
+ }
+
+ public static GeoRadiusCommand within(Circle circle) {
+ return within(circle.getRadius()).from(circle.getCenter());
+ }
+
+ public GeoRadiusCommand from(Point center) {
+ return new GeoRadiusCommand(getKey(), center, distance, args, store, storeDist);
+ }
+
+ public GeoRadiusCommand withFlag(Flag flag) {
+
+ GeoRadiusCommandArgs args = cloneArgs();
+ args.flags.add(flag);
+
+ return new GeoRadiusCommand(getKey(), point, distance, args, store, storeDist);
+ }
+
+ public GeoRadiusCommand withCoord() {
+ return withFlag(Flag.WITHCOORD);
+ }
+
+ public GeoRadiusCommand withDist() {
+ return withFlag(Flag.WITHDIST);
+ }
+
+ public GeoRadiusCommand withArgs(GeoRadiusCommandArgs args) {
+ return new GeoRadiusCommand(getKey(), point, distance, args, store, storeDist);
+ }
+
+ public GeoRadiusCommand limitTo(Long limit) {
+
+ GeoRadiusCommandArgs args = cloneArgs();
+ if (limit != null) {
+ args = args.limit(limit);
+ }
+
+ return new GeoRadiusCommand(getKey(), point, distance, args, store, storeDist);
+ }
+
+ public GeoRadiusCommand sort(Direction direction) {
+
+ GeoRadiusCommandArgs args = cloneArgs();
+ args.sortDirection = direction;
+
+ return new GeoRadiusCommand(getKey(), point, distance, args, store, storeDist);
+ }
+
+ public GeoRadiusCommand orderByDistanceAsc() {
+ return sort(Direction.ASC);
+ }
+
+ public GeoRadiusCommand orderByDistanceDesc() {
+ return sort(Direction.DESC);
+ }
+
+ public GeoRadiusCommand forKey(ByteBuffer key) {
+ return new GeoRadiusCommand(key, point, distance, args, store, storeDist);
+ }
+
+ /**
+ * NOTE: STORE option is not compatible with WITHDIST, WITHHASH and WITHCOORDS options.
+ *
+ * @param key
+ * @return
+ */
+ public GeoRadiusCommand storeAt(ByteBuffer key) {
+ return new GeoRadiusCommand(getKey(), point, distance, args, key, storeDist);
+ }
+
+ /**
+ * NOTE: STOREDIST option is not compatible with WITHDIST, WITHHASH and WITHCOORDS options.
+ *
+ * @param key
+ * @return
+ */
+ public GeoRadiusCommand storeDistAt(ByteBuffer key) {
+ return new GeoRadiusCommand(getKey(), point, distance, args, store, key);
+ }
+
+ public Optional getDirection() {
+ return Optional.ofNullable(args.getSortDirection());
+ }
+
+ public Distance getDistance() {
+ return distance;
+ }
+
+ public Set getFlags() {
+ return args.getFlags();
+ }
+
+ public Optional getLimit() {
+ return Optional.ofNullable(args.getLimit());
+ }
+
+ public Point getPoint() {
+ return point;
+ }
+
+ public Optional getStore() {
+ return Optional.ofNullable(store);
+ }
+
+ public Optional getStoreDist() {
+ return Optional.ofNullable(storeDist);
+ }
+
+ public Optional getArgs() {
+ return Optional.ofNullable(args);
+ }
+
+ private GeoRadiusCommandArgs cloneArgs() {
+
+ if (args == null) {
+ return GeoRadiusCommandArgs.newGeoRadiusArgs();
+ }
+
+ return args.clone();
+ }
+ }
+
+ /**
+ * Get the {@literal member}s within the boundaries of a given {@link Circle}.
+ *
+ * @param key must not be {@literal null}.
+ * @param circle must not be {@literal null}.
+ * @return
+ */
+ default Mono>> geoRadius(ByteBuffer key, Circle circle) {
+ return geoRadius(key, circle, null)
+ .map(res -> res.getContent().stream().map(val -> val.getContent()).collect(Collectors.toList()));
+ }
+
+ /**
+ * Get the {@literal member}s within the boundaries of a given {@link Circle} applying given parameters.
+ *
+ * @param key must not be {@literal null}.
+ * @param circle must not be {@literal null}.
+ * @param geoRadiusArgs can be {@literal null}.
+ * @return
+ */
+ default Mono>> geoRadius(ByteBuffer key, Circle circle,
+ GeoRadiusCommandArgs geoRadiusArgs) {
+
+ Assert.notNull(key, "key must not be null");
+ Assert.notNull(circle, "circle must not be null");
+
+ return geoRadius(Mono.just(GeoRadiusCommand.within(circle).withArgs(geoRadiusArgs).forKey(key))).next()
+ .map(CommandResponse::getOutput);
+ }
+
+ /**
+ * Get the {@literal member}s within the boundaries of a given {@link Circle} applying given parameters.
+ *
+ * @param commands
+ * @return
+ */
+ Flux>>> geoRadius(
+ Publisher commands);
+
+ /**
+ * @author Christoph Strobl
+ */
+ class GeoRadiusByMemberCommand extends KeyCommand {
+
+ private final Distance distance;
+ private final ByteBuffer member;
+ private final GeoRadiusCommandArgs args;
+ private final ByteBuffer store;
+ private final ByteBuffer storeDist;
+
+ private GeoRadiusByMemberCommand(ByteBuffer key, ByteBuffer member, Distance distance, GeoRadiusCommandArgs args,
+ ByteBuffer store, ByteBuffer storeDist) {
+
+ super(key);
+ this.distance = distance;
+ this.member = member;
+ this.args = args == null ? GeoRadiusCommandArgs.newGeoRadiusArgs() : args;
+ this.store = store;
+ this.storeDist = storeDist;
+ }
+
+ public static GeoRadiusByMemberCommand within(Distance distance) {
+ return new GeoRadiusByMemberCommand(null, null, distance, GeoRadiusCommandArgs.newGeoRadiusArgs(), null, null);
+ }
+
+ public static GeoRadiusByMemberCommand withinMeters(Double distance) {
+ return within(new Distance(distance, DistanceUnit.METERS));
+ }
+
+ public static GeoRadiusByMemberCommand withinKiometers(Double distance) {
+ return within(new Distance(distance, DistanceUnit.KILOMETERS));
+ }
+
+ public static GeoRadiusByMemberCommand withinMiles(Double distance) {
+ return within(new Distance(distance, DistanceUnit.MILES));
+ }
+
+ public static GeoRadiusByMemberCommand withinFeet(Double distance) {
+ return within(new Distance(distance, DistanceUnit.FEET));
+ }
+
+ public GeoRadiusByMemberCommand from(ByteBuffer member) {
+ return new GeoRadiusByMemberCommand(getKey(), member, distance, args, store, storeDist);
+ }
+
+ public GeoRadiusByMemberCommand withArgs(GeoRadiusCommandArgs args) {
+ return new GeoRadiusByMemberCommand(getKey(), member, distance, args, store, storeDist);
+ }
+
+ public GeoRadiusByMemberCommand withFlag(Flag flag) {
+
+ GeoRadiusCommandArgs args = cloneArgs();
+ args.flags.add(flag);
+
+ return new GeoRadiusByMemberCommand(getKey(), member, distance, args, store, storeDist);
+ }
+
+ public GeoRadiusByMemberCommand withCoord() {
+ return withFlag(Flag.WITHCOORD);
+ }
+
+ public GeoRadiusByMemberCommand withDist() {
+ return withFlag(Flag.WITHDIST);
+ }
+
+ public GeoRadiusByMemberCommand limitTo(Long limit) {
+
+ GeoRadiusCommandArgs args = cloneArgs();
+ if (limit != null) {
+ args = args.limit(limit);
+ }
+
+ return new GeoRadiusByMemberCommand(getKey(), member, distance, args, store, storeDist);
+ }
+
+ public GeoRadiusByMemberCommand sort(Direction direction) {
+
+ GeoRadiusCommandArgs args = cloneArgs();
+ args.sortDirection = direction;
+
+ return new GeoRadiusByMemberCommand(getKey(), member, distance, args, store, storeDist);
+ }
+
+ public GeoRadiusByMemberCommand orderByDistanceAsc() {
+ return sort(Direction.ASC);
+ }
+
+ public GeoRadiusByMemberCommand orderByDistanceDesc() {
+ return sort(Direction.DESC);
+ }
+
+ public GeoRadiusByMemberCommand forKey(ByteBuffer key) {
+ return new GeoRadiusByMemberCommand(key, member, distance, args, store, storeDist);
+ }
+
+ /**
+ * NOTE: STORE option is not compatible with WITHDIST, WITHHASH and WITHCOORDS options.
+ *
+ * @param key
+ * @return
+ */
+ public GeoRadiusByMemberCommand storeAt(ByteBuffer key) {
+ return new GeoRadiusByMemberCommand(getKey(), member, distance, args, key, storeDist);
+ }
+
+ /**
+ * NOTE: STOREDIST option is not compatible with WITHDIST, WITHHASH and WITHCOORDS options.
+ *
+ * @param key
+ * @return
+ */
+ public GeoRadiusByMemberCommand storeDistAt(ByteBuffer key) {
+ return new GeoRadiusByMemberCommand(getKey(), member, distance, args, store, key);
+ }
+
+ public Optional getDirection() {
+ return Optional.ofNullable(args.getSortDirection());
+ }
+
+ public Distance getDistance() {
+ return distance;
+ }
+
+ public Set getFlags() {
+ return args.getFlags();
+ }
+
+ public Optional getLimit() {
+ return Optional.ofNullable(args.getLimit());
+ }
+
+ public ByteBuffer getMember() {
+ return member;
+ }
+
+ public Optional getStore() {
+ return Optional.ofNullable(store);
+ }
+
+ public Optional getStoreDist() {
+ return Optional.ofNullable(storeDist);
+ }
+
+ public Optional getArgs() {
+ return Optional.ofNullable(args);
+ }
+
+ private GeoRadiusCommandArgs cloneArgs() {
+
+ if (args == null) {
+ return GeoRadiusCommandArgs.newGeoRadiusArgs();
+ }
+
+ return args.clone();
+ }
+
+ }
+
+ /**
+ * Get the {@literal member}s within given {@link Distance} from {@literal member} applying given parameters.
+ *
+ * @param key must not be {@literal null}.
+ * @param member must not be {@literal null}.
+ * @return
+ */
+ default Mono>> geoRadiusByMember(ByteBuffer key, ByteBuffer member, Distance distance) {
+ return geoRadiusByMember(key, member, distance, null)
+ .map(res -> res.getContent().stream().map(val -> val.getContent()).collect(Collectors.toList()));
+ }
+
+ /**
+ * Get the {@literal member}s within given {@link Distance} from {@literal member} applying given parameters.
+ *
+ * @param key must not be {@literal null}.
+ * @param member must not be {@literal null}.
+ * @param geoRadiusArgs can be {@literal null}.
+ * @return
+ */
+ default Mono>> geoRadiusByMember(ByteBuffer key, ByteBuffer member,
+ Distance distance, GeoRadiusCommandArgs geoRadiusArgs) {
+
+ Assert.notNull(key, "key must not be null");
+ Assert.notNull(member, "member must not be null");
+ Assert.notNull(distance, "distance must not be null");
+
+ return geoRadiusByMember(
+ Mono.just(GeoRadiusByMemberCommand.within(distance).from(member).forKey(key).withArgs(geoRadiusArgs))).next()
+ .map(CommandResponse::getOutput);
+ }
+
+ /**
+ * Get the {@literal member}s within given {@link Distance} from {@literal member} applying given parameters.
+ *
+ * @param commands
+ * @return
+ */
+ Flux>>> geoRadiusByMember(
+ Publisher commands);
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveHashCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveHashCommands.java
new file mode 100644
index 000000000..47969e569
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/ReactiveHashCommands.java
@@ -0,0 +1,414 @@
+/*
+ * Copyright 2016 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 java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import org.reactivestreams.Publisher;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.BooleanResponse;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiValueResponse;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
+import org.springframework.util.Assert;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author Christoph Strobl
+ * @since 2.0
+ */
+public interface ReactiveHashCommands {
+
+ /**
+ * @author Christoph Strobl
+ */
+ class HSetCommand extends KeyCommand {
+
+ private static final ByteBuffer SINGLE_VALUE_KEY = ByteBuffer.allocate(0);
+ private final Map fieldValueMap;
+ private final Boolean upsert;
+
+ private HSetCommand(ByteBuffer key, Map keyValueMap, Boolean upsert) {
+
+ super(key);
+ this.fieldValueMap = keyValueMap;
+ this.upsert = upsert;
+ }
+
+ public static HSetCommand value(ByteBuffer value) {
+ return new HSetCommand(null, Collections.singletonMap(SINGLE_VALUE_KEY, value), Boolean.TRUE);
+ }
+
+ public static HSetCommand fieldValues(Map fieldValueMap) {
+ return new HSetCommand(null, fieldValueMap, Boolean.TRUE);
+ }
+
+ public HSetCommand ofField(ByteBuffer field) {
+
+ if (!fieldValueMap.containsKey(SINGLE_VALUE_KEY)) {
+ throw new InvalidDataAccessApiUsageException("Value has not been set.");
+ }
+
+ return new HSetCommand(getKey(), Collections.singletonMap(field, fieldValueMap.get(SINGLE_VALUE_KEY)), upsert);
+ }
+
+ public HSetCommand forKey(ByteBuffer key) {
+ return new HSetCommand(key, fieldValueMap, upsert);
+ }
+
+ public HSetCommand ifValueNotExists() {
+ return new HSetCommand(getKey(), fieldValueMap, Boolean.FALSE);
+ }
+
+ public Boolean isUpsert() {
+ return upsert;
+ }
+
+ public Map getFieldValueMap() {
+ return fieldValueMap;
+ }
+ }
+
+ /**
+ * Set the {@code value} of a hash {@code field}.
+ *
+ * @param key must not be {@literal null}.
+ * @param field must not be {@literal null}.
+ * @param value must not be {@literal null}.
+ * @return
+ */
+ default Mono hSet(ByteBuffer key, ByteBuffer field, ByteBuffer value) {
+
+ Assert.notNull(key, "key must not be null");
+ Assert.notNull(field, "field must not be null");
+ Assert.notNull(value, "value must not be null");
+
+ return hSet(Mono.just(HSetCommand.value(value).ofField(field).forKey(key))).next().map(BooleanResponse::getOutput);
+ }
+
+ /**
+ * Set the {@code value} of a hash {@code field}.
+ *
+ * @param key must not be {@literal null}.
+ * @param field must not be {@literal null}.
+ * @param value must not be {@literal null}.
+ * @return
+ */
+ default Mono hSetNX(ByteBuffer key, ByteBuffer field, ByteBuffer value) {
+
+ Assert.notNull(key, "key must not be null");
+ Assert.notNull(field, "field must not be null");
+ Assert.notNull(value, "value must not be null");
+
+ return hSet(Mono.just(HSetCommand.value(value).ofField(field).forKey(key).ifValueNotExists())).next()
+ .map(BooleanResponse::getOutput);
+ }
+
+ /**
+ * Set multiple hash fields to multiple values using data provided in {@code fieldValueMap}.
+ *
+ * @param key must not be {@literal null}.
+ * @param fieldValueMap must not be {@literal null}.
+ * @return
+ */
+ default Mono hMSet(ByteBuffer key, Map fieldValueMap) {
+
+ Assert.notNull(key, "key must not be null");
+ Assert.notNull(fieldValueMap, "field must not be null");
+
+ return hSet(Mono.just(HSetCommand.fieldValues(fieldValueMap).forKey(key).ifValueNotExists())).next()
+ .map(BooleanResponse::getOutput);
+ }
+
+ /**
+ * Set the {@code value} of a hash {@code field}.
+ *
+ * @param commands must not be {@literal null}.
+ * @return
+ */
+ Flux> hSet(Publisher commands);
+
+ /**
+ * @author Christoph Strobl
+ */
+ class HGetCommand extends KeyCommand {
+
+ private List fields;
+
+ private HGetCommand(ByteBuffer key, List fields) {
+
+ super(key);
+ this.fields = fields;
+ }
+
+ public static HGetCommand field(ByteBuffer field) {
+ return new HGetCommand(null, Collections.singletonList(field));
+ }
+
+ public static HGetCommand fields(List fields) {
+ return new HGetCommand(null, new ArrayList<>(fields));
+ }
+
+ public HGetCommand from(ByteBuffer key) {
+ return new HGetCommand(key, fields);
+ }
+
+ public List getFields() {
+ return fields;
+ }
+ }
+
+ /**
+ * Get value for given {@code field} from hash at {@code key}.
+ *
+ * @param key must not be {@literal null}.
+ * @param field must not be {@literal null}.
+ * @return
+ */
+ default Mono hGet(ByteBuffer key, ByteBuffer field) {
+ return hMGet(key, Collections.singletonList(field)).map(val -> val.isEmpty() ? null : val.iterator().next());
+ }
+
+ /**
+ * Get values for given {@code fields} from hash at {@code key}.
+ *
+ * @param key must not be {@literal null}.
+ * @param fields must not be {@literal null}.
+ * @return
+ */
+ default Mono> hMGet(ByteBuffer key, List fields) {
+
+ Assert.notNull(key, "key must not be null");
+ Assert.notNull(fields, "fields must not be null");
+
+ return hMGet(Mono.just(HGetCommand.fields(fields).from(key))).next().map(MultiValueResponse::getOutput);
+ }
+
+ /**
+ * Get values for given {@code fields} from hash at {@code key}.
+ *
+ * @param commands must not be {@literal null}.
+ * @return
+ */
+ Flux> hMGet(Publisher commands);
+
+ /**
+ * @author Christoph Strobl
+ */
+ class HExistsCommand extends KeyCommand {
+
+ private final ByteBuffer field;
+
+ private HExistsCommand(ByteBuffer key, ByteBuffer field) {
+
+ super(key);
+ this.field = field;
+ }
+
+ public static HExistsCommand field(ByteBuffer field) {
+ return new HExistsCommand(null, field);
+ }
+
+ public HExistsCommand in(ByteBuffer key) {
+ return new HExistsCommand(key, field);
+ }
+
+ public ByteBuffer getField() {
+ return field;
+ }
+ }
+
+ /**
+ * Determine if given hash {@code field} exists.
+ *
+ * @param key must not be {@literal null}.
+ * @param field must not be {@literal null}.
+ * @return
+ */
+ default Mono hExists(ByteBuffer key, ByteBuffer field) {
+
+ Assert.notNull(key, "key must not be null");
+ Assert.notNull(field, "field must not be null");
+
+ return hExists(Mono.just(HExistsCommand.field(field).in(key))).next().map(BooleanResponse::getOutput);
+ }
+
+ /**
+ * Determine if given hash {@code field} exists.
+ *
+ * @param commands
+ * @return
+ */
+ Flux> hExists(Publisher commands);
+
+ /**
+ * @author Christoph Strobl
+ */
+ class HDelCommand extends KeyCommand {
+
+ private final List fields;
+
+ private HDelCommand(ByteBuffer key, List fields) {
+ super(key);
+ this.fields = fields;
+ }
+
+ public static HDelCommand field(ByteBuffer field) {
+ return new HDelCommand(null, Collections.singletonList(field));
+ }
+
+ public static HDelCommand fields(List fields) {
+ return new HDelCommand(null, new ArrayList<>(fields));
+ }
+
+ public HDelCommand from(ByteBuffer key) {
+ return new HDelCommand(key, fields);
+ }
+
+ public List getFields() {
+ return fields;
+ }
+ }
+
+ /**
+ * Delete given hash {@code field}.
+ *
+ * @param key must not be {@literal null}.
+ * @param field must not be {@literal null}.
+ * @return
+ */
+ default Mono hDel(ByteBuffer key, ByteBuffer field) {
+
+ Assert.notNull(field, "field must not be null");
+
+ return hDel(key, Collections.singletonList(field)).map(val -> val > 0 ? Boolean.TRUE : Boolean.FALSE);
+ }
+
+ /**
+ * Delete given hash {@code fields}.
+ *
+ * @param key must not be {@literal null}.
+ * @param fields must not be {@literal null}.
+ * @return
+ */
+ default Mono hDel(ByteBuffer key, List fields) {
+
+ Assert.notNull(key, "key must not be null");
+ Assert.notNull(fields, "fields must not be null");
+
+ return hDel(Mono.just(HDelCommand.fields(fields).from(key))).next().map(NumericResponse::getOutput);
+ }
+
+ /**
+ * Delete given hash {@code fields}.
+ *
+ * @param commands must not be {@literal null}.
+ * @return
+ */
+ Flux> hDel(Publisher commands);
+
+ /**
+ * Get size of hash at {@code key}.
+ *
+ * @param key must not be {@literal null}.
+ * @return
+ */
+ default Mono hLen(ByteBuffer key) {
+
+ Assert.notNull(key, "key must not be null");
+
+ return hLen(Mono.just(new KeyCommand(key))).next().map(NumericResponse::getOutput);
+ }
+
+ /**
+ * Get size of hash at {@code key}.
+ *
+ * @param commands must not be {@literal null}.
+ * @return
+ */
+ Flux> hLen(Publisher commands);
+
+ /**
+ * Get key set (fields) of hash at {@code key}.
+ *
+ * @param key must not be {@literal null}.
+ * @return
+ */
+ default Mono> hKeys(ByteBuffer key) {
+
+ Assert.notNull(key, "key must not be null");
+
+ return hKeys(Mono.just(new KeyCommand(key))).next().map(MultiValueResponse::getOutput);
+ }
+
+ /**
+ * Get key set (fields) of hash at {@code key}.
+ *
+ * @param commands must not be {@literal null}.
+ * @return
+ */
+ Flux> hKeys(Publisher commands);
+
+ /**
+ * Get entry set (values) of hash at {@code key}.
+ *
+ * @param key must not be {@literal null}.
+ * @return
+ */
+ default Mono> hVals(ByteBuffer key) {
+
+ Assert.notNull(key, "key must not be null");
+
+ return hVals(Mono.just(new KeyCommand(key))).next().map(MultiValueResponse::getOutput);
+ }
+
+ /**
+ * Get entry set (values) of hash at {@code key}.
+ *
+ * @param commands must not be {@literal null}.
+ * @return
+ */
+ Flux> hVals(Publisher commands);
+
+ /**
+ * Get entire hash stored at {@code key}.
+ *
+ * @param key must not be {@literal null}.
+ * @return
+ */
+ default Mono