DATAREDIS-525 - Polishing.

Enhance JavaDoc. Improve assertions. Reduce accepted collection-type boundary to Collection for commands that do not enforce a specific order. Add ByteUtil.getBytes(ByteBuffer) utility method. Reformat code. Extend documentation

Fix parameter ambiguity in LettuceConnection.watch().

Original pull request: #229.
This commit is contained in:
Mark Paluch
2016-11-14 17:44:52 +01:00
parent 62b1354e02
commit b172c237e4
46 changed files with 3101 additions and 880 deletions

View File

@@ -19,6 +19,7 @@ import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
import org.springframework.data.redis.util.ByteUtils;
import org.springframework.util.Assert;
/**
@@ -69,7 +70,10 @@ public final class ClusterSlotHashUtil {
return true;
}
return isSameSlotForAllKeys((byte[][]) keys.stream().map(ByteBuffer::array).toArray(size -> new byte[size][]));
return isSameSlotForAllKeys((byte[][]) keys.stream() //
.map(ByteBuffer::duplicate) //
.map(ByteUtils::getBytes) //
.toArray(byte[][]::new));
}
/**
@@ -164,5 +168,4 @@ public final class ClusterSlotHashUtil {
}
return crc & 0xFFFF;
}
}

View File

@@ -168,7 +168,7 @@ public class ClusterTopology {
*/
public RedisClusterNode lookup(String nodeId) {
Assert.notNull(nodeId, "NodeId must not be null");
Assert.notNull(nodeId, "NodeId must not be null!");
for (RedisClusterNode node : nodes) {
if (nodeId.equals(node.getId())) {
@@ -189,7 +189,7 @@ public class ClusterTopology {
*/
public RedisClusterNode lookup(RedisClusterNode node) {
Assert.notNull(node, "RedisClusterNode must not be null");
Assert.notNull(node, "RedisClusterNode must not be null!");
if(nodes.contains(node) && StringUtils.hasText(node.getHost()) && StringUtils.hasText(node.getId())){
return node;

View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection;
/**

View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection;
/**

View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection;
/**

View File

@@ -13,23 +13,6 @@
* 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;
@@ -59,5 +42,4 @@ public interface ReactiveClusterKeyCommands extends ReactiveKeyCommands {
* @return
*/
Mono<ByteBuffer> randomKey(RedisClusterNode node);
}

View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection;
/**

View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection;
/**

View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection;
/**

View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection;
/**

View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection;
/**

View File

@@ -17,6 +17,7 @@ package org.springframework.data.redis.connection;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -24,6 +25,7 @@ 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.Command;
import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiValueResponse;
@@ -34,63 +36,115 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* Redis Hash commands executed using reactive infrastructure.
*
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public interface ReactiveHashCommands {
/**
* {@literal HSET} {@link Command}.
*
* @author Christoph Strobl
*/
class HSetCommand extends KeyCommand {
private static final ByteBuffer SINGLE_VALUE_KEY = ByteBuffer.allocate(0);
private final Map<ByteBuffer, ByteBuffer> fieldValueMap;
private final Boolean upsert;
private final boolean upsert;
private HSetCommand(ByteBuffer key, Map<ByteBuffer, ByteBuffer> keyValueMap, Boolean upsert) {
private HSetCommand(ByteBuffer key, Map<ByteBuffer, ByteBuffer> keyValueMap, boolean upsert) {
super(key);
this.fieldValueMap = keyValueMap;
this.upsert = upsert;
}
/**
* Creates a new {@link HSetCommand} given a {@link ByteBuffer key}.
*
* @param value must not be {@literal null}.
* @return a new {@link HSetCommand} for {@link ByteBuffer key}.
*/
public static HSetCommand value(ByteBuffer value) {
Assert.notNull(value, "Value must not be null!");
return new HSetCommand(null, Collections.singletonMap(SINGLE_VALUE_KEY, value), Boolean.TRUE);
}
/**
* Creates a new {@link HSetCommand} given a {@link Map} of field values.
*
* @param fieldValueMap must not be {@literal null}.
* @return a new {@link HSetCommand} for a {@link Map} of field values.
*/
public static HSetCommand fieldValues(Map<ByteBuffer, ByteBuffer> fieldValueMap) {
Assert.notNull(fieldValueMap, "Field values map must not be null!");
return new HSetCommand(null, fieldValueMap, Boolean.TRUE);
}
/**
* Applies a field. Constructs a new command instance with all previously configured properties.
*
* @param field must not be {@literal null}.
* @return a new {@link HSetCommand} with {@literal field} applied.
*/
public HSetCommand ofField(ByteBuffer field) {
if (!fieldValueMap.containsKey(SINGLE_VALUE_KEY)) {
throw new InvalidDataAccessApiUsageException("Value has not been set.");
}
Assert.notNull(field, "Field not be null!");
return new HSetCommand(getKey(), Collections.singletonMap(field, fieldValueMap.get(SINGLE_VALUE_KEY)), upsert);
}
/**
* Applies the {@literal key}. Constructs a new command instance with all previously configured properties.
*
* @param key must not be {@literal null}.
* @return a new {@link HSetCommand} with {@literal key} applied.
*/
public HSetCommand forKey(ByteBuffer key) {
Assert.notNull(key, "Key not be null!");
return new HSetCommand(key, fieldValueMap, upsert);
}
/**
* Disable upsert. Constructs a new command instance with all previously configured properties.
*
* @return a new {@link HSetCommand} with upsert disabled.
*/
public HSetCommand ifValueNotExists() {
return new HSetCommand(getKey(), fieldValueMap, Boolean.FALSE);
}
public Boolean isUpsert() {
/**
* @return
*/
public boolean isUpsert() {
return upsert;
}
/**
* @return
*/
public Map<ByteBuffer, ByteBuffer> getFieldValueMap() {
return fieldValueMap;
}
}
/**
* Set the {@code value} of a hash {@code field}.
* Set the {@literal value} of a hash {@literal field}.
*
* @param key must not be {@literal null}.
* @param field must not be {@literal null}.
@@ -99,15 +153,15 @@ public interface ReactiveHashCommands {
*/
default Mono<Boolean> 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");
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}.
* Set the {@literal value} of a hash {@literal field}.
*
* @param key must not be {@literal null}.
* @param field must not be {@literal null}.
@@ -116,16 +170,16 @@ public interface ReactiveHashCommands {
*/
default Mono<Boolean> 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");
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}.
* Set multiple hash fields to multiple values using data provided in {@literal fieldValueMap}.
*
* @param key must not be {@literal null}.
* @param fieldValueMap must not be {@literal null}.
@@ -133,15 +187,15 @@ public interface ReactiveHashCommands {
*/
default Mono<Boolean> hMSet(ByteBuffer key, Map<ByteBuffer, ByteBuffer> fieldValueMap) {
Assert.notNull(key, "key must not be null");
Assert.notNull(fieldValueMap, "field must not be null");
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}.
* Set the {@literal value} of a hash {@literal field}.
*
* @param commands must not be {@literal null}.
* @return
@@ -149,6 +203,8 @@ public interface ReactiveHashCommands {
Flux<BooleanResponse<HSetCommand>> hSet(Publisher<HSetCommand> commands);
/**
* {@literal HGET} {@link Command}.
*
* @author Christoph Strobl
*/
class HGetCommand extends KeyCommand {
@@ -158,28 +214,59 @@ public interface ReactiveHashCommands {
private HGetCommand(ByteBuffer key, List<ByteBuffer> fields) {
super(key);
this.fields = fields;
}
/**
* Creates a new {@link HGetCommand} given a {@link ByteBuffer field name}.
*
* @param field must not be {@literal null}.
* @return a new {@link HGetCommand} for a {@link ByteBuffer field name}.
*/
public static HGetCommand field(ByteBuffer field) {
Assert.notNull(field, "Field must not be null!");
return new HGetCommand(null, Collections.singletonList(field));
}
public static HGetCommand fields(List<ByteBuffer> fields) {
/**
* Creates a new {@link HGetCommand} given a {@link Collection} of field names.
*
* @param fields must not be {@literal null}.
* @return a new {@link HGetCommand} for a {@link Collection} of field names.
*/
public static HGetCommand fields(Collection<ByteBuffer> fields) {
Assert.notNull(fields, "Fields must not be null!");
return new HGetCommand(null, new ArrayList<>(fields));
}
/**
* Applies the hash {@literal key}. Constructs a new command instance with all previously configured properties.
*
* @param key must not be {@literal null}.
* @return a new {@link HGetCommand} with {@literal key} applied.
*/
public HGetCommand from(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new HGetCommand(key, fields);
}
/**
* @return
*/
public List<ByteBuffer> getFields() {
return fields;
}
}
/**
* Get value for given {@code field} from hash at {@code key}.
* Get value for given {@literal field} from hash at {@literal key}.
*
* @param key must not be {@literal null}.
* @param field must not be {@literal null}.
@@ -190,22 +277,22 @@ public interface ReactiveHashCommands {
}
/**
* Get values for given {@code fields} from hash at {@code key}.
* Get values for given {@literal fields} from hash at {@literal key}.
*
* @param key must not be {@literal null}.
* @param fields must not be {@literal null}.
* @return
*/
default Mono<List<ByteBuffer>> hMGet(ByteBuffer key, List<ByteBuffer> fields) {
default Mono<List<ByteBuffer>> hMGet(ByteBuffer key, Collection<ByteBuffer> fields) {
Assert.notNull(key, "key must not be null");
Assert.notNull(fields, "fields must not be null");
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}.
* Get values for given {@literal fields} from hash at {@literal key}.
*
* @param commands must not be {@literal null}.
* @return
@@ -213,6 +300,8 @@ public interface ReactiveHashCommands {
Flux<MultiValueResponse<HGetCommand, ByteBuffer>> hMGet(Publisher<HGetCommand> commands);
/**
* {@literal HEXISTS} {@link Command}.
*
* @author Christoph Strobl
*/
class HExistsCommand extends KeyCommand {
@@ -222,24 +311,46 @@ public interface ReactiveHashCommands {
private HExistsCommand(ByteBuffer key, ByteBuffer field) {
super(key);
this.field = field;
}
/**
* Creates a new {@link HExistsCommand} given a {@link ByteBuffer field name}.
*
* @param field must not be {@literal null}.
* @return a new {@link HExistsCommand} for a {@link ByteBuffer field name}.
*/
public static HExistsCommand field(ByteBuffer field) {
Assert.notNull(field, "Field must not be null!");
return new HExistsCommand(null, field);
}
/**
* Applies the hash {@literal key}. Constructs a new command instance with all previously configured properties.
*
* @param key must not be {@literal null}.
* @return a new {@link HExistsCommand} with {@literal key} applied.
*/
public HExistsCommand in(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new HExistsCommand(key, field);
}
/**
* @return
*/
public ByteBuffer getField() {
return field;
}
}
/**
* Determine if given hash {@code field} exists.
* Determine if given hash {@literal field} exists.
*
* @param key must not be {@literal null}.
* @param field must not be {@literal null}.
@@ -247,14 +358,14 @@ public interface ReactiveHashCommands {
*/
default Mono<Boolean> hExists(ByteBuffer key, ByteBuffer field) {
Assert.notNull(key, "key must not be null");
Assert.notNull(field, "field must not be null");
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.
* Determine if given hash {@literal field} exists.
*
* @param commands
* @return
@@ -269,29 +380,61 @@ public interface ReactiveHashCommands {
private final List<ByteBuffer> fields;
private HDelCommand(ByteBuffer key, List<ByteBuffer> fields) {
super(key);
this.fields = fields;
}
/**
* Creates a new {@link HDelCommand} given a {@link ByteBuffer field name}.
*
* @param field must not be {@literal null}.
* @return a new {@link HDelCommand} for a {@link ByteBuffer field name}.
*/
public static HDelCommand field(ByteBuffer field) {
Assert.notNull(field, "Field must not be null!");
return new HDelCommand(null, Collections.singletonList(field));
}
public static HDelCommand fields(List<ByteBuffer> fields) {
/**
* Creates a new {@link HDelCommand} given a {@link Collection} of field names.
*
* @param fields must not be {@literal null}.
* @return a new {@link HDelCommand} for a {@link Collection} of field names.
*/
public static HDelCommand fields(Collection<ByteBuffer> fields) {
Assert.notNull(fields, "Fields must not be null!");
return new HDelCommand(null, new ArrayList<>(fields));
}
/**
* Applies the hash {@literal key}. Constructs a new command instance with all previously configured properties.
*
* @param key must not be {@literal null}.
* @return a new {@link HDelCommand} with {@literal key} applied.
*/
public HDelCommand from(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new HDelCommand(key, fields);
}
/**
* @return
*/
public List<ByteBuffer> getFields() {
return fields;
}
}
/**
* Delete given hash {@code field}.
* Delete given hash {@literal field}.
*
* @param key must not be {@literal null}.
* @param field must not be {@literal null}.
@@ -299,28 +442,28 @@ public interface ReactiveHashCommands {
*/
default Mono<Boolean> hDel(ByteBuffer key, ByteBuffer field) {
Assert.notNull(field, "field must not be null");
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}.
* Delete given hash {@literal fields}.
*
* @param key must not be {@literal null}.
* @param fields must not be {@literal null}.
* @return
*/
default Mono<Long> hDel(ByteBuffer key, List<ByteBuffer> fields) {
default Mono<Long> hDel(ByteBuffer key, Collection<ByteBuffer> fields) {
Assert.notNull(key, "key must not be null");
Assert.notNull(fields, "fields must not be null");
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}.
* Delete given hash {@literal fields}.
*
* @param commands must not be {@literal null}.
* @return
@@ -328,20 +471,20 @@ public interface ReactiveHashCommands {
Flux<NumericResponse<HDelCommand, Long>> hDel(Publisher<HDelCommand> commands);
/**
* Get size of hash at {@code key}.
* Get size of hash at {@literal key}.
*
* @param key must not be {@literal null}.
* @return
*/
default Mono<Long> hLen(ByteBuffer key) {
Assert.notNull(key, "key must not be null");
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}.
* Get size of hash at {@literal key}.
*
* @param commands must not be {@literal null}.
* @return
@@ -349,20 +492,20 @@ public interface ReactiveHashCommands {
Flux<NumericResponse<KeyCommand, Long>> hLen(Publisher<KeyCommand> commands);
/**
* Get key set (fields) of hash at {@code key}.
* Get key set (fields) of hash at {@literal key}.
*
* @param key must not be {@literal null}.
* @return
*/
default Mono<List<ByteBuffer>> hKeys(ByteBuffer key) {
Assert.notNull(key, "key must not be null");
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}.
* Get key set (fields) of hash at {@literal key}.
*
* @param commands must not be {@literal null}.
* @return
@@ -370,20 +513,20 @@ public interface ReactiveHashCommands {
Flux<MultiValueResponse<KeyCommand, ByteBuffer>> hKeys(Publisher<KeyCommand> commands);
/**
* Get entry set (values) of hash at {@code key}.
* Get entry set (values) of hash at {@literal key}.
*
* @param key must not be {@literal null}.
* @return
*/
default Mono<List<ByteBuffer>> hVals(ByteBuffer key) {
Assert.notNull(key, "key must not be null");
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}.
* Get entry set (values) of hash at {@literal key}.
*
* @param commands must not be {@literal null}.
* @return
@@ -391,24 +534,23 @@ public interface ReactiveHashCommands {
Flux<MultiValueResponse<KeyCommand, ByteBuffer>> hVals(Publisher<KeyCommand> commands);
/**
* Get entire hash stored at {@code key}.
* Get entire hash stored at {@literal key}.
*
* @param key must not be {@literal null}.
* @return
*/
default Mono<Map<ByteBuffer, ByteBuffer>> hGetAll(ByteBuffer key) {
Assert.notNull(key, "key must not be null");
Assert.notNull(key, "Key must not be null!");
return hGetAll(Mono.just(new KeyCommand(key))).next().map(CommandResponse::getOutput);
}
/**
* Get entire hash stored at {@code key}.
* Get entire hash stored at {@literal key}.
*
* @param commands must not be {@literal null}.
* @return
*/
Flux<CommandResponse<KeyCommand, Map<ByteBuffer, ByteBuffer>>> hGetAll(Publisher<KeyCommand> commands);
}

View File

@@ -17,6 +17,7 @@ package org.springframework.data.redis.connection;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
@@ -31,12 +32,17 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* Redis HyperLogLog commands executed using reactive infrastructure.
*
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public interface ReactiveHyperLogLogCommands {
/**
* {@code PFADD} command parameters.
*
* @author Christoph Strobl
*/
class PfAddCommand extends KeyCommand {
@@ -49,18 +55,48 @@ public interface ReactiveHyperLogLogCommands {
this.values = values;
}
/**
* Creates a new {@link PfAddCommand} given a {@link ByteBuffer value}.
*
* @param value must not be {@literal null}.
* @return a new {@link PfAddCommand} for {@link ByteBuffer value}.
*/
public static PfAddCommand value(ByteBuffer value) {
Assert.notNull(value, "Value must not be null!");
return values(Collections.singletonList(value));
}
public static PfAddCommand values(List<ByteBuffer> values) {
/**
* Creates a new {@link PfAddCommand} given a {@link Collection} of {@link ByteBuffer values}.
*
* @param values must not be {@literal null}.
* @return a new {@link PfAddCommand} for {@link ByteBuffer key}.
*/
public static PfAddCommand values(Collection<ByteBuffer> values) {
Assert.notNull(values, "Values must not be null!");
return new PfAddCommand(null, new ArrayList<>(values));
}
/**
* Applies the {@literal key}. Constructs a new command instance with all previously configured properties.
*
* @param key must not be {@literal null}.
* @return a new {@link PfAddCommand} with {@literal key} applied.
*/
public PfAddCommand to(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new PfAddCommand(key, values);
}
/**
* @return
*/
public List<ByteBuffer> getValues() {
return values;
}
@@ -75,7 +111,7 @@ public interface ReactiveHyperLogLogCommands {
*/
default Mono<Long> pfAdd(ByteBuffer key, ByteBuffer value) {
Assert.notNull(value, "value must not be null");
Assert.notNull(value, "Value must not be null!");
return pfAdd(key, Collections.singletonList(value));
}
@@ -87,10 +123,10 @@ public interface ReactiveHyperLogLogCommands {
* @param values must not be {@literal null}.
* @return
*/
default Mono<Long> pfAdd(ByteBuffer key, List<ByteBuffer> values) {
default Mono<Long> pfAdd(ByteBuffer key, Collection<ByteBuffer> values) {
Assert.notNull(key, "key must not be null");
Assert.notNull(values, "values must not be null");
Assert.notNull(key, "Key must not be null!");
Assert.notNull(values, "Values must not be null!");
return pfAdd(Mono.just(PfAddCommand.values(values).to(key))).next().map(NumericResponse::getOutput);
}
@@ -104,6 +140,8 @@ public interface ReactiveHyperLogLogCommands {
Flux<NumericResponse<PfAddCommand, Long>> pfAdd(Publisher<PfAddCommand> commands);
/**
* {@code PFCOUNT} command parameters.
*
* @author Christoph Strobl
*/
class PfCountCommand implements Command {
@@ -116,23 +154,46 @@ public interface ReactiveHyperLogLogCommands {
this.keys = keys;
}
/**
* Creates a new {@link PfCountCommand} given a {@link ByteBuffer key}.
*
* @param key must not be {@literal null}.
* @return a new {@link PfCountCommand} for {@link ByteBuffer key}.
*/
public static PfCountCommand valueIn(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return valuesIn(Collections.singletonList(key));
}
public static PfCountCommand valuesIn(List<ByteBuffer> keys) {
return new PfCountCommand(keys);
/**
* Creates a new {@link PfCountCommand} given a {@link Collection} of {@literal keys}.
*
* @param keys must not be {@literal null}.
* @return a new {@link PfCountCommand} for {@literal keys}.
*/
public static PfCountCommand valuesIn(Collection<ByteBuffer> keys) {
Assert.notNull(keys, "Keys must not be null!");
return new PfCountCommand(new ArrayList<>(keys));
}
/**
* @return
*/
public List<ByteBuffer> getKeys() {
return keys;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveRedisConnection.Command#getKey()
*/
@Override
public ByteBuffer getKey() {
return null;
}
}
/**
@@ -143,7 +204,7 @@ public interface ReactiveHyperLogLogCommands {
*/
default Mono<Long> pfCount(ByteBuffer key) {
Assert.notNull(key, "key must not be null");
Assert.notNull(key, "Key must not be null!");
return pfCount(Collections.singletonList(key));
}
@@ -154,9 +215,9 @@ public interface ReactiveHyperLogLogCommands {
* @param keys must not be {@literal null}.
* @return
*/
default Mono<Long> pfCount(List<ByteBuffer> keys) {
default Mono<Long> pfCount(Collection<ByteBuffer> keys) {
Assert.notNull(keys, "keys must not be null");
Assert.notNull(keys, "Keys must not be null!");
return pfCount(Mono.just(PfCountCommand.valuesIn(keys))).next().map(NumericResponse::getOutput);
}
@@ -170,6 +231,8 @@ public interface ReactiveHyperLogLogCommands {
Flux<NumericResponse<PfCountCommand, Long>> pfCount(Publisher<PfCountCommand> commands);
/**
* {@code PFMERGE} command parameters.
*
* @author Christoph Strobl
*/
class PfMergeCommand extends KeyCommand {
@@ -182,14 +245,36 @@ public interface ReactiveHyperLogLogCommands {
this.sourceKeys = sourceKeys;
}
public static PfMergeCommand valuesIn(List<ByteBuffer> sourceKeys) {
return new PfMergeCommand(null, sourceKeys);
/**
* Creates a new {@link PfMergeCommand} given a {@link Collection} of {@literal sourceKeys}.
*
* @param sourceKeys must not be {@literal null}.
* @return a new {@link PfMergeCommand} for {@literal sourceKeys}.
*/
public static PfMergeCommand valuesIn(Collection<ByteBuffer> sourceKeys) {
Assert.notNull(sourceKeys, "Source keys must not be null!");
return new PfMergeCommand(null, new ArrayList<>(sourceKeys));
}
/**
* Applies the {@literal destinationKey}. Constructs a new command instance with all previously configured
* properties.
*
* @param destinationKey must not be {@literal null}.
* @return a new {@link PfMergeCommand} with {@literal destinationKey} applied.
*/
public PfMergeCommand into(ByteBuffer destinationKey) {
Assert.notNull(destinationKey, "Destination key must not be null!");
return new PfMergeCommand(destinationKey, sourceKeys);
}
/**
* @return
*/
public List<ByteBuffer> getSourceKeys() {
return sourceKeys;
}
@@ -202,10 +287,10 @@ public interface ReactiveHyperLogLogCommands {
* @param sourceKeys must not be {@literal null}.
* @return
*/
default Mono<Boolean> pfMerge(ByteBuffer destinationKey, List<ByteBuffer> sourceKeys) {
default Mono<Boolean> pfMerge(ByteBuffer destinationKey, Collection<ByteBuffer> sourceKeys) {
Assert.notNull(destinationKey, "destinationKey must not be null");
Assert.notNull(sourceKeys, "sourceKeys must not be null");
Assert.notNull(destinationKey, "DestinationKey must not be null!");
Assert.notNull(sourceKeys, "SourceKeys must not be null!");
return pfMerge(Mono.just(PfMergeCommand.valuesIn(sourceKeys).into(destinationKey))).next()
.map(BooleanResponse::getOutput);
@@ -218,5 +303,4 @@ public interface ReactiveHyperLogLogCommands {
* @return
*/
Flux<BooleanResponse<PfMergeCommand>> pfMerge(Publisher<PfMergeCommand> commands);
}

View File

@@ -30,13 +30,16 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* Redis Key commands executed using reactive infrastructure.
*
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public interface ReactiveKeyCommands {
/**
* Determine if given {@code key} exists.
* Determine if given {@literal key} exists.
*
* @param key must not be {@literal null}.
* @return
@@ -49,7 +52,7 @@ public interface ReactiveKeyCommands {
}
/**
* Determine if given {@code key} exists.
* Determine if given {@literal key} exists.
*
* @param keys must not be {@literal null}.
* @return
@@ -57,20 +60,20 @@ public interface ReactiveKeyCommands {
Flux<BooleanResponse<KeyCommand>> exists(Publisher<KeyCommand> keys);
/**
* Determine the type stored at {@code key}.
* Determine the type stored at {@literal key}.
*
* @param key must not be {@literal null}.
* @return
*/
default Mono<DataType> type(ByteBuffer key) {
Assert.notNull(key, "key must not be null");
Assert.notNull(key, "Key must not be null!");
return type(Mono.just(new KeyCommand(key))).next().map(CommandResponse::getOutput);
}
/**
* Determine the type stored at {@code key}.
* Determine the type stored at {@literal key}.
*
* @param keys must not be {@literal null}.
* @return
@@ -78,14 +81,14 @@ public interface ReactiveKeyCommands {
Flux<CommandResponse<KeyCommand, DataType>> type(Publisher<KeyCommand> keys);
/**
* Find all keys matching the given {@code pattern}.
* Find all keys matching the given {@literal pattern}.
*
* @param pattern must not be {@literal null}.
* @return
*/
default Mono<List<ByteBuffer>> keys(ByteBuffer pattern) {
Assert.notNull(pattern, "pattern must not be null");
Assert.notNull(pattern, "Pattern must not be null!");
return keys(Mono.just(pattern)).next().map(MultiValueResponse::getOutput);
}
@@ -98,7 +101,7 @@ public interface ReactiveKeyCommands {
Mono<ByteBuffer> randomKey();
/**
* Find all keys matching the given {@code pattern}.
* Find all keys matching the given {@literal pattern}.
*
* @param patterns must not be {@literal null}.
* @return
@@ -106,6 +109,8 @@ public interface ReactiveKeyCommands {
Flux<MultiValueResponse<ByteBuffer, ByteBuffer>> keys(Publisher<ByteBuffer> patterns);
/**
* {@code RENAME} command parameters.
*
* @author Christoph Strobl
*/
class RenameCommand extends KeyCommand {
@@ -115,24 +120,46 @@ public interface ReactiveKeyCommands {
private RenameCommand(ByteBuffer key, ByteBuffer newName) {
super(key);
this.newName = newName;
}
public static ReactiveKeyCommands.RenameCommand key(ByteBuffer key) {
/**
* Creates a new {@link RenameCommand} given a {@link ByteBuffer key}.
*
* @param key must not be {@literal null}.
* @return a new {@link RenameCommand} for {@link ByteBuffer key}.
*/
public static RenameCommand key(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new RenameCommand(key, null);
}
public ReactiveKeyCommands.RenameCommand to(ByteBuffer newName) {
/**
* Applies the {@literal newName}. Constructs a new command instance with all previously configured properties.
*
* @param newName must not be {@literal null}.
* @return a new {@link RenameCommand} with {@literal newName} applied.
*/
public RenameCommand to(ByteBuffer newName) {
Assert.notNull(newName, "New name must not be null!");
return new RenameCommand(getKey(), newName);
}
/**
* @return
*/
public ByteBuffer getNewName() {
return newName;
}
}
/**
* Rename key {@code oleName} to {@code newName}.
* Rename key {@literal oleName} to {@literal newName}.
*
* @param key must not be {@literal null}.
* @param newName must not be {@literal null}.
@@ -140,15 +167,15 @@ public interface ReactiveKeyCommands {
*/
default Mono<Boolean> rename(ByteBuffer key, ByteBuffer newName) {
Assert.notNull(key, "key must not be null");
Assert.notNull(key, "Key must not be null!");
return rename(Mono.just(RenameCommand.key(key).to(newName))).next().map(BooleanResponse::getOutput);
}
Flux<BooleanResponse<ReactiveKeyCommands.RenameCommand>> rename(Publisher<ReactiveKeyCommands.RenameCommand> cmd);
Flux<BooleanResponse<RenameCommand>> rename(Publisher<RenameCommand> cmd);
/**
* Rename key {@code oleName} to {@code newName} only if {@code newName} does not exist.
* Rename key {@literal oleName} to {@literal newName} only if {@literal newName} does not exist.
*
* @param key must not be {@literal null}.
* @param newName must not be {@literal null}.
@@ -156,20 +183,18 @@ public interface ReactiveKeyCommands {
*/
default Mono<Boolean> renameNX(ByteBuffer key, ByteBuffer newName) {
Assert.notNull(key, "key must not be null");
Assert.notNull(key, "Key must not be null!");
return renameNX(Mono.just(RenameCommand.key(key).to(newName))).next().map(BooleanResponse::getOutput);
}
/**
* Rename key {@code oleName} to {@code newName} only if {@code newName} does not exist.
* Rename key {@literal oleName} to {@literal newName} only if {@literal newName} does not exist.
*
* @param keys must not be {@literal null}.
* @param newName must not be {@literal null}.
* @param command must not be {@literal null}.
* @return
*/
Flux<BooleanResponse<ReactiveKeyCommands.RenameCommand>> renameNX(
Publisher<ReactiveKeyCommands.RenameCommand> command);
Flux<BooleanResponse<RenameCommand>> renameNX(Publisher<RenameCommand> command);
/**
* Delete {@literal key}.
@@ -188,7 +213,7 @@ public interface ReactiveKeyCommands {
* Delete {@literal keys} one by one.
*
* @param keys must not be {@literal null}.
* @return {@link Flux} of {@link DelResponse} holding the {@literal key} removed along with the deletion result.
* @return {@link Flux} of {@link NumericResponse} holding the {@literal key} removed along with the deletion result.
*/
Flux<NumericResponse<KeyCommand, Long>> del(Publisher<KeyCommand> keys);
@@ -209,7 +234,7 @@ public interface ReactiveKeyCommands {
* Delete multiple {@literal keys} in batches.
*
* @param keys must not be {@literal null}.
* @return {@link Flux} of {@link MDelResponse} holding the {@literal keys} removed along with the deletion result.
* @return {@link Flux} of {@link NumericResponse} holding the {@literal keys} removed along with the deletion result.
*/
Flux<NumericResponse<List<ByteBuffer>, Long>> mDel(Publisher<List<ByteBuffer>> keys);
}

View File

@@ -26,27 +26,29 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* Redis numeric commands executed using reactive infrastructure.
*
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public interface ReactiveNumberCommands {
/**
* Increment value of {@code key} by 1.
* Increment value of {@literal key} by 1.
*
* @param key must not be {@literal null}.
* @return
*/
default Mono<Long> incr(ByteBuffer key) {
Assert.notNull(key, "key must not be null");
Assert.notNull(key, "Key must not be null!");
return incr(Mono.just(new KeyCommand(key))).next().map(NumericResponse::getOutput);
}
/**
* Increment value of {@code key} by 1.
* Increment value of {@literal key} by 1.
*
* @param keys must not be {@literal null}.
* @return
@@ -54,6 +56,8 @@ public interface ReactiveNumberCommands {
Flux<NumericResponse<KeyCommand, Long>> incr(Publisher<KeyCommand> keys);
/**
* {@code INCRBY} command parameters.
*
* @author Christoph Strobl
*/
class IncrByCommand<T extends Number> extends KeyCommand {
@@ -65,22 +69,43 @@ public interface ReactiveNumberCommands {
this.value = value;
}
/**
* Creates a new {@link IncrByCommand} given a {@link ByteBuffer key}.
*
* @param key must not be {@literal null}.
* @return a new {@link IncrByCommand} for {@link ByteBuffer key}.
*/
public static <T extends Number> IncrByCommand<T> incr(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new IncrByCommand<T>(key, null);
}
/**
* Applies the numeric {@literal value}. Constructs a new command instance with all previously configured
* properties.
*
* @param value must not be {@literal null}.
* @return a new {@link IncrByCommand} with {@literal value} applied.
*/
public IncrByCommand<T> by(T value) {
Assert.notNull(value, "Value must not be null!");
return new IncrByCommand<T>(getKey(), value);
}
/**
* @return
*/
public T getValue() {
return value;
}
}
/**
* Increment value of {@code key} by {@code value}.
* Increment value of {@literal key} by {@literal value}.
*
* @param key must not be {@literal null}.
* @param value must not be {@literal null}.
@@ -88,25 +113,24 @@ public interface ReactiveNumberCommands {
*/
default <T extends Number> Mono<T> incrBy(ByteBuffer key, T value) {
Assert.notNull(key, "key must not be null");
Assert.notNull(value, "value must not be null");
Assert.notNull(key, "Key must not be null!");
Assert.notNull(value, "Value must not be null!");
return incrBy(Mono.just(IncrByCommand.<T> incr(key).by(value))).next().map(NumericResponse::getOutput);
}
/**
* Increment value of {@code key} by {@code value}.
* Increment value of {@literal key} by {@literal value}.
*
* @param keys must not be {@literal null}.
* @param value must not be {@literal null}.
* @param commands must not be {@literal null}.
* @return
*/
<T extends Number> Flux<NumericResponse<ReactiveNumberCommands.IncrByCommand<T>, T>> incrBy(
Publisher<ReactiveNumberCommands.IncrByCommand<T>> commands);
/**
* {@code DECRBY} command parameters.
*
* @author Christoph Strobl
*/
class DecrByCommand<T extends Number> extends KeyCommand {
@@ -118,37 +142,56 @@ public interface ReactiveNumberCommands {
this.value = value;
}
public static <T extends Number> ReactiveNumberCommands.DecrByCommand<T> decr(ByteBuffer key) {
/**
* Creates a new {@link DecrByCommand} given a {@link ByteBuffer key}.
*
* @param key must not be {@literal null}.
* @return a new {@link DecrByCommand} for {@link ByteBuffer key}.
*/
public static <T extends Number> DecrByCommand<T> decr(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new DecrByCommand<T>(key, null);
}
public ReactiveNumberCommands.DecrByCommand<T> by(T value) {
/**
* Applies the numeric {@literal value}. Constructs a new command instance with all previously configured
* properties.
*
* @param value must not be {@literal null}.
* @return a new {@link DecrByCommand} with {@literal value} applied.
*/
public DecrByCommand<T> by(T value) {
Assert.notNull(value, "Value must not be null!");
return new DecrByCommand<T>(getKey(), value);
}
/**
* @return
*/
public T getValue() {
return value;
}
}
/**
* Decrement value of {@code key} by 1.
* Decrement value of {@literal key} by 1.
*
* @param key must not be {@literal null}.
* @return
*/
default Mono<Long> decr(ByteBuffer key) {
Assert.notNull(key, "key must not be null");
Assert.notNull(key, "Key must not be null!");
return decr(Mono.just(new KeyCommand(key))).next().map(NumericResponse::getOutput);
}
/**
* Decrement value of {@code key} by 1.
* Decrement value of {@literal key} by 1.
*
* @param keys must not be {@literal null}.
* @return
@@ -156,7 +199,7 @@ public interface ReactiveNumberCommands {
Flux<NumericResponse<KeyCommand, Long>> decr(Publisher<KeyCommand> keys);
/**
* Decrement value of {@code key} by {@code value}.
* Decrement value of {@literal key} by {@literal value}.
*
* @param key must not be {@literal null}.
* @param value must not be {@literal null}.
@@ -164,25 +207,23 @@ public interface ReactiveNumberCommands {
*/
default <T extends Number> Mono<T> decrBy(ByteBuffer key, T value) {
Assert.notNull(key, "key must not be null");
Assert.notNull(value, "value must not be null");
Assert.notNull(key, "Key must not be null!");
Assert.notNull(value, "Value must not be null!");
return decrBy(Mono.just(DecrByCommand.<T> decr(key).by(value))).next().map(NumericResponse::getOutput);
}
/**
* Decrement value of {@code key} by {@code value}.
* Decrement value of {@literal key} by {@literal value}.
*
* @param keys must not be {@literal null}.
* @param value must not be {@literal null}.
* @param commands must not be {@literal null}.
* @return
*/
<T extends Number> Flux<NumericResponse<ReactiveNumberCommands.DecrByCommand<T>, T>> decrBy(
Publisher<ReactiveNumberCommands.DecrByCommand<T>> commands);
<T extends Number> Flux<NumericResponse<DecrByCommand<T>, T>> decrBy(Publisher<DecrByCommand<T>> commands);
/**
* {@code HINCRBY} command parameters.
*
* @author Christoph Strobl
*/
class HIncrByCommand<T extends Number> extends KeyCommand {
@@ -193,33 +234,68 @@ public interface ReactiveNumberCommands {
private HIncrByCommand(ByteBuffer key, ByteBuffer field, T value) {
super(key);
this.field = field;
this.value = value;
}
/**
* Creates a new {@link HIncrByCommand} given a {@link ByteBuffer key}.
*
* @param field must not be {@literal null}.
* @return a new {@link HIncrByCommand} for {@link ByteBuffer key}.
*/
public static <T extends Number> HIncrByCommand<T> incr(ByteBuffer field) {
Assert.notNull(field, "Field must not be null!");
return new HIncrByCommand<T>(null, field, null);
}
/**
* Applies the numeric {@literal value}. Constructs a new command instance with all previously configured
* properties.
*
* @param value must not be {@literal null}.
* @return a new {@link HIncrByCommand} with {@literal value} applied.
*/
public HIncrByCommand<T> by(T value) {
Assert.notNull(value, "Value must not be null!");
return new HIncrByCommand<T>(getKey(), field, value);
}
/**
* Applies the {@literal key}. Constructs a new command instance with all previously configured properties.
*
* @param key must not be {@literal null}.
* @return a new {@link HIncrByCommand} with {@literal key} applied.
*/
public HIncrByCommand<T> forKey(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new HIncrByCommand<T>(key, field, value);
}
/**
* @return
*/
public T getValue() {
return value;
}
/**
* @return
*/
public ByteBuffer getField() {
return field;
}
}
/**
* Increment {@code value} of a hash {@code field} by the given {@code value}.
* Increment {@literal value} of a hash {@literal field} by the given {@literal value}.
*
* @param key must not be {@literal null}.
* @param field must not be {@literal null}.
@@ -228,18 +304,16 @@ public interface ReactiveNumberCommands {
*/
default <T extends Number> Mono<T> hIncrBy(ByteBuffer key, ByteBuffer field, T 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");
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 hIncrBy(Mono.just(HIncrByCommand.<T> incr(field).by(value).forKey(key))).next()
.map(NumericResponse::getOutput);
}
/**
* Increment {@code value} of a hash {@code field} by the given {@code value}.
* Increment {@literal value} of a hash {@literal field} by the given {@literal value}.
*
* @return
*/

View File

@@ -17,21 +17,33 @@ package org.springframework.data.redis.connection;
import java.io.Closeable;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Supplier;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.data.domain.Range;
import org.springframework.util.Assert;
import lombok.Data;
/**
* Redis connection using reactive infrastructure declaring entry points for reactive command execution.
* <p>
* {@link ReactiveRedisConnection} is typically implemented by a stateful object that requires to be {@link #close()
* closed} once it is no longer required.
* <p>
* Commands can be either executed by passing plain arguments like {@code key}, {@code value} or wrapped inside a
* command stream. Streaming command execution accepts {@link org.reactivestreams.Publisher} of a particular
* {@link Command}. Commands are executed at the time their emission.
* <p>
* Arguments are binary-safe by using {@link ByteBuffer} arguments. Expect {@link ByteBuffer} to be consumed by
* {@link ReactiveRedisConnection} invocation or during execution. Any {@link ByteBuffer} used as method parameter
* should not be altered after invocation.
*
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
* @see Command
* @see CommandResponse
* @see KeyCommand
*/
public interface ReactiveRedisConnection extends Closeable {
@@ -85,7 +97,7 @@ public interface ReactiveRedisConnection extends Closeable {
ReactiveHashCommands hashCommands();
/**
* Get {@link ReacktiveGeoCommands}
* Get {@link ReactiveGeoCommands}
*
* @return never {@literal null}.
*/
@@ -98,86 +110,48 @@ public interface ReactiveRedisConnection extends Closeable {
*/
ReactiveHyperLogLogCommands hyperLogLogCommands();
/**
* Base interface for Redis commands executed with a reactive infrastructure.
*
* @author Christoph Strobl
* @author Mark Paluch
*/
interface Command {
/**
* @return the key related to this command.
*/
ByteBuffer getKey();
/**
* @return command name as {@link String}.
*/
default String getName() {
return getClass().getSimpleName().replace("Command", "").toUpperCase();
}
static <T extends Command> Builder<T> create(Class<T> type) {
return new CommandBuilder<T>(type);
}
interface Builder<T extends Command> extends Consumer<Object> {
default Builder<T> forKey(String key) {
return forKey(key.getBytes(Charset.forName("UTF-8")));
}
default Builder<T> forKey(byte[] key) {
return forKey(ByteBuffer.wrap(key));
}
default Builder<T> forKey(ByteBuffer key) {
return forKey(() -> key);
}
Builder<T> forKey(Supplier<ByteBuffer> keySupplier);
T build();
}
class CommandBuilder<T extends Command> implements Builder<T> {
List<Object> argumentList = new ArrayList<>();
Class<T> type;
Supplier<ByteBuffer> key;
public CommandBuilder(Class<T> type) {
this.type = type;
}
public Builder<T> forKey(Supplier<ByteBuffer> key) {
this.key = key;
return this;
}
@Override
public void accept(Object t) {
argumentList.add(t);
}
@Override
public T build() {
try {
T x = BeanUtils.instantiateClass(type);
DirectFieldAccessor dfa = new DirectFieldAccessor(x);
dfa.setPropertyValue("key", key);
return x;
} catch (IllegalArgumentException | SecurityException e) {
throw new IllegalArgumentException(" ¯\\_(ツ)_/¯", e);
}
}
}
}
/**
* {@link Command} for key-bound operations.
*
* @author Christoph Strobl
*/
class KeyCommand implements Command {
private ByteBuffer key;
/**
* Creates a new {@link KeyCommand} given a {@code key}.
*
* @param key must not be {@literal null}.
*/
public KeyCommand(ByteBuffer key) {
this.key = key;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveRedisConnection.Command#getKey()
*/
@Override
public ByteBuffer getKey() {
return key;
@@ -191,33 +165,77 @@ public interface ReactiveRedisConnection extends Closeable {
Range<Long> range;
public RangeCommand(ByteBuffer key, Range<Long> range) {
/**
* Creates a new {@link RangeCommand} given a {@code key} and {@link Range}.
*
* @param key must not be {@literal null}.
* @param range may be {@literal null} if unbounded.
*/
private RangeCommand(ByteBuffer key, Range<Long> range) {
super(key);
this.range = range != null ? range : new Range<>(0L, Long.MAX_VALUE);
}
/**
* Creates a new {@link RangeCommand} given a {@code key}.
*
* @param key must not be {@literal null}.
* @return a new {@link RangeCommand} for {@code key}.
*/
public static RangeCommand key(ByteBuffer key) {
return new RangeCommand(key, null);
}
/**
* Applies a {@link Range}. Constructs a new command instance with all previously configured properties.
*
* @param range must not be {@literal null}.
* @return a new {@link RangeCommand} with {@link Range} applied.
*/
public RangeCommand within(Range<Long> range) {
Assert.notNull(range, "Range must not be null!");
return new RangeCommand(getKey(), range);
}
public RangeCommand fromIndex(Long start) {
/**
* Applies a lower bound to the {@link Range}. Constructs a new command instance with all previously configured
* properties.
*
* @param start
* @return a new {@link RangeCommand} with the lower bound applied.
*/
public RangeCommand fromIndex(long start) {
return new RangeCommand(getKey(), new Range<>(start, range.getUpperBound()));
}
public RangeCommand toIndex(Long end) {
/**
* Applies an upper bound to the {@link Range}. Constructs a new command instance with all previously configured
* properties.
*
* @param end
* @return a new {@link RangeCommand} with the upper bound applied.
*/
public RangeCommand toIndex(long end) {
return new RangeCommand(getKey(), new Range<>(range.getLowerBound(), end));
}
/**
* @return the {@link Range}.
*/
public Range<Long> getRange() {
return range;
}
}
/**
* Base class for command responses.
*
* @param <I> command input type.
* @param <O> command output type.
*/
@Data
class CommandResponse<I, O> {
@@ -225,6 +243,9 @@ public interface ReactiveRedisConnection extends Closeable {
private final O output;
}
/**
* {@link CommandResponse} implementation for {@link Boolean} responses.
*/
class BooleanResponse<I> extends CommandResponse<I, Boolean> {
public BooleanResponse(I input, Boolean output) {
@@ -232,6 +253,9 @@ public interface ReactiveRedisConnection extends Closeable {
}
}
/**
* {@link CommandResponse} implementation for {@link ByteBuffer} responses.
*/
class ByteBufferResponse<I> extends CommandResponse<I, ByteBuffer> {
public ByteBufferResponse(I input, ByteBuffer output) {
@@ -239,6 +263,9 @@ public interface ReactiveRedisConnection extends Closeable {
}
}
/**
* {@link CommandResponse} implementation for {@link List} responses.
*/
class MultiValueResponse<I, O> extends CommandResponse<I, List<O>> {
public MultiValueResponse(I input, List<O> output) {
@@ -246,11 +273,13 @@ public interface ReactiveRedisConnection extends Closeable {
}
}
/**
* {@link CommandResponse} implementation for {@link Number numeric} responses.
*/
class NumericResponse<I, O extends Number> extends CommandResponse<I, O> {
public NumericResponse(I input, O output) {
super(input, output);
}
}
}

View File

@@ -16,6 +16,8 @@
package org.springframework.data.redis.connection;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
@@ -33,12 +35,17 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* Redis Set commands executed using reactive infrastructure.
*
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public interface ReactiveSetCommands {
/**
* {@code SADD} command parameters.
*
* @author Christoph Strobl
*/
class SAddCommand extends KeyCommand {
@@ -48,28 +55,59 @@ public interface ReactiveSetCommands {
private SAddCommand(ByteBuffer key, List<ByteBuffer> values) {
super(key);
this.values = values;
}
public static SAddCommand value(ByteBuffer values) {
return values(Collections.singletonList(values));
/**
* Creates a new {@link SAddCommand} given a {@literal value}.
*
* @param value must not be {@literal null}.
* @return a new {@link SAddCommand} for a {@literal value}.
*/
public static SAddCommand value(ByteBuffer value) {
Assert.notNull(value, "Value must not be null!");
return values(Collections.singletonList(value));
}
public static SAddCommand values(List<ByteBuffer> values) {
return new SAddCommand(null, values);
/**
* Creates a new {@link SAddCommand} given a {@link Collection} of values.
*
* @param values must not be {@literal null}.
* @return a new {@link SAddCommand} for a {@link Collection} of values.
*/
public static SAddCommand values(Collection<ByteBuffer> values) {
Assert.notNull(values, "Values must not be null!");
return new SAddCommand(null, new ArrayList<>(values));
}
/**
* Applies the {@literal key}. Constructs a new command instance with all previously configured properties.
*
* @param key must not be {@literal null}.
* @return a new {@link SAddCommand} with {@literal key} applied.
*/
public SAddCommand to(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new SAddCommand(key, values);
}
/**
* @return
*/
public List<ByteBuffer> getValues() {
return values;
}
}
/**
* Add given {@code value} to set at {@code key}.
* Add given {@literal value} to set at {@literal key}.
*
* @param key must not be {@literal null}.
* @param value must not be {@literal null}.
@@ -77,22 +115,22 @@ public interface ReactiveSetCommands {
*/
default Mono<Long> sAdd(ByteBuffer key, ByteBuffer value) {
Assert.notNull(value, "value must not be null");
Assert.notNull(value, "Value must not be null!");
return sAdd(key, Collections.singletonList(value));
}
/**
* Add given {@code values} to set at {@code key}.
* Add given {@literal values} to set at {@literal key}.
*
* @param key must not be {@literal null}.
* @param values must not be {@literal null}.
* @return
*/
default Mono<Long> sAdd(ByteBuffer key, List<ByteBuffer> values) {
default Mono<Long> sAdd(ByteBuffer key, Collection<ByteBuffer> values) {
Assert.notNull(key, "key must not be null");
Assert.notNull(values, "values must not be null");
Assert.notNull(key, "Key must not be null!");
Assert.notNull(values, "Values must not be null!");
return sAdd(Mono.just(SAddCommand.values(values).to(key))).next().map(NumericResponse::getOutput);
}
@@ -106,37 +144,70 @@ public interface ReactiveSetCommands {
Flux<NumericResponse<SAddCommand, Long>> sAdd(Publisher<SAddCommand> commands);
/**
* {@code SREM} command parameters.
*
* @author Christoph Strobl
*/
class SRemCommand extends KeyCommand {
private final List<ByteBuffer> values;
public SRemCommand(ByteBuffer key, List<ByteBuffer> values) {
private SRemCommand(ByteBuffer key, List<ByteBuffer> values) {
super(key);
this.values = values;
}
public static SRemCommand value(ByteBuffer values) {
return values(Collections.singletonList(values));
/**
* Creates a new {@link SRemCommand} given a {@literal value}.
*
* @param value must not be {@literal null}.
* @return a new {@link SRemCommand} for a {@literal value}.
*/
public static SRemCommand value(ByteBuffer value) {
Assert.notNull(value, "Value must not be null!");
return values(Collections.singletonList(value));
}
public static SRemCommand values(List<ByteBuffer> values) {
return new SRemCommand(null, values);
/**
* Creates a new {@link SRemCommand} given a {@link Collection} of values.
*
* @param values must not be {@literal null}.
* @return a new {@link SRemCommand} for a {@link Collection} of values.
*/
public static SRemCommand values(Collection<ByteBuffer> values) {
Assert.notNull(values, "Values must not be null!");
return new SRemCommand(null, new ArrayList<>(values));
}
/**
* Applies the {@literal key}. Constructs a new command instance with all previously configured properties.
*
* @param key must not be {@literal null}.
* @return a new {@link SRemCommand} with {@literal key} applied.
*/
public SRemCommand from(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new SRemCommand(key, values);
}
/**
* @return
*/
public List<ByteBuffer> getValues() {
return values;
}
}
/**
* Remove given {@code value} from set at {@code key} and return the number of removed elements.
* Remove given {@literal value} from set at {@literal key} and return the number of removed elements.
*
* @param key must not be {@literal null}.
* @param value must not be {@literal null}.
@@ -144,22 +215,22 @@ public interface ReactiveSetCommands {
*/
default Mono<Long> sRem(ByteBuffer key, ByteBuffer value) {
Assert.notNull(value, "value must not be null");
Assert.notNull(value, "Value must not be null!");
return sRem(key, Collections.singletonList(value));
}
/**
* Remove given {@code values} from set at {@code key} and return the number of removed elements.
* Remove given {@literal values} from set at {@literal key} and return the number of removed elements.
*
* @param key must not be {@literal null}.
* @param values must not be {@literal null}.
* @return
*/
default Mono<Long> sRem(ByteBuffer key, List<ByteBuffer> values) {
default Mono<Long> sRem(ByteBuffer key, Collection<ByteBuffer> values) {
Assert.notNull(key, "key must not be null");
Assert.notNull(values, "values must not be null");
Assert.notNull(key, "Key must not be null!");
Assert.notNull(values, "Values must not be null!");
return sRem(Mono.just(SRemCommand.values(values).from(key))).next().map(NumericResponse::getOutput);
}
@@ -173,14 +244,14 @@ public interface ReactiveSetCommands {
Flux<NumericResponse<SRemCommand, Long>> sRem(Publisher<SRemCommand> commands);
/**
* Remove and return a random member from set at {@code key}.
* Remove and return a random member from set at {@literal key}.
*
* @param key must not be {@literal null}.
* @return
*/
default Mono<ByteBuffer> sPop(ByteBuffer key) {
Assert.notNull(key, "key must not be null");
Assert.notNull(key, "Key must not be null!");
return sPop(Mono.just(new KeyCommand(key))).next().map(ByteBufferResponse::getOutput);
}
@@ -194,6 +265,8 @@ public interface ReactiveSetCommands {
Flux<ByteBufferResponse<KeyCommand>> sPop(Publisher<KeyCommand> commands);
/**
* {@code SMOVE} command parameters.
*
* @author Christoph Strobl
*/
class SMoveCommand extends KeyCommand {
@@ -208,29 +281,63 @@ public interface ReactiveSetCommands {
this.value = value;
}
/**
* Creates a new {@link SMoveCommand} given a {@literal value}.
*
* @param value must not be {@literal null}.
* @return a new {@link SMoveCommand} for a {@literal value}.
*/
public static SMoveCommand value(ByteBuffer value) {
Assert.notNull(value, "Value must not be null!");
return new SMoveCommand(null, null, value);
}
/**
* Applies the {@literal source} key. Constructs a new command instance with all previously configured properties.
*
* @param source must not be {@literal null}.
* @return a new {@link SMoveCommand} with {@literal source} applied.
*/
public SMoveCommand from(ByteBuffer source) {
Assert.notNull(source, "Source key must not be null!");
return new SMoveCommand(source, destination, value);
}
/**
* Applies the {@literal destination} key. Constructs a new command instance with all previously configured
* properties.
*
* @param destination must not be {@literal null}.
* @return a new {@link SMoveCommand} with {@literal destination} applied.
*/
public SMoveCommand to(ByteBuffer destination) {
Assert.notNull(destination, "Destination key must not be null!");
return new SMoveCommand(getKey(), destination, value);
}
/**
* @return
*/
public ByteBuffer getDestination() {
return destination;
}
/**
* @return
*/
public ByteBuffer getValue() {
return value;
}
}
/**
* Move {@code value} from {@code sourceKey} to {@code destinationKey}
* Move {@literal value} from {@literal sourceKey} to {@literal destinationKey}
*
* @param sourceKey must not be {@literal null}.
* @param destinationKey must not be {@literal null}.
@@ -239,9 +346,9 @@ public interface ReactiveSetCommands {
*/
default Mono<Boolean> sMove(ByteBuffer sourceKey, ByteBuffer destinationKey, ByteBuffer value) {
Assert.notNull(sourceKey, "sourceKey must not be null");
Assert.notNull(destinationKey, "destinationKey must not be null");
Assert.notNull(value, "value must not be null");
Assert.notNull(sourceKey, "SourceKey must not be null!");
Assert.notNull(destinationKey, "DestinationKey must not be null!");
Assert.notNull(value, "Value must not be null!");
return sMove(Mono.just(SMoveCommand.value(value).from(sourceKey).to(destinationKey))).next()
.map(BooleanResponse::getOutput);
@@ -256,14 +363,14 @@ public interface ReactiveSetCommands {
Flux<BooleanResponse<SMoveCommand>> sMove(Publisher<SMoveCommand> commands);
/**
* Get size of set at {@code key}.
* Get size of set at {@literal key}.
*
* @param key must not be {@literal null}.
* @return
*/
default Mono<Long> sCard(ByteBuffer key) {
Assert.notNull(key, "key must not be null");
Assert.notNull(key, "Key must not be null!");
return sCard(Mono.just(new KeyCommand(key))).next().map(NumericResponse::getOutput);
}
@@ -277,6 +384,8 @@ public interface ReactiveSetCommands {
Flux<NumericResponse<KeyCommand, Long>> sCard(Publisher<KeyCommand> commands);
/**
* {@code SISMEMBER} command parameters.
*
* @author Christoph Strobl
*/
class SIsMemberCommand extends KeyCommand {
@@ -286,24 +395,46 @@ public interface ReactiveSetCommands {
private SIsMemberCommand(ByteBuffer key, ByteBuffer value) {
super(key);
this.value = value;
}
/**
* Creates a new {@link SIsMemberCommand} given a {@literal value}.
*
* @param value must not be {@literal null}.
* @return a new {@link SIsMemberCommand} for a {@literal value}.
*/
public static SIsMemberCommand value(ByteBuffer value) {
Assert.notNull(value, "Value must not be null!");
return new SIsMemberCommand(null, value);
}
/**
* Applies the {@literal set} key. Constructs a new command instance with all previously configured properties.
*
* @param set must not be {@literal null}.
* @return a new {@link SIsMemberCommand} with {@literal set} applied.
*/
public SIsMemberCommand of(ByteBuffer set) {
Assert.notNull(set, "Set key must not be null!");
return new SIsMemberCommand(set, value);
}
/**
* @return
*/
public ByteBuffer getValue() {
return value;
}
}
/**
* Check if set at {@code key} contains {@code value}.
* Check if set at {@literal key} contains {@literal value}.
*
* @param key must not be {@literal null}.
* @param value must not be {@literal null}.
@@ -311,8 +442,8 @@ public interface ReactiveSetCommands {
*/
default Mono<Boolean> sIsMember(ByteBuffer key, ByteBuffer value) {
Assert.notNull(key, "key must not be null");
Assert.notNull(value, "value must not be null");
Assert.notNull(key, "Key must not be null!");
Assert.notNull(value, "Value must not be null!");
return sIsMember(Mono.just(SIsMemberCommand.value(value).of(key))).next().map(BooleanResponse::getOutput);
}
@@ -326,6 +457,8 @@ public interface ReactiveSetCommands {
Flux<BooleanResponse<SIsMemberCommand>> sIsMember(Publisher<SIsMemberCommand> commands);
/**
* {@code SINTER} command parameters.
*
* @author Christoph Strobl
*/
class SInterCommand implements Command {
@@ -336,29 +469,44 @@ public interface ReactiveSetCommands {
this.keys = keys;
}
public static SInterCommand keys(List<ByteBuffer> keys) {
return new SInterCommand(keys);
/**
* Creates a new {@link SInterCommand} given a {@link Collection} of keys.
*
* @param keys must not be {@literal null}.
* @return a new {@link SInterCommand} for a {@link Collection} of values.
*/
public static SInterCommand keys(Collection<ByteBuffer> keys) {
Assert.notNull(keys, "Keys must not be null!");
return new SInterCommand(new ArrayList<>(keys));
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveRedisConnection.Command#getKey()
*/
@Override
public ByteBuffer getKey() {
return null;
}
/**
* @return
*/
public List<ByteBuffer> getKeys() {
return keys;
}
}
/**
* Returns the members intersecting all given sets at {@code keys}.
* Returns the members intersecting all given sets at {@literal keys}.
*
* @param keys must not be {@literal null}.
* @return
*/
default Mono<List<ByteBuffer>> sInter(List<ByteBuffer> keys) {
default Mono<List<ByteBuffer>> sInter(Collection<ByteBuffer> keys) {
Assert.notNull(keys, "keys must not be null");
Assert.notNull(keys, "Keys must not be null!");
return sInter(Mono.just(SInterCommand.keys(keys))).next().map(MultiValueResponse::getOutput);
}
@@ -372,6 +520,8 @@ public interface ReactiveSetCommands {
Flux<MultiValueResponse<SInterCommand, ByteBuffer>> sInter(Publisher<SInterCommand> commands);
/**
* {@code SINTERSTORE} command parameters.
*
* @author Christoph Strobl
*/
class SInterStoreCommand extends KeyCommand {
@@ -381,40 +531,63 @@ public interface ReactiveSetCommands {
private SInterStoreCommand(ByteBuffer key, List<ByteBuffer> keys) {
super(key);
this.keys = keys;
}
public static SInterStoreCommand keys(List<ByteBuffer> keys) {
return new SInterStoreCommand(null, keys);
/**
* Creates a new {@link SInterStoreCommand} given a {@link Collection} of keys.
*
* @param keys must not be {@literal null}.
* @return a new {@link SInterStoreCommand} for a {@link Collection} of values.
*/
public static SInterStoreCommand keys(Collection<ByteBuffer> keys) {
Assert.notNull(keys, "Keys must not be null!");
return new SInterStoreCommand(null, new ArrayList<>(keys));
}
/**
* Applies the {@literal key} at which the result is stored. Constructs a new command instance with all previously
* configured properties.
*
* @param key must not be {@literal null}.
* @return a new {@link SInterStoreCommand} with {@literal key} applied.
*/
public SInterStoreCommand storeAt(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new SInterStoreCommand(key, keys);
}
/**
* @return
*/
public List<ByteBuffer> getKeys() {
return keys;
}
}
/**
* Intersect all given sets at {@code keys} and store result in {@code destinationKey}.
* Intersect all given sets at {@literal keys} and store result in {@literal destinationKey}.
*
* @param destinationKey must not be {@literal null}.
* @param keys must not be {@literal null}.
* @return size of set stored a {@code destinationKey}.
* @return size of set stored a {@literal destinationKey}.
*/
default Mono<Long> sInterStore(ByteBuffer destinationKey, List<ByteBuffer> keys) {
default Mono<Long> sInterStore(ByteBuffer destinationKey, Collection<ByteBuffer> keys) {
Assert.notNull(destinationKey, "destinationKey must not be null");
Assert.notNull(keys, "keys must not be null");
Assert.notNull(destinationKey, "DestinationKey must not be null!");
Assert.notNull(keys, "Keys must not be null!");
return sInterStore(Mono.just(SInterStoreCommand.keys(keys).storeAt(destinationKey))).next()
.map(NumericResponse::getOutput);
}
/**
* Intersect all given sets at {@code keys} and store result in {@code destinationKey}.
* Intersect all given sets at {@literal keys} and store result in {@literal destinationKey}.
*
* @param commands must not be {@literal null}.
* @return
@@ -422,6 +595,8 @@ public interface ReactiveSetCommands {
Flux<NumericResponse<SInterStoreCommand, Long>> sInterStore(Publisher<SInterStoreCommand> commands);
/**
* {@code SUNION} command parameters.
*
* @author Christoph Strobl
*/
class SUnionCommand implements Command {
@@ -432,29 +607,44 @@ public interface ReactiveSetCommands {
this.keys = keys;
}
public static SUnionCommand keys(List<ByteBuffer> keys) {
return new SUnionCommand(keys);
/**
* Creates a new {@link SUnionCommand} given a {@link Collection} of keys.
*
* @param keys must not be {@literal null}.
* @return a new {@link SUnionCommand} for a {@link Collection} of values.
*/
public static SUnionCommand keys(Collection<ByteBuffer> keys) {
Assert.notNull(keys, "Keys must not be null!");
return new SUnionCommand(new ArrayList<>(keys));
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveRedisConnection.Command#getKey()
*/
@Override
public ByteBuffer getKey() {
return null;
}
/**
* @return
*/
public List<ByteBuffer> getKeys() {
return keys;
}
}
/**
* Returns the members intersecting all given sets at {@code keys}.
* Returns the members intersecting all given sets at {@literal keys}.
*
* @param keys must not be {@literal null}.
* @return
*/
default Mono<List<ByteBuffer>> sUnion(List<ByteBuffer> keys) {
default Mono<List<ByteBuffer>> sUnion(Collection<ByteBuffer> keys) {
Assert.notNull(keys, "keys must not be null");
Assert.notNull(keys, "Keys must not be null!");
return sUnion(Mono.just(SUnionCommand.keys(keys))).next().map(MultiValueResponse::getOutput);
}
@@ -477,40 +667,63 @@ public interface ReactiveSetCommands {
private SUnionStoreCommand(ByteBuffer key, List<ByteBuffer> keys) {
super(key);
this.keys = keys;
}
public static SUnionStoreCommand keys(List<ByteBuffer> keys) {
return new SUnionStoreCommand(null, keys);
/**
* Creates a new {@link SUnionStoreCommand} given a {@link Collection} of keys.
*
* @param keys must not be {@literal null}.
* @return a new {@link SUnionStoreCommand} for a {@link Collection} of values.
*/
public static SUnionStoreCommand keys(Collection<ByteBuffer> keys) {
Assert.notNull(keys, "Keys must not be null!");
return new SUnionStoreCommand(null, new ArrayList<>(keys));
}
/**
* Applies the {@literal key} at which the result is stored. Constructs a new command instance with all previously
* configured properties.
*
* @param key must not be {@literal null}.
* @return a new {@link SUnionStoreCommand} with {@literal key} applied.
*/
public SUnionStoreCommand storeAt(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new SUnionStoreCommand(key, keys);
}
/**
* @return
*/
public List<ByteBuffer> getKeys() {
return keys;
}
}
/**
* Union all given sets at {@code keys} and store result in {@code destinationKey}.
* Union all given sets at {@literal keys} and store result in {@literal destinationKey}.
*
* @param destinationKey must not be {@literal null}.
* @param keys must not be {@literal null}.
* @return size of set stored a {@code destinationKey}.
* @return size of set stored a {@literal destinationKey}.
*/
default Mono<Long> sUnionStore(ByteBuffer destinationKey, List<ByteBuffer> keys) {
default Mono<Long> sUnionStore(ByteBuffer destinationKey, Collection<ByteBuffer> keys) {
Assert.notNull(destinationKey, "destinationKey must not be null");
Assert.notNull(keys, "keys must not be null");
Assert.notNull(destinationKey, "DestinationKey must not be null!");
Assert.notNull(keys, "Keys must not be null!");
return sUnionStore(Mono.just(SUnionStoreCommand.keys(keys).storeAt(destinationKey))).next()
.map(NumericResponse::getOutput);
}
/**
* Union all given sets at {@code keys} and store result in {@code destinationKey}.
* Union all given sets at {@literal keys} and store result in {@literal destinationKey}.
*
* @param commands must not be {@literal null}.
* @return
@@ -518,6 +731,8 @@ public interface ReactiveSetCommands {
Flux<NumericResponse<SUnionStoreCommand, Long>> sUnionStore(Publisher<SUnionStoreCommand> commands);
/**
* {@code SDIFF} command parameters.
*
* @author Christoph Strobl
*/
class SDiffCommand implements Command {
@@ -528,29 +743,44 @@ public interface ReactiveSetCommands {
this.keys = keys;
}
public static SDiffCommand keys(List<ByteBuffer> keys) {
return new SDiffCommand(keys);
/**
* Creates a new {@link SDiffCommand} given a {@link Collection} of keys.
*
* @param keys must not be {@literal null}.
* @return a new {@link SDiffCommand} for a {@link Collection} of values.
*/
public static SDiffCommand keys(Collection<ByteBuffer> keys) {
Assert.notNull(keys, "Keys must not be null!");
return new SDiffCommand(new ArrayList<>(keys));
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveRedisConnection.Command#getKey()
*/
@Override
public ByteBuffer getKey() {
return null;
}
/**
* @return
*/
public List<ByteBuffer> getKeys() {
return keys;
}
}
/**
* Returns the diff of the members of all given sets at {@code keys}.
* Returns the diff of the members of all given sets at {@literal keys}.
*
* @param keys must not be {@literal null}.
* @return
*/
default Mono<List<ByteBuffer>> sDiff(List<ByteBuffer> keys) {
default Mono<List<ByteBuffer>> sDiff(Collection<ByteBuffer> keys) {
Assert.notNull(keys, "keys must not be null");
Assert.notNull(keys, "Keys must not be null!");
return sDiff(Mono.just(SDiffCommand.keys(keys))).next().map(MultiValueResponse::getOutput);
}
@@ -564,6 +794,8 @@ public interface ReactiveSetCommands {
Flux<MultiValueResponse<SDiffCommand, ByteBuffer>> sDiff(Publisher<SDiffCommand> commands);
/**
* {@code SDIFFSTORE} command parameters.
*
* @author Christoph Strobl
*/
class SDiffStoreCommand extends KeyCommand {
@@ -573,40 +805,63 @@ public interface ReactiveSetCommands {
private SDiffStoreCommand(ByteBuffer key, List<ByteBuffer> keys) {
super(key);
this.keys = keys;
}
public static SDiffStoreCommand keys(List<ByteBuffer> keys) {
return new SDiffStoreCommand(null, keys);
/**
* Creates a new {@link SDiffStoreCommand} given a {@link Collection} of keys.
*
* @param keys must not be {@literal null}.
* @return a new {@link SDiffStoreCommand} for a {@link Collection} of values.
*/
public static SDiffStoreCommand keys(Collection<ByteBuffer> keys) {
Assert.notNull(keys, "Keys must not be null!");
return new SDiffStoreCommand(null, new ArrayList<>(keys));
}
/**
* Applies the {@literal key} at which the result is stored. Constructs a new command instance with all previously
* configured properties.
*
* @param key must not be {@literal null}.
* @return a new {@link SDiffStoreCommand} with {@literal key} applied.
*/
public SDiffStoreCommand storeAt(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new SDiffStoreCommand(key, keys);
}
/**
* @return
*/
public List<ByteBuffer> getKeys() {
return keys;
}
}
/**
* Diff all given sets at {@code keys} and store result in {@code destinationKey}.
* Diff all given sets at {@literal keys} and store result in {@literal destinationKey}.
*
* @param destinationKey must not be {@literal null}.
* @param keys must not be {@literal null}.
* @return size of set stored a {@code destinationKey}.
* @return size of set stored a {@literal destinationKey}.
*/
default Mono<Long> sDiffStore(ByteBuffer destinationKey, List<ByteBuffer> keys) {
default Mono<Long> sDiffStore(ByteBuffer destinationKey, Collection<ByteBuffer> keys) {
Assert.notNull(destinationKey, "destinationKey must not be null");
Assert.notNull(keys, "keys must not be null");
Assert.notNull(destinationKey, "DestinationKey must not be null!");
Assert.notNull(keys, "Keys must not be null!");
return sDiffStore(Mono.just(SDiffStoreCommand.keys(keys).storeAt(destinationKey))).next()
.map(NumericResponse::getOutput);
}
/**
* Diff all given sets at {@code keys} and store result in {@code destinationKey}.
* Diff all given sets at {@literal keys} and store result in {@literal destinationKey}.
*
* @param commands must not be {@literal null}.
* @return
@@ -614,14 +869,14 @@ public interface ReactiveSetCommands {
Flux<NumericResponse<SDiffStoreCommand, Long>> sDiffStore(Publisher<SDiffStoreCommand> commands);
/**
* Get all elements of set at {@code key}.
* Get all elements of set at {@literal key}.
*
* @param key must not be {@literal null}.
* @return
*/
default Mono<List<ByteBuffer>> sMembers(ByteBuffer key) {
Assert.notNull(key, "key must not be null");
Assert.notNull(key, "Key must not be null!");
return sMembers(Mono.just(new KeyCommand(key))).next().map(MultiValueResponse::getOutput);
}
@@ -635,6 +890,8 @@ public interface ReactiveSetCommands {
Flux<MultiValueResponse<KeyCommand, ByteBuffer>> sMembers(Publisher<KeyCommand> commands);
/**
* {@code SRANDMEMBER} command parameters.
*
* @author Christoph Strobl
*/
class SRandMembersCommand extends KeyCommand {
@@ -647,25 +904,48 @@ public interface ReactiveSetCommands {
this.count = count;
}
public static SRandMembersCommand valueCount(Long nrValuesToRetrieve) {
/**
* Creates a new {@link SRandMembersCommand} given the number of values to retrieve.
*
* @param nrValuesToRetrieve
* @return a new {@link SRandMembersCommand} for a number of values to retrieve.
*/
public static SRandMembersCommand valueCount(long nrValuesToRetrieve) {
return new SRandMembersCommand(null, nrValuesToRetrieve);
}
/**
* Creates a new {@link SRandMembersCommand} to retrieve one random member.
*
* @return a new {@link SRandMembersCommand} to retrieve one random member.
*/
public static SRandMembersCommand singleValue() {
return new SRandMembersCommand(null, null);
}
/**
* Applies the {@literal key}. Constructs a new command instance with all previously configured properties.
*
* @param key must not be {@literal null}.
* @return a new {@link SRandMembersCommand} with {@literal key} applied.
*/
public SRandMembersCommand from(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new SRandMembersCommand(key, count);
}
/**
* @return
*/
public Optional<Long> getCount() {
return Optional.ofNullable(count);
}
}
/**
* Get random element from set at {@code key}.
* Get random element from set at {@literal key}.
*
* @param key must not be {@literal null}.
* @return
@@ -675,7 +955,7 @@ public interface ReactiveSetCommands {
}
/**
* Get {@code count} random elements from set at {@code key}.
* Get {@literal count} random elements from set at {@literal key}.
*
* @param key must not be {@literal null}.
* @param count must not be {@literal null}.
@@ -683,8 +963,8 @@ public interface ReactiveSetCommands {
*/
default Mono<List<ByteBuffer>> sRandMember(ByteBuffer key, Long count) {
Assert.notNull(key, "key must not be null");
Assert.notNull(count, "count must not be null");
Assert.notNull(key, "Key must not be null!");
Assert.notNull(count, "Count must not be null!");
return sRandMember(Mono.just(SRandMembersCommand.valueCount(count).from(key))).next()
.map(MultiValueResponse::getOutput);

View File

@@ -16,6 +16,8 @@
package org.springframework.data.redis.connection;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -25,6 +27,7 @@ import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.ReactiveRedisConnection.BooleanResponse;
import org.springframework.data.redis.connection.ReactiveRedisConnection.ByteBufferResponse;
import org.springframework.data.redis.connection.ReactiveRedisConnection.Command;
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;
@@ -38,12 +41,17 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* Redis String commands executed using reactive infrastructure.
*
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public interface ReactiveStringCommands {
/**
* {@code SET} command parameters.
*
* @author Christoph Strobl
*/
class SetCommand extends KeyCommand {
@@ -55,35 +63,81 @@ public interface ReactiveStringCommands {
private SetCommand(ByteBuffer key, ByteBuffer value, Expiration expiration, SetOption option) {
super(key);
this.value = value;
this.expiration = expiration;
this.option = option;
}
public static ReactiveStringCommands.SetCommand set(ByteBuffer key) {
/**
* Creates a new {@link SetCommand} given a {@literal key}.
*
* @param key must not be {@literal null}.
* @return a new {@link SetCommand} for a {@literal key}.
*/
public static SetCommand set(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new SetCommand(key, null, null, null);
}
public ReactiveStringCommands.SetCommand value(ByteBuffer value) {
/**
* Applies the {@literal value}. Constructs a new command instance with all previously configured properties.
*
* @param value must not be {@literal null}.
* @return a new {@link SetCommand} with {@literal value} applied.
*/
public SetCommand value(ByteBuffer value) {
Assert.notNull(value, "Value must not be null!");
return new SetCommand(getKey(), value, expiration, option);
}
public ReactiveStringCommands.SetCommand expiring(Expiration expiration) {
/**
* Applies {@link Expiration}. Constructs a new command instance with all previously configured properties.
*
* @param expiration must not be {@literal null}.
* @return a new {@link SetCommand} with {@link Expiration} applied.
*/
public SetCommand expiring(Expiration expiration) {
Assert.notNull(expiration, "Expiration must not be null!");
return new SetCommand(getKey(), value, expiration, option);
}
public ReactiveStringCommands.SetCommand withSetOption(SetOption option) {
/**
* Applies {@link SetOption}. Constructs a new command instance with all previously configured properties.
*
* @param option must not be {@literal null}.
* @return a new {@link SetCommand} with {@link SetOption} applied.
*/
public SetCommand withSetOption(SetOption option) {
Assert.notNull(option, "SetOption must not be null!");
return new SetCommand(getKey(), value, expiration, option);
}
/**
* @return
*/
public ByteBuffer getValue() {
return value;
}
/**
* @return
*/
public Optional<Expiration> getExpiration() {
return Optional.ofNullable(expiration);
}
/**
* @return
*/
public Optional<SetOption> getOption() {
return Optional.ofNullable(option);
}
@@ -123,12 +177,12 @@ public interface ReactiveStringCommands {
}
/**
* Set each and every {@link KeyValue} item separately.
* Set each and every item separately by invoking {@link SetCommand}.
*
* @param values must not be {@literal null}.
* @return {@link Flux} of {@link SetResponse} holding the {@link KeyValue} pair to set along with the command result.
* @param commands must not be {@literal null}.
* @return {@link Flux} of {@link BooleanResponse} holding the {@link SetCommand} along with the command result.
*/
Flux<BooleanResponse<ReactiveStringCommands.SetCommand>> set(Publisher<ReactiveStringCommands.SetCommand> commands);
Flux<BooleanResponse<SetCommand>> set(Publisher<SetCommand> commands);
/**
* Get single element stored at {@literal key}.
@@ -140,14 +194,15 @@ public interface ReactiveStringCommands {
Assert.notNull(key, "Key must not be null!");
return get(Mono.just(new KeyCommand(key))).next().map((result) -> result.getOutput());
return get(Mono.just(new KeyCommand(key))).next().map(CommandResponse::getOutput);
}
/**
* Get elements one by one.
*
* @param keys must not be {@literal null}.
* @return {@link Flux} of {@link GetResponse} holding the {@literal key} to get along with the value retrieved.
* @return {@link Flux} of {@link ByteBufferResponse} holding the {@literal key} to get along with the value
* retrieved.
*/
Flux<ByteBufferResponse<KeyCommand>> get(Publisher<KeyCommand> keys);
@@ -169,13 +224,11 @@ public interface ReactiveStringCommands {
/**
* Set {@literal value} for {@literal key} and return the existing value one by one.
*
* @param key must not be {@literal null}.
* @param value must not be {@literal null}.
* @return {@link Flux} of {@link GetSetResponse} holding the {@link KeyValue} pair to set along with the previously
* @param commands must not be {@literal null}.
* @return {@link Flux} of {@link ByteBufferResponse} holding the {@link SetCommand} along with the previously
* existing value.
*/
Flux<ByteBufferResponse<ReactiveStringCommands.SetCommand>> getSet(
Publisher<ReactiveStringCommands.SetCommand> command);
Flux<ByteBufferResponse<SetCommand>> getSet(Publisher<SetCommand> commands);
/**
* Get multiple values in one batch.
@@ -191,15 +244,15 @@ public interface ReactiveStringCommands {
}
/**
* Get multiple values at in batches.
* Get multiple values at for {@literal keysets} in batches.
*
* @param keys must not be {@literal null}.
* @param keysets must not be {@literal null}.
* @return
*/
Flux<MultiValueResponse<List<ByteBuffer>, ByteBuffer>> mGet(Publisher<List<ByteBuffer>> keysets);
/**
* Set {@code value} for {@code key}, only if {@code key} does not exist.
* Set {@literal value} for {@literal key}, only if {@literal key} does not exist.
*
* @param key must not be {@literal null}.
* @param value must not be {@literal null}.
@@ -207,22 +260,22 @@ public interface ReactiveStringCommands {
*/
default Mono<Boolean> setNX(ByteBuffer key, ByteBuffer value) {
Assert.notNull(key, "Keys must not be null!");
Assert.notNull(value, "Keys must not be null!");
Assert.notNull(key, "Key must not be null!");
Assert.notNull(value, "Value must not be null!");
return setNX(Mono.just(SetCommand.set(key).value(value))).next().map(BooleanResponse::getOutput);
}
/**
* Set {@code key value} pairs, only if {@code key} does not exist.
* Set {@literal key value} pairs, only if {@literal key} does not exist.
*
* @param values must not be {@literal null}.
* @return
*/
Flux<BooleanResponse<ReactiveStringCommands.SetCommand>> setNX(Publisher<ReactiveStringCommands.SetCommand> values);
Flux<BooleanResponse<SetCommand>> setNX(Publisher<SetCommand> values);
/**
* Set {@code key value} pair and {@link Expiration}.
* Set {@literal key value} pair and {@link Expiration}.
*
* @param key must not be {@literal null}.
* @param value must not be {@literal null}.
@@ -233,23 +286,22 @@ public interface ReactiveStringCommands {
Assert.notNull(key, "Keys must not be null!");
Assert.notNull(value, "Keys must not be null!");
Assert.notNull(key, "ExpireTimeout must not be null!");
Assert.notNull(expireTimeout, "ExpireTimeout must not be null!");
return setEX(Mono.just(SetCommand.set(key).value(value).expiring(expireTimeout))).next()
.map(BooleanResponse::getOutput);
}
/**
* Set {@code key value} pairs and {@link Expiration}.
* Set {@literal key value} pairs and {@link Expiration}.
*
* @param source must not be {@literal null}.
* @param expireTimeout must not be {@literal null}.
* @param commands must not be {@literal null}.
* @return
*/
Flux<BooleanResponse<ReactiveStringCommands.SetCommand>> setEX(Publisher<ReactiveStringCommands.SetCommand> command);
Flux<BooleanResponse<SetCommand>> setEX(Publisher<SetCommand> commands);
/**
* Set {@code key value} pair and {@link Expiration}.
* Set {@literal key value} pair and {@link Expiration}.
*
* @param key must not be {@literal null}.
* @param value must not be {@literal null}.
@@ -267,15 +319,16 @@ public interface ReactiveStringCommands {
}
/**
* Set {@code key value} pairs and {@link Expiration}.
* Set {@literal key value} pairs and {@link Expiration}.
*
* @param source must not be {@literal null}.
* @param expireTimeout must not be {@literal null}.
* @param commands must not be {@literal null}.
* @return
*/
Flux<BooleanResponse<ReactiveStringCommands.SetCommand>> pSetEX(Publisher<ReactiveStringCommands.SetCommand> command);
Flux<BooleanResponse<SetCommand>> pSetEX(Publisher<SetCommand> commands);
/**
* {@code MSET} command parameters.
*
* @author Christoph Strobl
*/
class MSetCommand implements Command {
@@ -286,66 +339,82 @@ public interface ReactiveStringCommands {
this.keyValuePairs = keyValuePairs;
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveRedisConnection.Command#getKey()
*/
@Override
public ByteBuffer getKey() {
return null;
}
public static ReactiveStringCommands.MSetCommand mset(Map<ByteBuffer, ByteBuffer> keyValuePairs) {
/**
* Creates a new {@link MSetCommand} given a {@link Map} of key-value tuples.
*
* @param keyValuePairs must not be {@literal null}.
* @return a new {@link MSetCommand} for a {@link Map} of key-value tuples.
*/
public static MSetCommand mset(Map<ByteBuffer, ByteBuffer> keyValuePairs) {
Assert.notNull(keyValuePairs, "Key-value pairs must not be null!");
return new MSetCommand(keyValuePairs);
}
/**
* @return
*/
public Map<ByteBuffer, ByteBuffer> getKeyValuePairs() {
return keyValuePairs;
}
}
/**
* Set multiple keys to multiple values using key-value pairs provided in {@code tuple}.
* Set multiple keys to multiple values using key-value pairs provided in {@literal tuple}.
*
* @param tuples must not be {@literal null}.
* @param keyValuePairs must not be {@literal null}.
* @return
*/
default Mono<Boolean> mSet(Map<ByteBuffer, ByteBuffer> tuples) {
default Mono<Boolean> mSet(Map<ByteBuffer, ByteBuffer> keyValuePairs) {
Assert.notNull(tuples, "Tuples must not be null!");
Assert.notNull(keyValuePairs, "Key-value pairs must not be null!");
return mSet(Mono.just(MSetCommand.mset(tuples))).next().map(BooleanResponse::getOutput);
return mSet(Mono.just(MSetCommand.mset(keyValuePairs))).next().map(BooleanResponse::getOutput);
}
/**
* Set multiple keys to multiple values using key-value pairs provided in {@code source}.
* Set multiple keys to multiple values using key-value pairs provided in {@literal commands}.
*
* @param commands must not be {@literal null}.
* @return
*/
Flux<BooleanResponse<MSetCommand>> mSet(Publisher<MSetCommand> commands);
/**
* Set multiple keys to multiple values using key-value pairs provided in {@literal keyValuePairs} only if the
* provided key does not exist.
*
* @param keyValuePairs must not be {@literal null}.
* @return
*/
default Mono<Boolean> mSetNX(Map<ByteBuffer, ByteBuffer> keyValuePairs) {
Assert.notNull(keyValuePairs, "Key-value pairs must not be null!");
return mSetNX(Mono.just(MSetCommand.mset(keyValuePairs))).next().map(BooleanResponse::getOutput);
}
/**
* Set multiple keys to multiple values using key-value pairs provided in {@literal tuples} only if the provided key
* does not exist.
*
* @param source must not be {@literal null}.
* @return
*/
Flux<BooleanResponse<ReactiveStringCommands.MSetCommand>> mSet(Publisher<ReactiveStringCommands.MSetCommand> source);
Flux<BooleanResponse<MSetCommand>> mSetNX(Publisher<MSetCommand> source);
/**
* Set multiple keys to multiple values using key-value pairs provided in {@code tuples} only if the provided key does
* not exist.
* {@code APPEND} command parameters.
*
* @param tuples must not be {@literal null}.
* @return
*/
default Mono<Boolean> mSetNX(Map<ByteBuffer, ByteBuffer> tuples) {
Assert.notNull(tuples, "Tuples must not be null!");
return mSetNX(Mono.just(MSetCommand.mset(tuples))).next().map(BooleanResponse::getOutput);
}
/**
* Set multiple keys to multiple values using key-value pairs provided in {@code tuples} only if the provided key does
* not exist.
*
* @param source must not be {@literal null}.
* @return
*/
Flux<BooleanResponse<ReactiveStringCommands.MSetCommand>> mSetNX(
Publisher<ReactiveStringCommands.MSetCommand> source);
/**
* @author Christoph Strobl
*/
class AppendCommand extends KeyCommand {
@@ -358,22 +427,43 @@ public interface ReactiveStringCommands {
this.value = value;
}
public static ReactiveStringCommands.AppendCommand key(ByteBuffer key) {
/**
* Creates a new {@link AppendCommand} given a {@literal key}.
*
* @param key must not be {@literal null}.
* @return a new {@link AppendCommand} for a {@literal key}.
*/
public static AppendCommand key(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new AppendCommand(key, null);
}
public ReactiveStringCommands.AppendCommand append(ByteBuffer value) {
/**
* Applies the {@literal value} to append. Constructs a new command instance with all previously configured
* properties.
*
* @param value must not be {@literal null}.
* @return a new {@link AppendCommand} with {@literal value} applied.
*/
public AppendCommand append(ByteBuffer value) {
Assert.notNull(value, "Value must not be null!");
return new AppendCommand(getKey(), value);
}
/**
* @return
*/
public ByteBuffer getValue() {
return value;
}
}
/**
* Append a {@code value} to {@code key}.
* Append a {@literal value} to {@literal key}.
*
* @param key must not be {@literal null}.
* @param value must not be {@literal null}.
@@ -388,16 +478,15 @@ public interface ReactiveStringCommands {
}
/**
* Append a {@link KeyValue#value} to {@link KeyValue#key}
* Append a {@link AppendCommand#getValue()} to the {@link AppendCommand#getKey()}.
*
* @param source must not be {@literal null}.
* @param commands must not be {@literal null}.
* @return
*/
Flux<NumericResponse<ReactiveStringCommands.AppendCommand, Long>> append(
Publisher<ReactiveStringCommands.AppendCommand> source);
Flux<NumericResponse<AppendCommand, Long>> append(Publisher<AppendCommand> commands);
/**
* Get a substring of value of {@code key} between {@code begin} and {@code end}.
* Get a substring of value of {@literal key} between {@literal begin} and {@literal end}.
*
* @param key must not be {@literal null}.
* @param begin
@@ -408,21 +497,22 @@ public interface ReactiveStringCommands {
Assert.notNull(key, "Key must not be null!");
return getRange(Mono.just(RangeCommand.key(key).fromIndex(begin).toIndex(end))).next()
return getRange(Mono.just(RangeCommand.key(key).fromIndex(begin).toIndex(end))) //
.next() //
.map(ByteBufferResponse::getOutput);
}
/**
* Get a substring of value of {@code key} between {@code begin} and {@code end}.
* Get a substring of value of {@literal key} between {@literal begin} and {@literal end}.
*
* @param keys must not be {@literal null}.
* @param begin
* @param end
* @param commands must not be {@literal null}.
* @return
*/
Flux<ByteBufferResponse<RangeCommand>> getRange(Publisher<RangeCommand> commands);
/**
* {@code SETRANGE} command parameters.
*
* @author Christoph Strobl
*/
class SetRangeCommand extends KeyCommand {
@@ -437,29 +527,59 @@ public interface ReactiveStringCommands {
this.offset = offset;
}
public static ReactiveStringCommands.SetRangeCommand overwrite(ByteBuffer key) {
/**
* Creates a new {@link SetRangeCommand} given a {@literal key}.
*
* @param key must not be {@literal null}.
* @return a new {@link SetRangeCommand} for a {@literal key}.
*/
public static SetRangeCommand overwrite(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new SetRangeCommand(key, null, null);
}
public ReactiveStringCommands.SetRangeCommand withValue(ByteBuffer value) {
/**
* Applies the {@literal value}. Constructs a new command instance with all previously configured properties.
*
* @param value must not be {@literal null}.
* @return a new {@link SetCommand} with {@literal value} applied.
*/
public SetRangeCommand withValue(ByteBuffer value) {
Assert.notNull(value, "Value must not be null!");
return new SetRangeCommand(getKey(), value, offset);
}
public ReactiveStringCommands.SetRangeCommand atPosition(Long index) {
/**
* Applies the {@literal index}. Constructs a new command instance with all previously configured properties.
*
* @param index
* @return a new {@link SetRangeCommand} with {@literal key} applied.
*/
public SetRangeCommand atPosition(long index) {
return new SetRangeCommand(getKey(), value, index);
}
/**
* @return
*/
public ByteBuffer getValue() {
return value;
}
/**
* @return
*/
public Long getOffset() {
return offset;
}
}
/**
* Overwrite parts of {@code key} starting at the specified {@code offset} with given {@code value}.
* Overwrite parts of {@literal key} starting at the specified {@literal offset} with given {@literal value}.
*
* @param key must not be {@literal null}.
* @param value must not be {@literal null}.
@@ -476,40 +596,63 @@ public interface ReactiveStringCommands {
}
/**
* Overwrite parts of {@link KeyValue#key} starting at the specified {@code offset} with given {@link KeyValue#value}.
* Overwrite parts of {@link SetRangeCommand#key} starting at the specified {@literal offset} with given
* {@link SetRangeCommand#value}.
*
* @param keys must not be {@literal null}.
* @param offset must not be {@literal null}.
* @param commands must not be {@literal null}.
* @return
*/
Flux<NumericResponse<ReactiveStringCommands.SetRangeCommand, Long>> setRange(
Publisher<ReactiveStringCommands.SetRangeCommand> commands);
Flux<NumericResponse<SetRangeCommand, Long>> setRange(Publisher<SetRangeCommand> commands);
/**
* {@code GETBIT} command parameters.
*
* @author Christoph Strobl
*/
class GetBitCommand extends KeyCommand {
public Long offset;
private Long offset;
public GetBitCommand(ByteBuffer key, Long offset) {
private GetBitCommand(ByteBuffer key, Long offset) {
super(key);
this.offset = offset;
}
public static ReactiveStringCommands.GetBitCommand bit(ByteBuffer key) {
/**
* Creates a new {@link GetBitCommand} given a {@literal key}.
*
* @param key must not be {@literal null}.
* @return a new {@link GetBitCommand} for a {@literal key}.
*/
public static GetBitCommand bit(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new GetBitCommand(key, null);
}
public ReactiveStringCommands.GetBitCommand atOffset(Long offset) {
/**
* Applies the offset {@literal index}. Constructs a new command instance with all previously configured properties.
*
* @param offset
* @return a new {@link GetBitCommand} with {@literal offset} applied.
*/
public GetBitCommand atOffset(long offset) {
return new GetBitCommand(getKey(), offset);
}
/**
* @return
*/
public Long getOffset() {
return offset;
}
}
/**
* Get the bit value at {@code offset} of value at {@code key}.
* Get the bit value at {@literal offset} of value at {@literal key}.
*
* @param key must not be {@literal null}.
* @param offset
@@ -523,50 +666,81 @@ public interface ReactiveStringCommands {
}
/**
* Get the bit value at {@code offset} of value at {@code key}.
* Get the bit value at {@literal offset} of value at {@literal key}.
*
* @param keys must not be {@literal null}.
* @param offset must not be {@literal null}.
* @param commands must not be {@literal null}.
* @return
*/
Flux<BooleanResponse<ReactiveStringCommands.GetBitCommand>> getBit(
Publisher<ReactiveStringCommands.GetBitCommand> commands);
Flux<BooleanResponse<GetBitCommand>> getBit(Publisher<GetBitCommand> commands);
/**
* {@code SETBIT} command parameters.
*
* @author Christoph Strobl
*/
class SetBitCommand extends KeyCommand {
private Long offset;
private Boolean value;
private boolean value;
private SetBitCommand(ByteBuffer key, Long offset, Boolean value) {
private SetBitCommand(ByteBuffer key, Long offset, boolean value) {
super(key);
this.offset = offset;
this.value = value;
}
public static ReactiveStringCommands.SetBitCommand bit(ByteBuffer key) {
return new SetBitCommand(key, null, null);
/**
* Creates a new {@link SetBitCommand} given a {@literal key}.
*
* @param key must not be {@literal null}.
* @return a new {@link SetBitCommand} for a {@literal key}.
*/
public static SetBitCommand bit(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new SetBitCommand(key, null, false);
}
public ReactiveStringCommands.SetBitCommand atOffset(Long index) {
return new ReactiveStringCommands.SetBitCommand(getKey(), index, value);
/**
* Applies the offset {@literal index}. Constructs a new command instance with all previously configured properties.
*
* @param index
* @return a new {@link SetBitCommand} with {@literal offset} applied.
*/
public SetBitCommand atOffset(long index) {
return new SetBitCommand(getKey(), index, value);
}
public ReactiveStringCommands.SetBitCommand to(Boolean bit) {
return new ReactiveStringCommands.SetBitCommand(getKey(), offset, bit);
/**
* Applies the {@literal bit}. Constructs a new command instance with all previously configured properties.
*
* @param bit
* @return a new {@link SetBitCommand} with {@literal offset} applied.
*/
public SetBitCommand to(boolean bit) {
return new SetBitCommand(getKey(), offset, bit);
}
/**
* @return
*/
public Long getOffset() {
return offset;
}
public Boolean getValue() {
/**
* @return
*/
public boolean getValue() {
return value;
}
}
/**
* Sets the bit at {@code offset} in value stored at {@code key} and return the original value.
* Sets the bit at {@literal offset} in value stored at {@literal key} and return the original value.
*
* @param key must not be {@literal null}.
* @param offset
@@ -581,59 +755,79 @@ public interface ReactiveStringCommands {
}
/**
* Sets the bit at {@code offset} in value stored at {@code key} and return the original value.
* Sets the bit at {@literal offset} in value stored at {@literal key} and return the original value.
*
* @param keys must not be {@literal null}.
* @param offset must not be {@literal null}.
* @param value must not be {@literal null}.
* @param commands must not be {@literal null}.
* @return
*/
Flux<BooleanResponse<ReactiveStringCommands.SetBitCommand>> setBit(
Publisher<ReactiveStringCommands.SetBitCommand> commands);
Flux<BooleanResponse<SetBitCommand>> setBit(Publisher<SetBitCommand> commands);
/**
* {@code BITCOUNT} command parameters.
*
* @author Christoph Strobl
*/
class BitCountCommand extends KeyCommand {
private Range<Long> range;
public BitCountCommand(ByteBuffer key, Range<Long> range) {
private BitCountCommand(ByteBuffer key, Range<Long> range) {
super(key);
this.range = range;
}
public static ReactiveStringCommands.BitCountCommand bitCount(ByteBuffer key) {
return new ReactiveStringCommands.BitCountCommand(key, null);
/**
* Creates a new {@link BitCountCommand} given a {@literal key}.
*
* @param key must not be {@literal null}.
* @return a new {@link BitCountCommand} for a {@literal key}.
*/
public static BitCountCommand bitCount(ByteBuffer key) {
Assert.notNull(key, "Key must not be null!");
return new BitCountCommand(key, null);
}
public ReactiveStringCommands.BitCountCommand within(Range<Long> range) {
return new ReactiveStringCommands.BitCountCommand(getKey(), range);
/**
* Applies the {@link Range}. Constructs a new command instance with all previously configured properties.
*
* @param range must not be {@literal null}.
* @return a new {@link BitCountCommand} with {@link Range} applied.
*/
public BitCountCommand within(Range<Long> range) {
Assert.notNull(range, "Range must not be null!");
return new BitCountCommand(getKey(), range);
}
/**
* @return
*/
public Range<Long> getRange() {
return range;
}
}
/**
* Count the number of set bits (population counting) in value stored at {@code key}.
* Count the number of set bits (population counting) in value stored at {@literal key}.
*
* @param key must not be {@literal null}.
* @return
*/
default Mono<Long> bitCount(ByteBuffer key) {
Assert.notNull(key, "Key must not be null");
Assert.notNull(key, "Key must not be null!");
return bitCount(Mono.just(BitCountCommand.bitCount(key))).next().map(NumericResponse::getOutput);
}
/**
* Count the number of set bits (population counting) of value stored at {@code key} between {@code begin} and
* {@code end}.
* Count the number of set bits (population counting) of value stored at {@literal key} between {@literal begin} and
* {@literal end}.
*
* @param key must not be {@literal null}.
* @param begin
@@ -642,25 +836,24 @@ public interface ReactiveStringCommands {
*/
default Mono<Long> bitCount(ByteBuffer key, long begin, long end) {
Assert.notNull(key, "Key must not be null");
Assert.notNull(key, "Key must not be null!");
return bitCount(Mono.just(BitCountCommand.bitCount(key).within(new Range<>(begin, end)))).next()
.map(NumericResponse::getOutput);
}
/**
* Count the number of set bits (population counting) of value stored at {@code key} between {@code begin} and
* {@code end}.
* Count the number of set bits (population counting) of value stored at {@literal key} between {@literal begin} and
* {@literal end}.
*
* @param keys must not be {@literal null}.
* @param begin must not be {@literal null}.
* @param end must not be {@literal null}.
* @param commands must not be {@literal null}.
* @return
*/
Flux<NumericResponse<ReactiveStringCommands.BitCountCommand, Long>> bitCount(
Publisher<ReactiveStringCommands.BitCountCommand> commands);
Flux<NumericResponse<BitCountCommand, Long>> bitCount(Publisher<BitCountCommand> commands);
/**
* {@code BITOP} command parameters.
*
* @author Christoph Strobl
*/
class BitOpCommand {
@@ -676,30 +869,67 @@ public interface ReactiveStringCommands {
this.destinationKey = destinationKey;
}
public static ReactiveStringCommands.BitOpCommand perform(BitOperation bitOp) {
return new ReactiveStringCommands.BitOpCommand(null, bitOp, null);
/**
* Creates a new {@link BitOpCommand} given a {@link BitOperation}.
*
* @param bitOp must not be {@literal null}.
* @return a new {@link BitCountCommand} for a {@link BitOperation}.
*/
public static BitOpCommand perform(BitOperation bitOp) {
Assert.notNull(bitOp, "BitOperation must not be null!");
return new BitOpCommand(null, bitOp, null);
}
/**
* Applies the operation to {@literal keys}. Constructs a new command instance with all previously configured
* properties.
*
* @param keys must not be {@literal null}.
* @return a new {@link BitOpCommand} with {@link Range} applied.
*/
public BitOpCommand onKeys(Collection<ByteBuffer> keys) {
Assert.notNull(keys, "Keys must not be null!");
return new BitOpCommand(new ArrayList<>(keys), bitOp, destinationKey);
}
/**
* Applies the {@literal key} to store the result at. Constructs a new command instance with all previously
* configured properties.
*
* @param destinationKey must not be {@literal null}.
* @return a new {@link BitOpCommand} with {@link Range} applied.
*/
public BitOpCommand andSaveAs(ByteBuffer destinationKey) {
Assert.notNull(destinationKey, "Destination key must not be null!");
return new BitOpCommand(keys, bitOp, destinationKey);
}
/**
* @return
*/
public BitOperation getBitOp() {
return bitOp;
}
public ReactiveStringCommands.BitOpCommand onKeys(List<ByteBuffer> keys) {
return new ReactiveStringCommands.BitOpCommand(keys, bitOp, destinationKey);
}
/**
* @return
*/
public List<ByteBuffer> getKeys() {
return keys;
}
public ReactiveStringCommands.BitOpCommand andSaveAs(ByteBuffer destinationKey) {
return new ReactiveStringCommands.BitOpCommand(keys, bitOp, destinationKey);
}
/**
* @return
*/
public ByteBuffer getDestinationKey() {
return destinationKey;
}
}
/**
@@ -710,40 +940,40 @@ public interface ReactiveStringCommands {
* @param destination must not be {@literal null}.
* @return
*/
default Mono<Long> bitOp(List<ByteBuffer> keys, BitOperation bitOp, ByteBuffer destination) {
default Mono<Long> bitOp(Collection<ByteBuffer> keys, BitOperation bitOp, ByteBuffer destination) {
Assert.notNull(keys, "keys must not be null");
Assert.notNull(keys, "Keys must not be null!");
Assert.notNull(bitOp, "BitOperation must not be null!");
Assert.notNull(destination, "Destination must not be null!");
return bitOp(Mono.just(BitOpCommand.perform(bitOp).onKeys(keys).andSaveAs(destination))).next()
return bitOp(Mono.just(BitOpCommand.perform(bitOp).onKeys(keys).andSaveAs(destination))) //
.next() //
.map(NumericResponse::getOutput);
}
/**
* Perform bitwise operations between strings.
*
* @param keys must not be {@literal null}.
* @param bitOp must not be {@literal null}.
* @param destination must not be {@literal null}.
* @param commands must not be {@literal null}.
* @return
*/
Flux<NumericResponse<ReactiveStringCommands.BitOpCommand, Long>> bitOp(
Publisher<ReactiveStringCommands.BitOpCommand> commands);
Flux<NumericResponse<BitOpCommand, Long>> bitOp(Publisher<BitOpCommand> commands);
/**
* Get the length of the value stored at {@code key}.
* Get the length of the value stored at {@literal key}.
*
* @param key must not be {@literal null}.
* @return
*/
default Mono<Long> strLen(ByteBuffer key) {
Assert.notNull(key, "key must not be null");
Assert.notNull(key, "Key must not be null!");
return strLen(Mono.just(new KeyCommand(key))).next().map(NumericResponse::getOutput);
}
/**
* Get the length of the value stored at {@code key}.
* Get the length of the value stored at {@literal key}.
*
* @param keys must not be {@literal null}.
* @return

View File

@@ -58,7 +58,7 @@ public class RedisClusterNode extends RedisNode {
public RedisClusterNode(String id) {
this(new SlotRange(Collections.<Integer> emptySet()));
Assert.notNull(id, "Id must not be null");
Assert.notNull(id, "Id must not be null!");
this.id = id;
}

View File

@@ -1219,14 +1219,14 @@ public class LettuceConnection extends AbstractRedisConnection {
}
try {
if (isPipelined()) {
pipeline(new LettuceStatusResult(((RedisAsyncCommands) getAsyncDedicatedConnection()).watch(keys)));
pipeline(new LettuceStatusResult(((RedisAsyncCommands) getAsyncDedicatedConnection()).watch((Object[]) keys)));
return;
}
if (isQueueing()) {
transaction(new LettuceTxStatusResult(((RedisAsyncCommands) getDedicatedConnection()).watch()));
transaction(new LettuceTxStatusResult(((RedisAsyncCommands) getDedicatedConnection()).watch((Object[]) keys)));
return;
}
((RedisCommands) getDedicatedConnection()).watch(keys);
((RedisCommands) getDedicatedConnection()).watch((Object[]) keys);
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}

View File

@@ -54,6 +54,7 @@ import org.springframework.data.redis.connection.convert.LongToBooleanConverter;
import org.springframework.data.redis.connection.convert.StringToRedisClientInfoConverter;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.data.redis.core.types.RedisClientInfo;
import org.springframework.data.redis.util.ByteUtils;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
@@ -739,7 +740,7 @@ abstract public class LettuceConverters extends Converters {
ByteBuffer buffer = ByteBuffer.allocate(prefix.length + value.length);
buffer.put(prefix);
buffer.put(value);
return toString(buffer.array());
return toString(ByteUtils.getBytes(buffer));
}
public static List<RedisClusterNode> partitionsToClusterNodes(Partitions partitions) {

View File

@@ -13,7 +13,6 @@
* 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.ReactiveClusterGeoCommands;

View File

@@ -13,7 +13,6 @@
* 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.ReactiveClusterHashCommands;

View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection.lettuce;
import java.nio.ByteBuffer;
@@ -24,7 +23,8 @@ import org.reactivestreams.Publisher;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.redis.connection.ClusterSlotHashUtil;
import org.springframework.data.redis.connection.ReactiveClusterHyperLogLogCommands;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveRedisConnection.BooleanResponse;
import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
@@ -46,8 +46,11 @@ public class LettuceReactiveClusterHyperLogLogCommands extends LettuceReactiveHy
super(connection);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveHyperLogLogCommands#pfMerge(org.reactivestreams.Publisher)
*/
@Override
public Flux<ReactiveRedisConnection.BooleanResponse<PfMergeCommand>> pfMerge(Publisher<PfMergeCommand> commands) {
public Flux<BooleanResponse<PfMergeCommand>> pfMerge(Publisher<PfMergeCommand> commands) {
return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {
@@ -66,9 +69,11 @@ public class LettuceReactiveClusterHyperLogLogCommands extends LettuceReactiveHy
}));
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveHyperLogLogCommands#pfCount(org.reactivestreams.Publisher)
*/
@Override
public Flux<ReactiveRedisConnection.NumericResponse<PfCountCommand, Long>> pfCount(
Publisher<PfCountCommand> commands) {
public Flux<NumericResponse<PfCountCommand, Long>> pfCount(Publisher<PfCountCommand> commands) {
return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {

View File

@@ -23,7 +23,7 @@ import org.reactivestreams.Publisher;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.ClusterSlotHashUtil;
import org.springframework.data.redis.connection.ReactiveClusterKeyCommands;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveRedisConnection.BooleanResponse;
import org.springframework.data.redis.connection.RedisClusterNode;
import org.springframework.util.Assert;
@@ -51,6 +51,7 @@ public class LettuceReactiveClusterKeyCommands extends LettuceReactiveKeyCommand
public LettuceReactiveClusterKeyCommands(LettuceReactiveRedisClusterConnection connection) {
super(connection);
this.connection = connection;
}
@@ -76,12 +77,12 @@ public class LettuceReactiveClusterKeyCommands extends LettuceReactiveKeyCommand
* @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveKeyCommands#rename(org.reactivestreams.Publisher, java.util.function.Supplier)
*/
@Override
public Flux<ReactiveRedisConnection.BooleanResponse<RenameCommand>> rename(Publisher<RenameCommand> commands) {
public Flux<BooleanResponse<RenameCommand>> rename(Publisher<RenameCommand> commands) {
return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
Assert.notNull(command.getKey(), "key must not be null.");
Assert.notNull(command.getNewName(), "NewName must not be null");
Assert.notNull(command.getNewName(), "NewName must not be null!");
if (ClusterSlotHashUtil.isSameSlotForAllKeys(command.getKey(), command.getNewName())) {
return super.rename(Mono.just(command));
@@ -93,7 +94,7 @@ public class LettuceReactiveClusterKeyCommands extends LettuceReactiveKeyCommand
.flatMap(value -> cmd.restore(command.getNewName(), 0, value).flatMap(res -> cmd.del(command.getKey())))
.map(LettuceConverters.longToBooleanConverter()::convert);
return result.map(val -> new ReactiveRedisConnection.BooleanResponse<>(command, val));
return result.map(val -> new BooleanResponse<>(command, val));
}));
}
@@ -102,12 +103,12 @@ public class LettuceReactiveClusterKeyCommands extends LettuceReactiveKeyCommand
* @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveKeyCommands#renameNX(org.reactivestreams.Publisher, java.util.function.Supplier)
*/
@Override
public Flux<ReactiveRedisConnection.BooleanResponse<RenameCommand>> renameNX(Publisher<RenameCommand> commands) {
public Flux<BooleanResponse<RenameCommand>> renameNX(Publisher<RenameCommand> commands) {
return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
Assert.notNull(command.getKey(), "Key must not be null.");
Assert.notNull(command.getNewName(), "NewName must not be null");
Assert.notNull(command.getNewName(), "NewName must not be null!");
if (ClusterSlotHashUtil.isSameSlotForAllKeys(command.getKey(), command.getNewName())) {
return super.renameNX(Mono.just(command));
@@ -126,7 +127,7 @@ public class LettuceReactiveClusterKeyCommands extends LettuceReactiveKeyCommand
.map(LettuceConverters.longToBooleanConverter()::convert);
});
return result.map(val -> new ReactiveRedisConnection.BooleanResponse<>(command, val));
return result.map(val -> new BooleanResponse<>(command, val));
}));
}
}

View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection.lettuce;
import java.nio.ByteBuffer;
@@ -22,7 +21,7 @@ import org.reactivestreams.Publisher;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.redis.connection.ClusterSlotHashUtil;
import org.springframework.data.redis.connection.ReactiveClusterListCommands;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveRedisConnection.ByteBufferResponse;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
@@ -45,6 +44,9 @@ public class LettuceReactiveClusterListCommands extends LettuceReactiveListComma
super(connection);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveListCommands#bPop(org.reactivestreams.Publisher)
*/
@Override
public Flux<PopResponse> bPop(Publisher<BPopCommand> commands) {
@@ -61,9 +63,11 @@ public class LettuceReactiveClusterListCommands extends LettuceReactiveListComma
}));
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveListCommands#rPopLPush(org.reactivestreams.Publisher)
*/
@Override
public Flux<ReactiveRedisConnection.ByteBufferResponse<RPopLPushCommand>> rPopLPush(
Publisher<RPopLPushCommand> commands) {
public Flux<ByteBufferResponse<RPopLPushCommand>> rPopLPush(Publisher<RPopLPushCommand> commands) {
return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {
@@ -77,7 +81,7 @@ public class LettuceReactiveClusterListCommands extends LettuceReactiveListComma
Flux<ByteBuffer> result = cmd.rpop(command.getKey())
.flatMap(value -> cmd.lpush(command.getDestination(), value).map(x -> value));
return result.map(value -> new ReactiveRedisConnection.ByteBufferResponse<>(command, value));
return result.map(value -> new ByteBufferResponse<>(command, value));
}));
}
}

View File

@@ -13,7 +13,6 @@
* 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.ReactiveClusterNumberCommands;

View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection.lettuce;
import java.nio.ByteBuffer;
@@ -24,7 +23,9 @@ import java.util.stream.Collectors;
import org.reactivestreams.Publisher;
import org.springframework.data.redis.connection.ClusterSlotHashUtil;
import org.springframework.data.redis.connection.ReactiveClusterSetCommands;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveRedisConnection.BooleanResponse;
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;
@@ -47,9 +48,11 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
super(connection);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveSetCommands#sUnion(org.reactivestreams.Publisher)
*/
@Override
public Flux<ReactiveRedisConnection.MultiValueResponse<SUnionCommand, ByteBuffer>> sUnion(
Publisher<SUnionCommand> commands) {
public Flux<MultiValueResponse<SUnionCommand, ByteBuffer>> sUnion(Publisher<SUnionCommand> commands) {
return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {
@@ -62,13 +65,15 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
Mono<List<ByteBuffer>> result = Flux
.merge(command.getKeys().stream().map(cmd::smembers).collect(Collectors.toList())).distinct().collectList();
return result.map(value -> new ReactiveRedisConnection.MultiValueResponse<>(command, value));
return result.map(value -> new MultiValueResponse<>(command, value));
}));
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveSetCommands#sUnionStore(org.reactivestreams.Publisher)
*/
@Override
public Flux<ReactiveRedisConnection.NumericResponse<SUnionStoreCommand, Long>> sUnionStore(
Publisher<SUnionStoreCommand> commands) {
public Flux<NumericResponse<SUnionStoreCommand, Long>> sUnionStore(Publisher<SUnionStoreCommand> commands) {
return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {
@@ -84,14 +89,16 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
return sUnion(Mono.just(SUnionCommand.keys(command.getKeys()))).next().flatMap(values -> {
Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().stream().toArray(ByteBuffer[]::new));
return result.map(value -> new ReactiveRedisConnection.NumericResponse<>(command, value));
return result.map(value -> new NumericResponse<>(command, value));
});
}));
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveSetCommands#sInter(org.reactivestreams.Publisher)
*/
@Override
public Flux<ReactiveRedisConnection.MultiValueResponse<SInterCommand, ByteBuffer>> sInter(
Publisher<SInterCommand> commands) {
public Flux<MultiValueResponse<SInterCommand, ByteBuffer>> sInter(Publisher<SInterCommand> commands) {
return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {
@@ -115,13 +122,15 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
return source;
});
return result.map(value -> new ReactiveRedisConnection.MultiValueResponse<>(command, value));
return result.map(value -> new MultiValueResponse<>(command, value));
}));
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveSetCommands#sInterStore(org.reactivestreams.Publisher)
*/
@Override
public Flux<ReactiveRedisConnection.NumericResponse<SInterStoreCommand, Long>> sInterStore(
Publisher<SInterStoreCommand> commands) {
public Flux<NumericResponse<SInterStoreCommand, Long>> sInterStore(Publisher<SInterStoreCommand> commands) {
return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {
@@ -137,14 +146,16 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
return sInter(Mono.just(SInterCommand.keys(command.getKeys()))).next().flatMap(values -> {
Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().stream().toArray(ByteBuffer[]::new));
return result.map(value -> new ReactiveRedisConnection.NumericResponse<>(command, value));
return result.map(value -> new NumericResponse<>(command, value));
});
}));
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveSetCommands#sDiff(org.reactivestreams.Publisher)
*/
@Override
public Flux<ReactiveRedisConnection.MultiValueResponse<SDiffCommand, ByteBuffer>> sDiff(
Publisher<SDiffCommand> commands) {
public Flux<MultiValueResponse<SDiffCommand, ByteBuffer>> sDiff(Publisher<SDiffCommand> commands) {
return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {
@@ -168,14 +179,16 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
return source;
});
return result.map(value -> new ReactiveRedisConnection.MultiValueResponse<>(command, value));
return result.map(value -> new MultiValueResponse<>(command, value));
}));
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveSetCommands#sDiffStore(org.reactivestreams.Publisher)
*/
@Override
public Flux<ReactiveRedisConnection.NumericResponse<SDiffStoreCommand, Long>> sDiffStore(
Publisher<SDiffStoreCommand> commands) {
public Flux<NumericResponse<SDiffStoreCommand, Long>> sDiffStore(Publisher<SDiffStoreCommand> commands) {
return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {
@@ -191,13 +204,16 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
return sDiff(Mono.just(SDiffCommand.keys(command.getKeys()))).next().flatMap(values -> {
Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().stream().toArray(ByteBuffer[]::new));
return result.map(value -> new ReactiveRedisConnection.NumericResponse<>(command, value));
return result.map(value -> new NumericResponse<>(command, value));
});
}));
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveSetCommands#sMove(org.reactivestreams.Publisher)
*/
@Override
public Flux<ReactiveRedisConnection.BooleanResponse<SMoveCommand>> sMove(Publisher<SMoveCommand> commands) {
public Flux<BooleanResponse<SMoveCommand>> sMove(Publisher<SMoveCommand> commands) {
return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {
@@ -227,8 +243,7 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
});
return result.defaultIfEmpty(Boolean.FALSE)
.map(value -> new ReactiveRedisConnection.BooleanResponse<>(command, value));
return result.defaultIfEmpty(Boolean.FALSE).map(value -> new BooleanResponse<>(command, value));
}));
}
}

View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection.lettuce;
import java.nio.ByteBuffer;
@@ -45,6 +44,9 @@ public class LettuceReactiveClusterStringCommands extends LettuceReactiveStringC
super(connection);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveStringCommands#bitOp(org.reactivestreams.Publisher)
*/
@Override
public Flux<ReactiveRedisConnection.NumericResponse<BitOpCommand, Long>> bitOp(Publisher<BitOpCommand> commands) {
@@ -62,6 +64,9 @@ public class LettuceReactiveClusterStringCommands extends LettuceReactiveStringC
}));
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveStringCommands#mSetNX(org.reactivestreams.Publisher)
*/
@Override
public Flux<ReactiveRedisConnection.BooleanResponse<MSetCommand>> mSetNX(Publisher<MSetCommand> commands) {

View File

@@ -13,14 +13,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection.lettuce;
import org.reactivestreams.Publisher;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.redis.connection.ClusterSlotHashUtil;
import org.springframework.data.redis.connection.ReactiveClusterZSetCommands;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
@@ -42,9 +41,11 @@ public class LettuceReactiveClusterZSetCommands extends LettuceReactiveZSetComma
super(connection);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveZSetCommands#zUnionStore(org.reactivestreams.Publisher)
*/
@Override
public Flux<ReactiveRedisConnection.NumericResponse<ZUnionStoreCommand, Long>> zUnionStore(
Publisher<ZUnionStoreCommand> commands) {
public Flux<NumericResponse<ZUnionStoreCommand, Long>> zUnionStore(Publisher<ZUnionStoreCommand> commands) {
return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {
@@ -59,9 +60,11 @@ public class LettuceReactiveClusterZSetCommands extends LettuceReactiveZSetComma
}));
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveZSetCommands#zInterStore(org.reactivestreams.Publisher)
*/
@Override
public Flux<ReactiveRedisConnection.NumericResponse<ZInterStoreCommand, Long>> zInterStore(
Publisher<ZInterStoreCommand> commands) {
public Flux<NumericResponse<ZInterStoreCommand, Long>> zInterStore(Publisher<ZInterStoreCommand> commands) {
return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {
Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty.");

View File

@@ -19,7 +19,6 @@ import java.nio.ByteBuffer;
import java.util.List;
import java.util.stream.Collectors;
import com.lambdaworks.redis.api.reactive.RedisKeyReactiveCommands;
import org.reactivestreams.Publisher;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.connection.ReactiveKeyCommands;
@@ -33,6 +32,8 @@ import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import com.lambdaworks.redis.api.reactive.RedisKeyReactiveCommands;
/**
* @author Christoph Strobl
* @since 2.0
@@ -49,6 +50,7 @@ public class LettuceReactiveKeyCommands implements ReactiveKeyCommands {
public LettuceReactiveKeyCommands(LettuceReactiveRedisConnection connection) {
Assert.notNull(connection, "Connection must not be null!");
this.connection = connection;
}

View File

@@ -52,6 +52,7 @@ public class LettuceReactiveListCommands implements ReactiveListCommands {
public LettuceReactiveListCommands(LettuceReactiveRedisConnection connection) {
Assert.notNull(connection, "Connection must not be null!");
this.connection = connection;
}

View File

@@ -36,12 +36,13 @@ public class LettuceReactiveNumberCommands implements ReactiveNumberCommands {
/**
* Create new {@link LettuceReactiveStringCommands}.
*
*
* @param connection must not be {@literal null}.
*/
public LettuceReactiveNumberCommands(LettuceReactiveRedisConnection connection) {
Assert.notNull(connection, "Connection must not be null!");
this.connection = connection;
}

View File

@@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection.lettuce;
import java.nio.ByteBuffer;
@@ -42,46 +41,73 @@ public class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisC
super(client);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#keyCommands()
*/
@Override
public LettuceReactiveClusterKeyCommands keyCommands() {
return new LettuceReactiveClusterKeyCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#listCommands()
*/
@Override
public LettuceReactiveClusterListCommands listCommands() {
return new LettuceReactiveClusterListCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#setCommands()
*/
@Override
public LettuceReactiveClusterSetCommands setCommands() {
return new LettuceReactiveClusterSetCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#zSetCommands()
*/
@Override
public LettuceReactiveClusterZSetCommands zSetCommands() {
return new LettuceReactiveClusterZSetCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#hyperLogLogCommands()
*/
@Override
public LettuceReactiveClusterHyperLogLogCommands hyperLogLogCommands() {
return new LettuceReactiveClusterHyperLogLogCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#stringCommands()
*/
@Override
public LettuceReactiveClusterStringCommands stringCommands() {
return new LettuceReactiveClusterStringCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#geoCommands()
*/
@Override
public LettuceReactiveClusterGeoCommands geoCommands() {
return new LettuceReactiveClusterGeoCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#hashCommands()
*/
@Override
public LettuceReactiveClusterHashCommands hashCommands() {
return new LettuceReactiveClusterHashCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#numberCommands()
*/
@Override
public LettuceReactiveClusterNumberCommands numberCommands() {
return new LettuceReactiveClusterNumberCommands(this);
@@ -103,6 +129,10 @@ public class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisC
return Flux.defer(() -> callback.doWithCommands(getCommands(node))).onErrorResumeWith(translateExeception());
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#getConnection()
*/
@SuppressWarnings("unchecked")
@Override
protected StatefulRedisClusterConnection<ByteBuffer, ByteBuffer> getConnection() {
@@ -112,10 +142,14 @@ public class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisC
return (StatefulRedisClusterConnection) super.getConnection();
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#getCommands()
*/
protected RedisClusterReactiveCommands<ByteBuffer, ByteBuffer> getCommands() {
return getConnection().reactive();
}
@SuppressWarnings("unchecked")
protected RedisReactiveCommands<ByteBuffer, ByteBuffer> getCommands(RedisNode node) {
if (!(getConnection() instanceof StatefulRedisClusterConnection)) {

View File

@@ -19,7 +19,6 @@ import java.nio.ByteBuffer;
import java.util.function.Function;
import org.reactivestreams.Publisher;
import org.springframework.core.convert.converter.Converter;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.InvalidDataAccessResourceUsageException;
import org.springframework.data.redis.connection.*;
@@ -35,8 +34,6 @@ import com.lambdaworks.redis.cluster.api.reactive.RedisClusterReactiveCommands;
import com.lambdaworks.redis.codec.RedisCodec;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import rx.Observable;
/**
* @author Christoph Strobl
@@ -71,41 +68,65 @@ public class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
return new LettuceReactiveKeyCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveRedisConnection#stringCommands()
*/
@Override
public ReactiveStringCommands stringCommands() {
return new LettuceReactiveStringCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveRedisConnection#numberCommands()
*/
@Override
public ReactiveNumberCommands numberCommands() {
return new LettuceReactiveNumberCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveRedisConnection#listCommands()
*/
@Override
public ReactiveListCommands listCommands() {
return new LettuceReactiveListCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveRedisConnection#setCommands()
*/
@Override
public ReactiveSetCommands setCommands() {
return new LettuceReactiveSetCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveRedisConnection#zSetCommands()
*/
@Override
public ReactiveZSetCommands zSetCommands() {
return new LettuceReactiveZSetCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveRedisConnection#hashCommands()
*/
@Override
public ReactiveHashCommands hashCommands() {
return new LettuceReactiveHashCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveRedisConnection#geoCommands()
*/
@Override
public ReactiveGeoCommands geoCommands() {
return new LettuceReactiveGeoCommands(this);
}
/* (non-Javadoc)
* @see org.springframework.data.redis.connection.ReactiveRedisConnection#hyperLogLogCommands()
*/
@Override
public ReactiveHyperLogLogCommands hyperLogLogCommands() {
return new LettuceReactiveHyperLogLogCommands(this);
@@ -119,6 +140,9 @@ public class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
return Flux.defer(() -> callback.doWithCommands(getCommands())).onErrorResumeWith(translateExeception());
}
/* (non-Javadoc)
* @see java.io.Closeable#close()
*/
@Override
public void close() {
connection.close();

View File

@@ -48,6 +48,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
public LettuceReactiveSetCommands(LettuceReactiveRedisConnection connection) {
Assert.notNull(connection, "Connection must not be null!");
this.connection = connection;
}

View File

@@ -52,6 +52,7 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
public LettuceReactiveStringCommands(LettuceReactiveRedisConnection connection) {
Assert.notNull(connection, "Connection must not be null!");
this.connection = connection;
}

View File

@@ -28,6 +28,7 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection.Numeric
import org.springframework.data.redis.connection.ReactiveZSetCommands;
import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate;
import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
import org.springframework.data.redis.util.ByteUtils;
import org.springframework.data.util.DirectFieldAccessFallbackBeanWrapper;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
@@ -61,6 +62,7 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
public LettuceReactiveZSetCommands(LettuceReactiveRedisConnection connection) {
Assert.notNull(connection, "Connection must not be null!");
this.connection = connection;
}
@@ -75,17 +77,16 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notEmpty(command.getTuples(), "Tuples must not be empty or null.");
Assert.notEmpty(command.getTuples(), "Tuples must not be empty or null!");
ZAddArgs args = null;
if (command.getIncr().isPresent() || command.getUpsert().isPresent()
|| command.getReturnTotalChanged().isPresent()) {
if (command.isIncr() || command.isUpsert() || command.isReturnTotalChanged()) {
if (command.getIncr().isPresent() && ObjectUtils.nullSafeEquals(command.getIncr().get(), Boolean.TRUE)) {
if (command.isIncr()) {
if (command.getTuples().size() > 1) {
throw new IllegalArgumentException("ZADD INCR must not contain more than one tuple.");
throw new IllegalArgumentException("ZADD INCR must not contain more than one tuple!");
}
Tuple tuple = command.getTuples().iterator().next();
@@ -94,18 +95,14 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
.map(value -> new NumericResponse<>(command, value));
}
if (command.getReturnTotalChanged().isPresent()
&& ObjectUtils.nullSafeEquals(command.getReturnTotalChanged().get(), Boolean.TRUE)) {
if (command.isReturnTotalChanged()) {
args = ZAddArgs.Builder.ch();
}
if (command.getUpsert().isPresent()) {
if (command.getUpsert().get().equals(Boolean.TRUE)) {
args = ZAddArgs.Builder.nx();
} else {
args = ZAddArgs.Builder.xx();
}
if (command.isUpsert()) {
args = ZAddArgs.Builder.nx();
} else {
args = ZAddArgs.Builder.xx();
}
}
@@ -188,8 +185,7 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Mono<List<Tuple>> result;
if (ObjectUtils.nullSafeEquals(command.getDirection(), Direction.ASC)) {
if (command.getWithScores().isPresent()
&& ObjectUtils.nullSafeEquals(command.getWithScores().get(), Boolean.TRUE)) {
if (command.isWithScores()) {
result = cmd
.zrangeWithScores(command.getKey(), command.getRange().getLowerBound(),
@@ -198,11 +194,10 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
} else {
result = cmd.zrange(command.getKey(), command.getRange().getLowerBound(), command.getRange().getUpperBound())
.map(value -> (Tuple) new DefaultTuple(getBytes(value), Double.NaN)).collectList();
.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN)).collectList();
}
} else {
if (command.getWithScores().isPresent()
&& ObjectUtils.nullSafeEquals(command.getWithScores().get(), Boolean.TRUE)) {
if (command.isWithScores()) {
result = cmd
.zrevrangeWithScores(command.getKey(), command.getRange().getLowerBound(),
@@ -212,7 +207,7 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
result = cmd
.zrevrange(command.getKey(), command.getRange().getLowerBound(), command.getRange().getUpperBound())
.map(value -> (Tuple) new DefaultTuple(getBytes(value), Double.NaN)).collectList();
.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN)).collectList();
}
}
@@ -237,60 +232,55 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Mono<List<Tuple>> result;
if (ObjectUtils.nullSafeEquals(command.getDirection(), Direction.ASC)) {
Range<Number> range = ArgumentConverters.toRange(command.getRange());
if (command.getWithScores().isPresent()
&& ObjectUtils.nullSafeEquals(command.getWithScores().get(), Boolean.TRUE)) {
if (command.isWithScores()) {
if (!isLimited) {
result = cmd.zrangebyscoreWithScores(command.getKey(), range)
.map(sc -> (Tuple) new DefaultTuple(getBytes(sc.getValue()), sc.getScore())).collectList();
.map(sc -> (Tuple) new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore())).collectList();
} else {
result = cmd
.zrangebyscoreWithScores(command.getKey(), range,
LettuceConverters.toLimit(command.getLimit().get()))
.map(sc -> (Tuple) new DefaultTuple(getBytes(sc.getValue()), sc.getScore())).collectList();
.zrangebyscoreWithScores(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
.map(sc -> (Tuple) new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore())).collectList();
}
} else {
if (!isLimited) {
result = cmd.zrangebyscore(command.getKey(), range)
.map(value -> (Tuple) new DefaultTuple(getBytes(value), Double.NaN)).collectList();
.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN)).collectList();
} else {
result = cmd
.zrangebyscore(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
.map(value -> (Tuple) new DefaultTuple(getBytes(value), Double.NaN)).collectList();
result = cmd.zrangebyscore(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN)).collectList();
}
}
} else {
Range<Number> range = ArgumentConverters.toRevRange(command.getRange());
if (command.getWithScores().isPresent()
&& ObjectUtils.nullSafeEquals(command.getWithScores().get(), Boolean.TRUE)) {
if (command.isWithScores()) {
if (!isLimited) {
result = cmd.zrevrangebyscoreWithScores(command.getKey(), range)
.map(sc -> (Tuple) new DefaultTuple(getBytes(sc.getValue()), sc.getScore())).collectList();
.map(sc -> (Tuple) new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore())).collectList();
} else {
result = cmd
.zrevrangebyscoreWithScores(command.getKey(), range,
LettuceConverters.toLimit(command.getLimit().get()))
.map(sc -> (Tuple) new DefaultTuple(getBytes(sc.getValue()), sc.getScore())).collectList();
.map(sc -> (Tuple) new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore())).collectList();
}
} else {
if (!isLimited) {
result = cmd.zrevrangebyscore(command.getKey(), range)
.map(value -> (Tuple) new DefaultTuple(getBytes(value), Double.NaN)).collectList();
.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN)).collectList();
} else {
result = cmd
.zrevrangebyscore(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
.map(value -> (Tuple) new DefaultTuple(getBytes(value), Double.NaN)).collectList();
result = cmd.zrevrangebyscore(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN)).collectList();
}
}
}
@@ -451,13 +441,14 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Flux<ByteBuffer> result;
if (command.getLimit() != null) {
if (ObjectUtils.nullSafeEquals(command.getDirection(), Direction.ASC)) {
result = cmd.zrangebylex(command.getKey(), ArgumentConverters.toRange(command.getRange()), LettuceConverters.toLimit(command.getLimit()));
result = cmd.zrangebylex(command.getKey(), ArgumentConverters.toRange(command.getRange()),
LettuceConverters.toLimit(command.getLimit()));
} else {
result = cmd.zrevrangebylex(command.getKey(), ArgumentConverters.toRevRange(command.getRange()), LettuceConverters.toLimit(command.getLimit()));
result = cmd.zrevrangebylex(command.getKey(), ArgumentConverters.toRevRange(command.getRange()),
LettuceConverters.toLimit(command.getLimit()));
}
} else {
if (ObjectUtils.nullSafeEquals(command.getDirection(), Direction.ASC)) {
@@ -499,14 +490,7 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
}
private static byte[] getBytes(ScoredValue<ByteBuffer> scoredValue) {
return scoredValue.optional().map(LettuceReactiveZSetCommands::getBytes).orElse(new byte[0]);
}
private static byte[] getBytes(ByteBuffer byteBuffer) {
byte[] bytes = new byte[byteBuffer.remaining()];
byteBuffer.get(bytes);
return bytes;
return scoredValue.optional().map(ByteUtils::getBytes).orElse(new byte[0]);
}
protected LettuceReactiveRedisConnection getConnection() {
@@ -517,23 +501,23 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
* @author Christoph Strobl
* @author Mark Paluch
*/
static class ArgumentConverters {
private static class ArgumentConverters {
public static <T> Range<T> toRange(org.springframework.data.domain.Range<?> range) {
static <T> Range<T> toRange(org.springframework.data.domain.Range<?> range) {
return Range.from(lowerBoundArgOf(range), upperBoundArgOf(range));
}
public static <T> Range<T> toRevRange(org.springframework.data.domain.Range<?> range) {
static <T> Range<T> toRevRange(org.springframework.data.domain.Range<?> range) {
return Range.from(upperBoundArgOf(range), lowerBoundArgOf(range));
}
@SuppressWarnings("unchecked")
public static <T> Boundary<T> lowerBoundArgOf(org.springframework.data.domain.Range<?> range) {
static <T> Boundary<T> lowerBoundArgOf(org.springframework.data.domain.Range<?> range) {
return (Boundary<T>) rangeToBoundArgumentConverter(false).convert(range);
}
@SuppressWarnings("unchecked")
public static <T> Boundary<T> upperBoundArgOf(org.springframework.data.domain.Range<?> range) {
static <T> Boundary<T> upperBoundArgOf(org.springframework.data.domain.Range<?> range) {
return (Boundary<T>) rangeToBoundArgumentConverter(true).convert(range);
}

View File

@@ -15,6 +15,7 @@
*/
package org.springframework.data.redis.util;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -98,4 +99,21 @@ public final class ByteUtils {
return result;
}
/**
* Extract a byte array from {@link ByteBuffer} without consuming it.
*
* @param byteBuffer must not be {@literal null}.
* @return
* @since 2.0
*/
public static byte[] getBytes(ByteBuffer byteBuffer) {
Assert.notNull(byteBuffer, "ByteBuffer must not be null!");
ByteBuffer duplicate = byteBuffer.duplicate();
byte[] bytes = new byte[duplicate.remaining()];
duplicate.get(bytes);
return bytes;
}
}