RScoredSortedSet.pollFirst and pollLast with count parameter added

redisson/src/main/java/org/redisson/RedissonScoredSortedSet.java

@@ -36,6 +36,7 @@ import org.redisson.client.codec.Codec;
 import org.redisson.client.codec.DoubleCodec;
 import org.redisson.client.codec.LongCodec;
 import org.redisson.client.codec.ScanCodec;
+import org.redisson.client.protocol.RedisCommand;
 import org.redisson.client.protocol.RedisCommands;
 import org.redisson.client.protocol.ScoredEntry;
 import org.redisson.client.protocol.decoder.ListScanResult;
@@ -89,25 +90,52 @@ public class RedissonScoredSortedSet<V> extends RedissonExpirable implements RSc
         return get(pollLastAsync());
     }
 
+    @Override
+    public Collection<V> pollFirst(int count) {
+        return get(pollFirstAsync(count));
+    }
+    
+    @Override
+    public Collection<V> pollLast(int count) {
+        return get(pollLastAsync(count));
+    }
+    
+    @Override
+    public RFuture<Collection<V>> pollFirstAsync(int count) {
+        if (count <= 0) {
+            return RedissonPromise.<Collection<V>>newSucceededFuture(Collections.<V>emptyList());
+        }
+
+        return poll(0, count-1, RedisCommands.EVAL_LIST);
+    }
+    
+    @Override
+    public RFuture<Collection<V>> pollLastAsync(int count) {
+        if (count <= 0) {
+            return RedissonPromise.<Collection<V>>newSucceededFuture(Collections.<V>emptyList());
+        }
+        return poll(-count, -1, RedisCommands.EVAL_LIST);
+    }
+    
     @Override
     public RFuture<V> pollFirstAsync() {
-        return poll(0);
+        return poll(0, 0, RedisCommands.EVAL_FIRST_LIST);
     }
 
     @Override
     public RFuture<V> pollLastAsync() {
-        return poll(-1);
+        return poll(-1, -1, RedisCommands.EVAL_FIRST_LIST);
     }
 
-    private RFuture<V> poll(int index) {
-        return commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_OBJECT,
+    private <T> RFuture<T> poll(int from, int to, RedisCommand<?> command) {
+        return commandExecutor.evalWriteAsync(getName(), codec, command,
                 "local v = redis.call('zrange', KEYS[1], ARGV[1], ARGV[2]); "
-                + "if v[1] ~= nil then "
+                + "if #v > 0 then "
                     + "redis.call('zremrangebyrank', KEYS[1], ARGV[1], ARGV[2]); "
-                    + "return v[1]; "
+                    + "return v; "
                 + "end "
-                + "return nil;",
-                Collections.<Object>singletonList(getName()), index, index);
+                + "return v;",
+                Collections.<Object>singletonList(getName()), from, to);
     }
 
     @Override

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

@@ -45,16 +45,64 @@ public interface RScoredSortedSet<V> extends RScoredSortedSetAsync<V>, Iterable<
      */
     <KOut, VOut> RCollectionMapReduce<V, KOut, VOut> mapReduce();
     
+    /**
+     * Removes and returns the head elements or {@code null} if this sorted set is empty.
+     *
+     * @param count - elements amount
+     * @return the head element, 
+     *         or {@code null} if this sorted set is empty
+     */
+    Collection<V> pollFirst(int count);
+
+    /**
+     * Removes and returns the tail elements or {@code null} if this sorted set is empty.
+     * 
+     * @param count - elements amount
+     * @return the tail element or {@code null} if this sorted set is empty
+     */
+    Collection<V> pollLast(int count);
+    
+    /**
+     * Removes and returns the head element or {@code null} if this sorted set is empty.
+     *
+     * @return the head element, 
+     *         or {@code null} if this sorted set is empty
+     */
     V pollFirst();
 
+    /**
+     * Removes and returns the tail element or {@code null} if this sorted set is empty.
+     *
+     * @return the tail element or {@code null} if this sorted set is empty
+     */
     V pollLast();
 
