Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Support for approximate trimming in XAddOptions and AddStreamRecord.

Browse Source 
Closes #2247
This commit is contained in:
dengliming
2022-01-26 22:53:46 +08:00
committed by Mark Paluch
parent e97ee20da5
commit 9b8bca689a
5 changed files with 58 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java
View File

@@ -201,13 +201,17 @@ public interface ReactiveStreamCommands {
private final ByteBufferRecord record;
private final @Nullable Long maxlen;
private final boolean nomkstream;
private final boolean approximateTrimming;
private AddStreamRecord(ByteBufferRecord record, @Nullable Long maxlen, boolean nomkstream) {
private AddStreamRecord(ByteBufferRecord record, @Nullable Long maxlen, boolean nomkstream,
boolean approximateTrimming) {
super(record.getStream());
this.record = record;
this.maxlen = maxlen;
this.nomkstream = nomkstream;
this.approximateTrimming = approximateTrimming;
}
/**
@@ -220,7 +224,7 @@ public interface ReactiveStreamCommands {
Assert.notNull(record, "Record must not be null!");
return new AddStreamRecord(record, null, false);
return new AddStreamRecord(record, null, false, false);
}
/**
@@ -233,7 +237,7 @@ public interface ReactiveStreamCommands {
Assert.notNull(body, "Body must not be null!");
return new AddStreamRecord(StreamRecords.rawBuffer(body), null, false);
return new AddStreamRecord(StreamRecords.rawBuffer(body), null, false, false);
}
/**
@@ -243,7 +247,7 @@ public interface ReactiveStreamCommands {
* @return a new {@link ReactiveGeoCommands.GeoAddCommand} with {@literal key} applied.
*/
public AddStreamRecord to(ByteBuffer key) {
return new AddStreamRecord(record.withStreamKey(key), maxlen, false);
return new AddStreamRecord(record.withStreamKey(key), maxlen, false, false);
}
/**
@@ -252,7 +256,7 @@ public interface ReactiveStreamCommands {
* @return new instance of {@link AddStreamRecord}.
*/
public AddStreamRecord maxlen(long maxlen) {
return new AddStreamRecord(record, maxlen, false);
return new AddStreamRecord(record, maxlen, false, false);
}
/**
@@ -262,7 +266,7 @@ public interface ReactiveStreamCommands {
* @since 2.6
*/
public AddStreamRecord makeNoStream() {
return new AddStreamRecord(record, maxlen, true);
return new AddStreamRecord(record, maxlen, true, false);
}
/**
@@ -273,7 +277,23 @@ public interface ReactiveStreamCommands {
* @since 2.6
*/
public AddStreamRecord makeNoStream(boolean makeNoStream) {
return new AddStreamRecord(record, maxlen, makeNoStream);
return new AddStreamRecord(record, maxlen, makeNoStream, false);
}
/**
* Apply efficient trimming for capped streams using the {@code ~} flag.
*
* @return new instance of {@link AddStreamRecord}.
*/
public AddStreamRecord approximateTrimming(boolean approximateTrimming) {
return new AddStreamRecord(record, maxlen, nomkstream, approximateTrimming);
}
/**
* @return {@literal true} if {@literal approximateTrimming} is set.
*/
public boolean isApproximateTrimming() {
return approximateTrimming;
}
/**

31
src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java
View File

@@ -116,18 +116,21 @@ public interface RedisStreamCommands {
*
* @author Christoph Strobl
* @author Mark John Moreno
* @author Liming Deng
* @since 2.3
*/
class XAddOptions {
private static final XAddOptions NONE = new XAddOptions(null, false);
private static final XAddOptions NONE = new XAddOptions(null, false, false);
private final @Nullable Long maxlen;
private final boolean nomkstream;
private final boolean approximateTrimming;
private XAddOptions(@Nullable Long maxlen, boolean nomkstream) {
private XAddOptions(@Nullable Long maxlen, boolean nomkstream, boolean approximateTrimming) {
this.maxlen = maxlen;
this.nomkstream = nomkstream;
this.approximateTrimming = approximateTrimming;
}
/**
@@ -143,7 +146,7 @@ public interface RedisStreamCommands {
* @return new instance of {@link XAddOptions}.
*/
public static XAddOptions maxlen(long maxlen) {
return new XAddOptions(maxlen, false);
return new XAddOptions(maxlen, false, false);
}
/**
@@ -153,7 +156,7 @@ public interface RedisStreamCommands {
* @since 2.6
*/
public static XAddOptions makeNoStream() {
return new XAddOptions(null, true);
return new XAddOptions(null, true, false);
}
/**
@@ -164,7 +167,23 @@ public interface RedisStreamCommands {
* @since 2.6
*/
public static XAddOptions makeNoStream(boolean makeNoStream) {
return new XAddOptions(null, makeNoStream);
return new XAddOptions(null, makeNoStream, false);
}
/**
* Apply efficient trimming for capped streams using the {@code ~} flag.
*
* @return new instance of {@link XAddOptions}.
*/
public XAddOptions approximateTrimming(boolean approximateTrimming) {
return new XAddOptions(null, nomkstream, approximateTrimming);
}
/**
* @return {@literal true} if {@literal approximateTrimming} is set.
*/
public boolean isApproximateTrimming() {
return approximateTrimming;
}
/**
@@ -202,6 +221,7 @@ public interface RedisStreamCommands {
}
XAddOptions that = (XAddOptions) o;
if (this.nomkstream != that.nomkstream) return false;
if (this.approximateTrimming != that.approximateTrimming) return false;
return ObjectUtils.nullSafeEquals(this.maxlen, that.maxlen);
}
@@ -209,6 +229,7 @@ public interface RedisStreamCommands {
public int hashCode() {
int result = ObjectUtils.nullSafeHashCode(this.maxlen);
result = 31 * result + ObjectUtils.nullSafeHashCode(this.nomkstream);
result = 31 * result + ObjectUtils.nullSafeHashCode(this.approximateTrimming);
return result;
}
}

3
src/main/java/org/springframework/data/redis/connection/jedis/JedisStreamCommands.java
View File

@@ -79,6 +79,9 @@ class JedisStreamCommands implements RedisStreamCommands {
if (options.isNoMkStream()) {
xAddParams.noMkStream();
}
if (options.isApproximateTrimming()) {
xAddParams.approximateTrimming();
}
return connection.invoke()
.from(BinaryJedis::xadd, MultiKeyPipelineBase::xadd, record.getStream(), record.getValue(), xAddParams)

1
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java
View File

@@ -104,6 +104,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands {
args.maxlen(command.getMaxlen());
}
args.nomkstream(command.isNoMkStream());
args.approximateTrimming(command.isApproximateTrimming());
return cmd.xadd(command.getKey(), args, command.getBody())
.map(value -> new CommandResponse<>(command, RecordId.of(value)));

1
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java
View File

@@ -84,6 +84,7 @@ class LettuceStreamCommands implements RedisStreamCommands {
args.maxlen(options.getMaxlen());
}
args.nomkstream(options.isNoMkStream());
args.approximateTrimming(options.isApproximateTrimming());
return connection.invoke().from(RedisStreamAsyncCommands::xadd, record.getStream(), args, record.getValue())
.get(RecordId::of);