Feature - Add readIntersection() method to RScoredSortedSet object #3352

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

@@ -866,7 +866,65 @@ public class RedissonScoredSortedSet<V> extends RedissonExpirable implements RSc
         args.add(aggregate.name());
         return commandExecutor.writeAsync(getName(), LongCodec.INSTANCE, RedisCommands.ZINTERSTORE_INT, args.toArray());
     }
-    
+
+    @Override
+    public Collection<V> readIntersection(String... names) {
+        return get(readIntersectionAsync(names));
+    }
+
+    @Override
+    public RFuture<Collection<V>> readIntersectionAsync(String... names) {
+        return readIntersectionAsync(Aggregate.SUM, names);
+    }
+
+    @Override
+    public Collection<V> readIntersection(Aggregate aggregate, String... names) {
+        return get(readIntersectionAsync(aggregate, names));
+    }
+
+    @Override
+    public RFuture<Collection<V>> readIntersectionAsync(Aggregate aggregate, String... names) {
+        List<Object> args = new ArrayList<Object>(names.length + 4);
+        args.add(names.length + 1);
+        args.add(getName());
+        args.addAll(Arrays.asList(names));
+        args.add("AGGREGATE");
+        args.add(aggregate.name());
+        return commandExecutor.writeAsync(getName(), codec, RedisCommands.ZINTER, args.toArray());
+    }
+
+    @Override
+    public Collection<V> readIntersection(Map<String, Double> nameWithWeight) {
+        return get(readIntersectionAsync(nameWithWeight));
+    }
+
+    @Override
+    public RFuture<Collection<V>> readIntersectionAsync(Map<String, Double> nameWithWeight) {
+        return readIntersectionAsync(Aggregate.SUM, nameWithWeight);
+    }
+
+    @Override
+    public Collection<V> readIntersection(Aggregate aggregate, Map<String, Double> nameWithWeight) {
+        return get(readIntersectionAsync(aggregate, nameWithWeight));
+    }
+
+    @Override
+    public RFuture<Collection<V>> readIntersectionAsync(Aggregate aggregate, Map<String, Double> nameWithWeight) {
+        List<Object> args = new ArrayList<Object>(nameWithWeight.size()*2 + 5);
+        args.add(nameWithWeight.size() + 1);
+        args.add(getName());
+        args.addAll(nameWithWeight.keySet());
+        args.add("WEIGHTS");
+        List<String> weights = new ArrayList<String>();
+        for (Double weight : nameWithWeight.values()) {
+            weights.add(BigDecimal.valueOf(weight).toPlainString());
+        }
+        args.addAll(weights);
+        args.add("AGGREGATE");
+        args.add(aggregate.name());
+        return commandExecutor.writeAsync(getName(), codec, RedisCommands.ZINTER, args.toArray());
+    }
+
     @Override
     public int union(String... names) {
         return get(unionAsync(names));        
@@ -942,8 +1000,9 @@ public class RedissonScoredSortedSet<V> extends RedissonExpirable implements RSc
 
     @Override
     public RFuture<Collection<V>> readUnionAsync(Aggregate aggregate, String... names) {
-        List<Object> args = new ArrayList<Object>(names.length + 4);
-        args.add(names.length);
+        List<Object> args = new ArrayList<>(names.length + 4);
+        args.add(names.length + 1);
+        args.add(getName());
         args.addAll(Arrays.asList(names));
         args.add("AGGREGATE");
         args.add(aggregate.name());
@@ -968,7 +1027,8 @@ public class RedissonScoredSortedSet<V> extends RedissonExpirable implements RSc
     @Override
     public RFuture<Collection<V>> readUnionAsync(Aggregate aggregate, Map<String, Double> nameWithWeight) {
         List<Object> args = new ArrayList<Object>(nameWithWeight.size()*2 + 5);
-        args.add(nameWithWeight.size());
+        args.add(nameWithWeight.size() + 1);
+        args.add(getName());
         args.addAll(nameWithWeight.keySet());
         args.add("WEIGHTS");
         List<String> weights = new ArrayList<String>();

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

@@ -766,6 +766,53 @@ public interface RScoredSortedSet<V> extends RScoredSortedSetAsync<V>, Iterable<
      */
     int intersection(Aggregate aggregate, Map<String, Double> nameWithWeight);
 
+    /**
+     * Intersect provided ScoredSortedSets
+     * with current ScoredSortedSet without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param names - names of ScoredSortedSet
+     * @return result of intersection
+     */
+    Collection<V> readIntersection(String... names);
+
+    /**
+     * Intersect provided ScoredSortedSets with current ScoredSortedSet using defined aggregation method
+     * without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param aggregate - score aggregation mode
+     * @param names - names of ScoredSortedSet
+     * @return result of intersection
+     */
+    Collection<V> readIntersection(Aggregate aggregate, String... names);
+
+    /**
+     * Intersect provided ScoredSortedSets mapped to weight multiplier
+     * with current ScoredSortedSet without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
+     * @return result of intersection
+     */
+    Collection<V> readIntersection(Map<String, Double> nameWithWeight);
+
+    /**
+     * Intersect provided ScoredSortedSets mapped to weight multiplier
+     * with current ScoredSortedSet using defined aggregation method
+     * without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param aggregate - score aggregation mode
+     * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
+     * @return result of intersection
+     */
+    Collection<V> readIntersection(Aggregate aggregate, Map<String, Double> nameWithWeight);
+
     /**
      * Union provided ScoredSortedSets 
      * and store result to current ScoredSortedSet
@@ -808,28 +855,34 @@ public interface RScoredSortedSet<V> extends RScoredSortedSetAsync<V>, Iterable<
     /**
      * Union ScoredSortedSets specified by name with current ScoredSortedSet
      * without state change.
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param names - names of ScoredSortedSet
-     * @return length of union
+     * @return result of union
      */
     Collection<V> readUnion(String... names);
 
     /**
      * Union ScoredSortedSets specified by name with defined aggregation method
      * and current ScoredSortedSet without state change.
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param aggregate - score aggregation mode
      * @param names - names of ScoredSortedSet
-     * @return length of union
+     * @return result of union
      */
     Collection<V> readUnion(Aggregate aggregate, String... names);
 
     /**
      * Union provided ScoredSortedSets mapped to weight multiplier
      * and current ScoredSortedSet without state change.
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
-     * @return length of union
+     * @return result of union
      */
     Collection<V> readUnion(Map<String, Double> nameWithWeight);
 
@@ -837,10 +890,12 @@ public interface RScoredSortedSet<V> extends RScoredSortedSetAsync<V>, Iterable<
      * Union provided ScoredSortedSets mapped to weight multiplier
      * with defined aggregation method
      * and current ScoredSortedSet without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param aggregate - score aggregation mode
      * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
-     * @return length of union
+     * @return result of union
      */
     Collection<V> readUnion(Aggregate aggregate, Map<String, Double> nameWithWeight);
 

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

@@ -633,6 +633,53 @@ public interface RScoredSortedSetAsync<V> extends RExpirableAsync, RSortableAsyn
      */
     RFuture<Integer> intersectionAsync(Aggregate aggregate, Map<String, Double> nameWithWeight);
 
+    /**
+     * Intersect provided ScoredSortedSets
+     * with current ScoredSortedSet without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param names - names of ScoredSortedSet
+     * @return result of intersection
+     */
+    RFuture<Collection<V>> readIntersectionAsync(String... names);
+
+    /**
+     * Intersect provided ScoredSortedSets with current ScoredSortedSet using defined aggregation method
+     * without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param aggregate - score aggregation mode
+     * @param names - names of ScoredSortedSet
+     * @return result of intersection
+     */
+    RFuture<Collection<V>> readIntersectionAsync(Aggregate aggregate, String... names);
+
+    /**
+     * Intersect provided ScoredSortedSets mapped to weight multiplier
+     * with current ScoredSortedSet without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
+     * @return result of intersection
+     */
+    RFuture<Collection<V>> readIntersectionAsync(Map<String, Double> nameWithWeight);
+
+    /**
+     * Intersect provided ScoredSortedSets mapped to weight multiplier
+     * with current ScoredSortedSet using defined aggregation method
+     * without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param aggregate - score aggregation mode
+     * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
+     * @return result of intersection
+     */
+    RFuture<Collection<V>> readIntersectionAsync(Aggregate aggregate, Map<String, Double> nameWithWeight);
+
     /**
      * Union provided ScoredSortedSets 
      * and store result to current ScoredSortedSet
@@ -675,28 +722,34 @@ public interface RScoredSortedSetAsync<V> extends RExpirableAsync, RSortableAsyn
     /**
      * Union ScoredSortedSets specified by name with current ScoredSortedSet
      * without state change.
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param names - names of ScoredSortedSet
-     * @return length of union
+     * @return result of union
      */
     RFuture<Collection<V>> readUnionAsync(String... names);
 
     /**
      * Union ScoredSortedSets specified by name with defined aggregation method
      * and current ScoredSortedSet without state change.
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param aggregate - score aggregation mode
      * @param names - names of ScoredSortedSet
-     * @return length of union
+     * @return result of union
      */
     RFuture<Collection<V>> readUnionAsync(Aggregate aggregate, String... names);
 
     /**
      * Union provided ScoredSortedSets mapped to weight multiplier
      * and current ScoredSortedSet without state change.
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
-     * @return length of union
+     * @return result of union
      */
     RFuture<Collection<V>> readUnionAsync(Map<String, Double> nameWithWeight);
 
@@ -704,10 +757,12 @@ public interface RScoredSortedSetAsync<V> extends RExpirableAsync, RSortableAsyn
      * Union provided ScoredSortedSets mapped to weight multiplier
      * with defined aggregation method
      * and current ScoredSortedSet without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param aggregate - score aggregation mode
      * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
-     * @return length of union
+     * @return result of union
      */
     RFuture<Collection<V>> readUnionAsync(Aggregate aggregate, Map<String, Double> nameWithWeight);
 

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

@@ -656,6 +656,53 @@ public interface RScoredSortedSetReactive<V> extends RExpirableReactive, RSortab
      */
     Mono<Integer> intersection(Aggregate aggregate, Map<String, Double> nameWithWeight);
 
+    /**
+     * Intersect provided ScoredSortedSets
+     * with current ScoredSortedSet without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param names - names of ScoredSortedSet
+     * @return result of intersection
+     */
+    Mono<Collection<V>> readIntersection(String... names);
+
+    /**
+     * Intersect provided ScoredSortedSets with current ScoredSortedSet using defined aggregation method
+     * without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param aggregate - score aggregation mode
+     * @param names - names of ScoredSortedSet
+     * @return result of intersection
+     */
+    Mono<Collection<V>> readIntersection(Aggregate aggregate, String... names);
+
+    /**
+     * Intersect provided ScoredSortedSets mapped to weight multiplier
+     * with current ScoredSortedSet without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
+     * @return result of intersection
+     */
+    Mono<Collection<V>> readIntersection(Map<String, Double> nameWithWeight);
+
+    /**
+     * Intersect provided ScoredSortedSets mapped to weight multiplier
+     * with current ScoredSortedSet using defined aggregation method
+     * without state change
+      * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+    *
+     * @param aggregate - score aggregation mode
+     * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
+     * @return result of intersection
+     */
+    Mono<Collection<V>> readIntersection(Aggregate aggregate, Map<String, Double> nameWithWeight);
+
     /**
      * Union provided ScoredSortedSets 
      * and store result to current ScoredSortedSet
@@ -698,28 +745,34 @@ public interface RScoredSortedSetReactive<V> extends RExpirableReactive, RSortab
     /**
      * Union ScoredSortedSets specified by name with current ScoredSortedSet
      * without state change.
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param names - names of ScoredSortedSet
-     * @return length of union
+     * @return result of union
      */
     Mono<Collection<V>> readUnion(String... names);
 
     /**
      * Union ScoredSortedSets specified by name with defined aggregation method
      * and current ScoredSortedSet without state change.
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param aggregate - score aggregation mode
      * @param names - names of ScoredSortedSet
-     * @return length of union
+     * @return result of union
      */
     Mono<Collection<V>> readUnion(Aggregate aggregate, String... names);
 
     /**
      * Union provided ScoredSortedSets mapped to weight multiplier
      * and current ScoredSortedSet without state change.
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
-     * @return length of union
+     * @return result of union
      */
     Mono<Collection<V>> readUnion(Map<String, Double> nameWithWeight);
 
@@ -727,10 +780,12 @@ public interface RScoredSortedSetReactive<V> extends RExpirableReactive, RSortab
      * Union provided ScoredSortedSets mapped to weight multiplier
      * with defined aggregation method
      * and current ScoredSortedSet without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param aggregate - score aggregation mode
      * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
-     * @return length of union
+     * @return result of union
      */
     Mono<Collection<V>> readUnion(Aggregate aggregate, Map<String, Double> nameWithWeight);
 

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

@@ -658,6 +658,53 @@ public interface RScoredSortedSetRx<V> extends RExpirableRx, RSortableRx<Set<V>>
      */
     Single<Integer> intersection(Aggregate aggregate, Map<String, Double> nameWithWeight);
 
+    /**
+     * Intersect provided ScoredSortedSets
+     * with current ScoredSortedSet without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param names - names of ScoredSortedSet
+     * @return result of intersection
+     */
+    Single<Collection<V>> readIntersection(String... names);
+
+    /**
+     * Intersect provided ScoredSortedSets with current ScoredSortedSet using defined aggregation method
+     * without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param aggregate - score aggregation mode
+     * @param names - names of ScoredSortedSet
+     * @return result of intersection
+     */
+    Single<Collection<V>> readIntersection(Aggregate aggregate, String... names);
+
+    /**
+     * Intersect provided ScoredSortedSets mapped to weight multiplier
+     * with current ScoredSortedSet without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
+     * @return result of intersection
+     */
+    Single<Collection<V>> readIntersection(Map<String, Double> nameWithWeight);
+
+    /**
+     * Intersect provided ScoredSortedSets mapped to weight multiplier
+     * with current ScoredSortedSet using defined aggregation method
+     * without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
+     *
+     * @param aggregate - score aggregation mode
+     * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
+     * @return result of intersection
+     */
+    Single<Collection<V>> readIntersection(Aggregate aggregate, Map<String, Double> nameWithWeight);
+
     /**
      * Union provided ScoredSortedSets 
      * and store result to current ScoredSortedSet
@@ -700,28 +747,34 @@ public interface RScoredSortedSetRx<V> extends RExpirableRx, RSortableRx<Set<V>>
     /**
      * Union ScoredSortedSets specified by name with current ScoredSortedSet
      * without state change.
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param names - names of ScoredSortedSet
-     * @return length of union
+     * @return result of union
      */
     Single<Collection<V>> readUnion(String... names);
 
     /**
      * Union ScoredSortedSets specified by name with defined aggregation method
      * and current ScoredSortedSet without state change.
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param aggregate - score aggregation mode
      * @param names - names of ScoredSortedSet
-     * @return length of union
+     * @return result of union
      */
     Single<Collection<V>> readUnion(Aggregate aggregate, String... names);
 
     /**
      * Union provided ScoredSortedSets mapped to weight multiplier
      * and current ScoredSortedSet without state change.
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
-     * @return length of union
+     * @return result of union
      */
     Single<Collection<V>> readUnion(Map<String, Double> nameWithWeight);
 
@@ -729,10 +782,12 @@ public interface RScoredSortedSetRx<V> extends RExpirableRx, RSortableRx<Set<V>>
      * Union provided ScoredSortedSets mapped to weight multiplier
      * with defined aggregation method
      * and current ScoredSortedSet without state change
+     * <p>
+     * Requires <b>Redis 6.2.0 and higher.</b>
      *
      * @param aggregate - score aggregation mode
      * @param nameWithWeight - name of ScoredSortedSet mapped to weight multiplier
-     * @return length of union
+     * @return result of union
      */
     Single<Collection<V>> readUnion(Aggregate aggregate, Map<String, Double> nameWithWeight);
 

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

@@ -95,6 +95,7 @@ public interface RedisCommands {
     RedisStrictCommand<Void> READONLY = new RedisStrictCommand<Void>("READONLY", new VoidReplayConvertor());
 
     RedisCommand<List<Object>> ZUNION = new RedisCommand<>("ZUNION", new ObjectListReplayDecoder<>());
+    RedisCommand<List<Object>> ZINTER = new RedisCommand<>("ZINTER", new ObjectListReplayDecoder<>());
     RedisStrictCommand<Integer> ZUNIONSTORE_INT = new RedisStrictCommand<Integer>("ZUNIONSTORE", new IntegerReplayConvertor());
     RedisStrictCommand<Integer> ZINTERSTORE_INT = new RedisStrictCommand<Integer>("ZINTERSTORE", new IntegerReplayConvertor());
     RedisCommand<Boolean> ZADD_BOOL = new RedisCommand<Boolean>("ZADD", new BooleanAmountReplayConvertor());

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

@@ -1307,6 +1307,22 @@ public class RedissonScoredSortedSetTest extends BaseTest {
         assertThat(set.revRank("three")).isEqualTo(0);
     }
 
+    @Test
+    public void testReadIntersection() {
+        RScoredSortedSet<String> set1 = redisson.getScoredSortedSet("simple1");
+        set1.add(1, "one");
+        set1.add(2, "two");
+        set1.add(2, "four");
+
+        RScoredSortedSet<String> set2 = redisson.getScoredSortedSet("simple2");
+        set2.add(1, "one");
+        set2.add(2, "two");
+        set2.add(3, "three");
+
+        RScoredSortedSet<String> out = redisson.getScoredSortedSet("simple1");
+        assertThat(out.readIntersection(set1.getName(), set2.getName())).containsOnly("one", "two");
+    }
+
     @Test
     public void testIntersection() {
         RScoredSortedSet<String> set1 = redisson.getScoredSortedSet("simple1");
@@ -1376,7 +1392,7 @@ public class RedissonScoredSortedSetTest extends BaseTest {
         set2.add(2, "two");
         set2.add(3, "three");
 
-        RScoredSortedSet<String> out = redisson.getScoredSortedSet("out");
+        RScoredSortedSet<String> out = redisson.getScoredSortedSet("simple1");
         assertThat(out.readUnion(set1.getName(), set2.getName())).containsOnly("one", "two", "three", "four");
     }