+    /**
+     * Returns the head element or {@code null} if this sorted set is empty.
+     *
+     * @return the head element or {@code null} if this sorted set is empty
+     */
     V first();
 
+    /**
+     * Returns the tail element or {@code null} if this sorted set is empty.
+     *
+     * @return the tail element or {@code null} if this sorted set is empty
+     */
     V last();
-    
+
+    /**
+     * Returns score of the tail element or returns {@code null} if this sorted set is empty.
+     *
+     * @return the tail element or {@code null} if this sorted set is empty
+     */
     Double firstScore();
-    
+
+    /**
+     * Returns score of the head element or returns {@code null} if this sorted set is empty.
+     *
+     * @return the tail element or {@code null} if this sorted set is empty
+     */
     Double lastScore();
 
     Long addAll(Map<V, Double> objects);

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

@@ -30,16 +30,64 @@ import org.redisson.client.protocol.ScoredEntry;
  */
 public interface RScoredSortedSetAsync<V> extends RExpirableAsync, RSortableAsync<Set<V>> {
 
-    RFuture<V> pollLastAsync();
+    /**
+     * Removes and returns the head elements or {@code null} if this sorted set is empty.
+     *
+     * @param count - elements amount
+     * @return the head element, 
+     *         or {@code null} if this sorted set is empty
+     */
+    RFuture<Collection<V>> pollFirstAsync(int count);
+
+    /**
+     * Removes and returns the tail elements or {@code null} if this sorted set is empty.
+     *
+     * @param count - elements amount
+     * @return the tail element or {@code null} if this sorted set is empty
+     */
+    RFuture<Collection<V>> pollLastAsync(int count);
 
+    /**
+     * Removes and returns the head element or {@code null} if this sorted set is empty.
+     *
+     * @return the head element, 
+     *         or {@code null} if this sorted set is empty
+     */
     RFuture<V> pollFirstAsync();
 
+    /**
+     * Removes and returns the tail element or {@code null} if this sorted set is empty.
+     *
+     * @return the tail element or {@code null} if this sorted set is empty
+     */
+    RFuture<V> pollLastAsync();
+
+    /**
+     * Returns the head element or {@code null} if this sorted set is empty.
+     *
+     * @return the head element or {@code null} if this sorted set is empty
+     */
     RFuture<V> firstAsync();
 
+    /**
+     * Returns the tail element or {@code null} if this sorted set is empty.
+     *
+     * @return the tail element or {@code null} if this sorted set is empty
+     */
     RFuture<V> lastAsync();
-    
+
+    /**
+     * Returns score of the head element or returns {@code null} if this sorted set is empty.
+     *
+     * @return the tail element or {@code null} if this sorted set is empty
+     */
     RFuture<Double> firstScoreAsync();
-    
+
+    /**
+     * Returns score of the tail element or returns {@code null} if this sorted set is empty.
+     *
+     * @return the tail element or {@code null} if this sorted set is empty
+     */
     RFuture<Double> lastScoreAsync();
 
     RFuture<Long> addAllAsync(Map<V, Double> objects);
@@ -48,8 +96,20 @@ public interface RScoredSortedSetAsync<V> extends RExpirableAsync, RSortableAsyn
 
     RFuture<Integer> removeRangeByRankAsync(int startIndex, int endIndex);
 
+    /**
+     * Returns rank of value, with the scores ordered from low to high.
+     * 
+     * @param o - object
+     * @return rank or <code>null</code> if value does not exist
+     */
     RFuture<Integer> rankAsync(V o);
     
+    /**
+     * Returns rank of value, with the scores ordered from high to low.
+     * 
+     * @param o - object
+     * @return rank or <code>null</code> if value does not exist
+     */
     RFuture<Integer> revRankAsync(V o);
 
     /**

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

@@ -36,13 +36,14 @@ import org.redisson.client.protocol.convertor.DoubleNullSafeReplayConvertor;
 import org.redisson.client.protocol.convertor.DoubleReplayConvertor;
 import org.redisson.client.protocol.convertor.IntegerReplayConvertor;
 import org.redisson.client.protocol.convertor.KeyValueConvertor;
-import org.redisson.client.protocol.convertor.TimeObjectDecoder;
 import org.redisson.client.protocol.convertor.LongReplayConvertor;
+import org.redisson.client.protocol.convertor.TimeObjectDecoder;
 import org.redisson.client.protocol.convertor.TrueReplayConvertor;
 import org.redisson.client.protocol.convertor.TypeConvertor;
 import org.redisson.client.protocol.convertor.VoidReplayConvertor;
 import org.redisson.client.protocol.decoder.ClusterNodesDecoder;
 import org.redisson.client.protocol.decoder.KeyValueObjectDecoder;
+import org.redisson.client.protocol.decoder.ListFirstObjectDecoder;
 import org.redisson.client.protocol.decoder.ListMultiDecoder;
 import org.redisson.client.protocol.decoder.ListResultReplayDecoder;
 import org.redisson.client.protocol.decoder.ListScanResult;
@@ -51,7 +52,6 @@ import org.redisson.client.protocol.decoder.Long2MultiDecoder;
 import org.redisson.client.protocol.decoder.LongMultiDecoder;
 import org.redisson.client.protocol.decoder.MapScanResult;
 import org.redisson.client.protocol.decoder.MapScanResultReplayDecoder;
-import org.redisson.client.protocol.decoder.ObjectFirstResultReplayDecoder;
 import org.redisson.client.protocol.decoder.ObjectFirstScoreReplayDecoder;
 import org.redisson.client.protocol.decoder.ObjectListReplayDecoder;
 import org.redisson.client.protocol.decoder.ObjectMapEntryReplayDecoder;
@@ -119,9 +119,11 @@ public interface RedisCommands {
     RedisStrictCommand<Double> ZSCORE = new RedisStrictCommand<Double>("ZSCORE", new DoubleReplayConvertor());
     RedisCommand<Integer> ZRANK_INT = new RedisCommand<Integer>("ZRANK", new IntegerReplayConvertor());
     RedisCommand<Integer> ZREVRANK_INT = new RedisCommand<Integer>("ZREVRANK", new IntegerReplayConvertor());
-    RedisCommand<Object> ZRANGE_SINGLE = new RedisCommand<Object>("ZRANGE", new ObjectFirstResultReplayDecoder<Object>());
+    RedisCommand<Object> ZRANGE_SINGLE = new RedisCommand<Object>("ZRANGE", new ListFirstObjectDecoder());
     RedisStrictCommand<Double> ZRANGE_SINGLE_SCORE = new RedisStrictCommand<Double>("ZRANGE", new ObjectFirstScoreReplayDecoder());
     RedisCommand<List<Object>> ZRANGE = new RedisCommand<List<Object>>("ZRANGE", new ObjectListReplayDecoder<Object>());
+    RedisCommand<List<Object>> ZPOPMIN = new RedisCommand<List<Object>>("ZPOPMIN", new ObjectListReplayDecoder<Object>());
+    RedisCommand<List<Object>> ZPOPMAX = new RedisCommand<List<Object>>("ZPOPMAX", new ObjectListReplayDecoder<Object>());
     RedisStrictCommand<Integer> ZREMRANGEBYRANK = new RedisStrictCommand<Integer>("ZREMRANGEBYRANK", new IntegerReplayConvertor());
     RedisStrictCommand<Integer> ZREMRANGEBYSCORE = new RedisStrictCommand<Integer>("ZREMRANGEBYSCORE", new IntegerReplayConvertor());
     RedisStrictCommand<Integer> ZREMRANGEBYLEX = new RedisStrictCommand<Integer>("ZREMRANGEBYLEX", new IntegerReplayConvertor());
@@ -228,6 +230,7 @@ public interface RedisCommands {
     RedisStrictCommand<Long> EVAL_LONG = new RedisStrictCommand<Long>("EVAL");
     RedisStrictCommand<Long> EVAL_LONG_SAFE = new RedisStrictCommand<Long>("EVAL", new LongReplayConvertor());
     RedisStrictCommand<Void> EVAL_VOID = new RedisStrictCommand<Void>("EVAL", new VoidReplayConvertor());
+    RedisCommand<Object> EVAL_FIRST_LIST = new RedisCommand<Object>("EVAL", new ListFirstObjectDecoder());
     RedisCommand<List<Object>> EVAL_LIST = new RedisCommand<List<Object>>("EVAL", new ObjectListReplayDecoder<Object>());
     RedisCommand<Set<Object>> EVAL_SET = new RedisCommand<Set<Object>>("EVAL", new ObjectSetReplayDecoder<Object>());
     RedisCommand<Object> EVAL_OBJECT = new RedisCommand<Object>("EVAL");

redisson/src/main/java/org/redisson/client/protocol/decoder/ObjectFirstResultReplayDecoder.java

@@ -1,41 +0,0 @@
-/**
- * Copyright 2018 Nikita Koksharov
- *
- * 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.redisson.client.protocol.decoder;
-
-import java.util.List;
-
-import org.redisson.client.handler.State;
-import org.redisson.client.protocol.Decoder;
-
-/**
- * 
- * @author Nikita Koksharov
- *
- * @param <T> type
- */
-public class ObjectFirstResultReplayDecoder<T> implements MultiDecoder<T> {
-
-    @Override
-    public T decode(List<Object> parts, State state) {
-        return (T) parts.get(0);
-    }
-
-    @Override
-    public Decoder<Object> getDecoder(int paramNum, State state) {
-        return null;
-    }
-
-}

redisson/src/main/java/org/redisson/client/protocol/decoder/ObjectFirstScoreReplayDecoder.java

@@ -39,6 +39,9 @@ public class ObjectFirstScoreReplayDecoder implements MultiDecoder<Double> {
     
     @Override
     public Double decode(List<Object> parts, State state) {
+        if (parts.isEmpty()) {
+            return null;
+        }
         return (Double) parts.get(1);
     }
 

redisson/src/test/java/org/redisson/RedissonScoredSortedSetTest.java

@@ -249,7 +249,33 @@ public class RedissonScoredSortedSetTest extends BaseTest {
         Assert.assertEquals("c", set.pollLast());
         assertThat(set).containsExactly("a", "b");
     }
+    
+    @Test
+    public void testPollLastAmount() {
+        RScoredSortedSet<String> set = redisson.getScoredSortedSet("simple");
+        assertThat(set.pollLast(2)).isEmpty();
+
+        set.add(0.1, "a");
+        set.add(0.2, "b");
+        set.add(0.3, "c");
+
+        assertThat(set.pollLast(2)).containsExactly("b", "c");
+        assertThat(set).containsExactly("a");
+    }
+
+    @Test
+    public void testPollFistAmount() {
+        RScoredSortedSet<String> set = redisson.getScoredSortedSet("simple");
+        assertThat(set.pollFirst(2)).isEmpty();
+
+        set.add(0.1, "a");
+        set.add(0.2, "b");
+        set.add(0.3, "c");
 
+        assertThat(set.pollFirst(2)).containsExactly("a", "b");
+        assertThat(set).containsExactly("c");
+    }
+    
     @Test
     public void testPollFirst() {
         RScoredSortedSet<String> set = redisson.getScoredSortedSet("simple");
@@ -271,6 +297,9 @@ public class RedissonScoredSortedSetTest extends BaseTest {
         set.add(0.3, "c");
         set.add(0.4, "d");
 
+        RScoredSortedSet<String> set2 = redisson.getScoredSortedSet("simple2");
+        assertThat(set2.first()).isNull();
+        assertThat(set2.last()).isNull();
         Assert.assertEquals("a", set.first());
         Assert.assertEquals("d", set.last());
     }
@@ -283,6 +312,9 @@ public class RedissonScoredSortedSetTest extends BaseTest {
         set.add(0.3, "c");
         set.add(0.4, "d");
 
+        RScoredSortedSet<String> set2 = redisson.getScoredSortedSet("simple2");
+        assertThat(set2.firstScore()).isNull();
+        assertThat(set2.lastScore()).isNull();
         assertThat(set.firstScore()).isEqualTo(0.1);
         assertThat(set.lastScore()).isEqualTo(0.4);
     }