DATAREDIS-1085 - Support command xtrim "~"

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

@@ -3841,7 +3841,16 @@ public List<StringRecord> xRevRange(String key, org.springframework.data.domain.
 	 */
 	@Override
 	public Long xTrim(String key, long count) {
-		return convertAndReturn(delegate.xTrim(serialize(key), count), identityConverter);
+		return xTrim(key, count, false);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.StringRedisConnection#xTrim(java.lang.String, long, boolean)
+	 */
+	@Override
+	public Long xTrim(String key, long count, boolean approximateTrimming) {
+		return convertAndReturn(delegate.xTrim(serialize(key), count, approximateTrimming), identityConverter);
 	}
 
 	/*
@@ -4022,7 +4031,16 @@ public List<ByteRecord> xRevRange(byte[] key, org.springframework.data.domain.Ra
 	 */
 	@Override
 	public Long xTrim(byte[] key, long count) {
-		return delegate.xTrim(key, count);
+		return xTrim(key, count, false);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xTrim(byte[], long, boolean)
+	 */
+	@Override
+	public Long xTrim(byte[] key, long count, boolean approximateTrimming) {
+		return delegate.xTrim(key, count, approximateTrimming);
 	}
 
 	/**

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

@@ -609,7 +609,13 @@ default List<ByteRecord> xRevRange(byte[] key, org.springframework.data.domain.R
 	@Override
 	@Deprecated
 	default Long xTrim(byte[] key, long count) {
-		return streamCommands().xTrim(key, count);
+		return xTrim(key, count, false);
+	}
+
+	@Override
+	@Deprecated
+	default Long xTrim(byte[] key, long count, boolean approximateTrimming) {
+		return streamCommands().xTrim(key, count, approximateTrimming);
 	}
 
 	// LIST COMMANDS

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

@@ -1376,11 +1376,12 @@ default Flux<ByteBufferRecord> xRevRange(ByteBuffer key, Range<String> range, Li
 	class TrimCommand extends KeyCommand {
 
 		private @Nullable Long count;
+		private boolean approximateTrimming;
 
-		private TrimCommand(ByteBuffer key, @Nullable Long count) {
-
+		private TrimCommand(ByteBuffer key, @Nullable Long count, boolean approximateTrimming) {
 			super(key);
 			this.count = count;
+			this.approximateTrimming = approximateTrimming;
 		}
 
 		/**
@@ -1393,18 +1394,19 @@ public static TrimCommand stream(ByteBuffer key) {
 
 			Assert.notNull(key, "Key must not be null!");
 
-			return new TrimCommand(key, null);
+			return new TrimCommand(key, null, false);
 		}
 
 		/**
 		 * Applies the numeric {@literal count}. Constructs a new command instance with all previously configured
 		 * properties.
 		 *
 		 * @param count
+		 * @param approximateTrimming
 		 * @return a new {@link TrimCommand} with {@literal count} applied.
 		 */
-		public TrimCommand to(long count) {
-			return new TrimCommand(getKey(), count);
+		public TrimCommand to(long count, boolean approximateTrimming) {
+			return new TrimCommand(getKey(), count, approximateTrimming);
 		}
 
 		/**
@@ -1414,6 +1416,10 @@ public TrimCommand to(long count) {
 		public Long getCount() {
 			return count;
 		}
+
+		public boolean isApproximateTrimming() {
+			return approximateTrimming;
+		}
 	}
 
 	/**
@@ -1425,10 +1431,23 @@ public Long getCount() {
 	 * @see <a href="https://redis.io/commands/xtrim">Redis Documentation: XTRIM</a>
 	 */
 	default Mono<Long> xTrim(ByteBuffer key, long count) {
+		return xTrim(key, count, false);
+	}
+
+	/**
+	 * Trims the stream to {@code count} elements.
+	 *
+	 * @param key the stream key.
+	 * @param count length of the stream.
+	 * @param approximateTrimming the trimming must be performed in a approximated way in order to maximize performances.
+	 * @return {@link Mono} emitting the number of removed entries.
+	 * @see <a href="https://redis.io/commands/xtrim">Redis Documentation: XTRIM</a>
+	 */
+	default Mono<Long> xTrim(ByteBuffer key, long count, boolean approximateTrimming) {
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return xTrim(Mono.just(TrimCommand.stream(key).to(count))).next().map(NumericResponse::getOutput);
+		return xTrim(Mono.just(TrimCommand.stream(key).to(count, approximateTrimming))).next().map(NumericResponse::getOutput);
 	}
 
 	/**

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

@@ -873,4 +873,16 @@ default List<ByteRecord> xRevRange(byte[] key, Range<String> range) {
 	 */
 	@Nullable
 	Long xTrim(byte[] key, long count);
+
+	/**
+	 * Trims the stream to {@code count} elements.
+	 *
+	 * @param key the stream key.
+	 * @param count length of the stream.
+	 * @param approximateTrimming the trimming must be performed in a approximated way in order to maximize performances.
+	 * @return number of removed entries. {@literal null} when used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/xtrim">Redis Documentation: XTRIM</a>
+	 */
+	@Nullable
+	Long xTrim(byte[] key, long count, boolean approximateTrimming);
 }

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

@@ -2428,4 +2428,17 @@ default List<StringRecord> xRevRange(String key, org.springframework.data.domain
 	 */
 	@Nullable
 	Long xTrim(String key, long count);
+
+	/**
+	 * Trims the stream to {@code count} elements.
+	 *
+	 * @param key the stream key.
+	 * @param count length of the stream.
+	 * @param approximateTrimming the trimming must be performed in a approximated way in order to maximize performances.
+	 * @return number of removed entries. {@literal null} when used in pipeline / transaction.
+	 * @since 2.2
+	 * @see <a href="https://redis.io/commands/xtrim">Redis Documentation: XTRIM</a>
+	 */
+	@Nullable
+	Long xTrim(String key, long count, boolean approximateTrimming);
 }

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

@@ -420,7 +420,7 @@ public Flux<NumericResponse<KeyCommand, Long>> xTrim(Publisher<TrimCommand> comm
 			Assert.notNull(command.getKey(), "Key must not be null!");
 			Assert.notNull(command.getCount(), "Count must not be null!");
 
-			return cmd.xtrim(command.getKey(), command.getCount()).map(value -> new NumericResponse<>(command, value));
+			return cmd.xtrim(command.getKey(), command.isApproximateTrimming(), command.getCount()).map(value -> new NumericResponse<>(command, value));
 		}));
 	}
 

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

