Feature - trim method added to RStream object. #1490

redisson/src/main/java/org/redisson/RedissonStream.java

@@ -684,4 +684,24 @@ public class RedissonStream<K, V> extends RedissonExpirable implements RStream<K
         return get(removeAsync(ids));
     }
 
+    @Override
+    public RFuture<Long> trimAsync(int count) {
+        return commandExecutor.writeAsync(getName(), StringCodec.INSTANCE, RedisCommands.XTRIM, "MAXLEN", count);
+    }
+
+    @Override
+    public RFuture<Long> trimNonStrictAsync(int count) {
+        return commandExecutor.writeAsync(getName(), StringCodec.INSTANCE, RedisCommands.XTRIM, "MAXLEN", "~", count);
+    }
+
+    @Override
+    public long trim(int count) {
+        return get(trimAsync(count));
+    }
+
+    @Override
+    public long trimNonStrict(int count) {
+        return get(trimNonStrictAsync(count));
+    }
+
 }

redisson/src/main/java/org/redisson/api/RStream.java

@@ -482,5 +482,21 @@ public interface RStream<K, V> extends RStreamAsync<K, V>, RExpirable {
      * @return deleted messages amount
      */
     long remove(StreamMessageId... ids);
+    
+    /**
+     * Trims stream to specified size
+     * 
+     * @param size - new size of stream
+     * @return number of deleted messages
+     */
+    long trim(int size);
+
+    /**
+     * Trims stream to few tens of entries more than specified length to trim.
+     * 
+     * @param size - new size of stream
+     * @return number of deleted messages
+     */
+    long trimNonStrict(int size);
 
 }

redisson/src/main/java/org/redisson/api/RStreamAsync.java

@@ -488,5 +488,21 @@ public interface RStreamAsync<K, V> extends RExpirableAsync {
      * @return deleted messages amount
      */
     RFuture<Long> removeAsync(StreamMessageId... ids);
+
+    /**
+     * Trims stream to specified size
+     * 
+     * @param size - new size of stream
+     * @return number of deleted messages
+     */
+    RFuture<Long> trimAsync(int size);
+
+    /**
+     * Trims stream to few tens of entries more than specified length to trim.
+     * 
+     * @param size - new size of stream
+     * @return number of deleted messages
+     */
+    RFuture<Long> trimNonStrictAsync(int size);
     
 }

redisson/src/main/java/org/redisson/client/protocol/RedisCommands.java

@@ -408,6 +408,7 @@ public interface RedisCommands {
     RedisStrictCommand<Long> XLEN = new RedisStrictCommand<Long>("XLEN");
     RedisStrictCommand<Long> XACK = new RedisStrictCommand<Long>("XACK");
     RedisStrictCommand<Long> XDEL = new RedisStrictCommand<Long>("XDEL");
+    RedisStrictCommand<Long> XTRIM = new RedisStrictCommand<Long>("XTRIM");
     RedisCommand<Object> XPENDING = new RedisCommand<Object>("XPENDING", 
             new ListMultiDecoder(new ObjectListReplayDecoder(), new ObjectListReplayDecoder(ListMultiDecoder.RESET), new PendingResultDecoder()));
     RedisCommand<Object> XPENDING_ENTRIES = new RedisCommand<Object>("XPENDING",