@@ -662,19 +662,27 @@ public List<ByteRecord> xRevRange(byte[] key, Range<String> range, Limit limit)
 	 */
 	@Override
 	public Long xTrim(byte[] key, long count) {
+		return xTrim(key, count, false);
+	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xTrim(byte[], long, boolean)
+	 */
+	@Override
+	public Long xTrim(byte[] key, long count, boolean approximateTrimming) {
 		Assert.notNull(key, "Key must not be null!");
 
 		try {
 			if (isPipelined()) {
-				pipeline(connection.newLettuceResult(getAsyncConnection().xtrim(key, count)));
+				pipeline(connection.newLettuceResult(getAsyncConnection().xtrim(key, approximateTrimming, count)));
 				return null;
 			}
 			if (isQueueing()) {
-				transaction(connection.newLettuceResult(getAsyncConnection().xtrim(key, count)));
+				transaction(connection.newLettuceResult(getAsyncConnection().xtrim(key, approximateTrimming, count)));
 				return null;
 			}
-			return getConnection().xtrim(key, count);
+			return getConnection().xtrim(key, approximateTrimming, count);
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}

src/main/java/org/springframework/data/redis/core/BoundStreamOperations.java

@@ -208,4 +208,15 @@ default List<MapRecord<K, HK, HV>> reverseRange(Range<String> range) {
 	 */
 	@Nullable
 	Long trim(long count);
+
+	/**
+	 * Trims the stream to {@code count} elements.
+	 *
+	 * @param count length of the stream.
+	 * @param approximateTrimming the trimming must be performed in a approximated way in order to maximize performances.
+	 * @return number of removed entries. {@literal null} when used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/xtrim">Redis Documentation: XTRIM</a>
+	 */
+	@Nullable
+	Long trim(long count, boolean approximateTrimming);
 }

src/main/java/org/springframework/data/redis/core/DefaultBoundStreamOperations.java

@@ -170,7 +170,17 @@ public List<MapRecord<K, HK, HV>> reverseRange(Range<String> range, Limit limit)
 	@Nullable
 	@Override
 	public Long trim(long count) {
-		return ops.trim(getKey(), count);
+		return trim(count, false);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.BoundStreamOperations#trim(long,boolean)
+	 */
+	@Nullable
+	@Override
+	public Long trim(long count, boolean approximateTrimming) {
+		return ops.trim(getKey(), count, approximateTrimming);
 	}
 
 	/* 

src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java

@@ -341,10 +341,18 @@ public Flux<MapRecord<K, HK, HV>> reverseRange(K key, Range<String> range, Limit
 	 */
 	@Override
 	public Mono<Long> trim(K key, long count) {
+		return trim(key, count, false);
+	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveStreamOperations#trim(java.lang.Object, long, boolean)
+	 */
+	@Override
+	public Mono<Long> trim(K key, long count, boolean approximateTrimming) {
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.xTrim(rawKey(key), count));
+		return createMono(connection -> connection.xTrim(rawKey(key), count, approximateTrimming));
 	}
 
 	@Override

src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java

@@ -336,6 +336,12 @@ public Long trim(K key, long count) {
 		return execute(connection -> connection.xTrim(rawKey, count), true);
 	}
 
+	@Override
+	public Long trim(K key, long count, boolean approximateTrimming) {
+		byte[] rawKey = rawKey(key);
+		return execute(connection -> connection.xTrim(rawKey, count, approximateTrimming), true);
+	}
+
 	@Override
 	public <V> HashMapper<V, HK, HV> getHashMapper(Class<V> targetType) {
 		return objectMapper.getHashMapper(targetType);

src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java

@@ -544,6 +544,17 @@ default <V> Flux<ObjectRecord<K, V>> reverseRange(Class<V> targetType, K key, Ra
 	 */
 	Mono<Long> trim(K key, long count);
 
+	/**
+	 * Trims the stream to {@code count} elements.
+	 *
+	 * @param key the stream key.
+	 * @param count length of the stream.
+	 * @param approximateTrimming the trimming must be performed in a approximated way in order to maximize performances.
+	 * @return number of removed entries.
+	 * @see <a href="https://redis.io/commands/xtrim">Redis Documentation: XTRIM</a>
+	 */
+	Mono<Long> trim(K key, long count, boolean approximateTrimming);
+
 	/**
 	 * Get the {@link HashMapper} for a specific type.
 	 *

src/main/java/org/springframework/data/redis/core/StreamOperations.java

@@ -525,6 +525,18 @@ default <V> List<ObjectRecord<K, V>> reverseRange(Class<V> targetType, K key, Ra
 	@Nullable
 	Long trim(K key, long count);
 
+	/**
+	 * Trims the stream to {@code count} elements.
+	 *
+	 * @param key the stream key.
+	 * @param count length of the stream.
+	 * @param approximateTrimming the trimming must be performed in a approximated way in order to maximize performances.
+	 * @return number of removed entries. {@literal null} when used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/xtrim">Redis Documentation: XTRIM</a>
+	 */
+	@Nullable
+	Long trim(K key, long count, boolean approximateTrimming);
+
 	/**
 	 * Get the {@link HashMapper} for a specific type.
 	 *

src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionPipelineTests.java

@@ -1726,6 +1726,13 @@ public void xTrimShouldDelegateAndConvertCorrectly() {
 		super.xTrimShouldDelegateAndConvertCorrectly();
 	}
 
+	@Test
+	public void xTrimApproximateShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(1L)).when(nativeConnection).closePipeline();
+		super.xTrimApproximateShouldDelegateAndConvertCorrectly();
+	}
+
 	protected List<Object> getResults() {
 		return connection.closePipeline();
 	}

src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionPipelineTxTests.java

@@ -1846,6 +1846,13 @@ public void xTrimShouldDelegateAndConvertCorrectly() {
 		super.xTrimShouldDelegateAndConvertCorrectly();
 	}
 
+	@Test
+	public void xTrimApproximateShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(Arrays.asList(1L))).when(nativeConnection).closePipeline();
+		super.xTrimApproximateShouldDelegateAndConvertCorrectly();
+	}
+
 	@SuppressWarnings("unchecked")
 	protected List<Object> getResults() {
 		connection.exec();

src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java

@@ -2169,6 +2169,15 @@ public void xTrimShouldDelegateAndConvertCorrectly() {
 		Assertions.assertThat(getResults()).containsExactly(1L);
 	}
 
+	@Test
+	public void xTrimApproximateShouldDelegateAndConvertCorrectly() {
+
+		doReturn(1L).when(nativeConnection).xTrim(any(), anyLong(), anyBoolean());
+
+		actual.add(connection.xTrim("key", 2L, true));
+		Assertions.assertThat(getResults()).containsExactly(1L);
+	}
+
 	protected List<Object> getResults() {
 		return actual;
 	}

src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTxTests.java

@@ -1733,6 +1733,13 @@ public void xTrimShouldDelegateAndConvertCorrectly() {
 		super.xTrimShouldDelegateAndConvertCorrectly();
 	}
 
+	@Test
+	public void xTrimApproximateShouldDelegateAndConvertCorrectly() {
+
+		doReturn(Arrays.asList(1L)).when(nativeConnection).exec();
+		super.xTrimApproximateShouldDelegateAndConvertCorrectly();
+	}
+
 	protected List<Object> getResults() {
 		return connection.exec();
 	}