Added support for commands: zunion, zinter, zdiff, smismember (profun…

…ktor#652)

README.md

@@ -72,6 +72,17 @@ Start both a single Redis node and a cluster using `docker-compose`:
 > sbt +test
 ```
 
+If you are trying to run cluster mode tests on macOS you might receive host not found errors. As a workaround add
+new environment variable in `docker-compose.yml` for `RedisCluster`: `IP=0.0.0.0`
+
+The environment section should look like this:
+```
+    environment:
+      - INITIAL_PORT=30001
+      - DEBUG=false
+      - IP=0.0.0.0
+```
+
 ## Code of Conduct
 
 See the [Code of Conduct](https://redis4cats.profunktor.dev/CODE_OF_CONDUCT)

docker-compose.yml

@@ -1,21 +1,23 @@
-SingleNode:
-  restart: always
-  image: redis:6.0.8
-  ports:
-    - "6379:6379"
-  environment:
-    - DEBUG=false
+version: "3.0"
+services:
+  SingleNode:
+    restart: always
+    image: redis:6.2.0
+    ports:
+      - "6379:6379"
+    environment:
+      - DEBUG=false
 
-RedisCluster:
-  restart: always
-  image: grokzen/redis-cluster:6.0.8
-  ports:
-    - "30001:30001"
-    - "30002:30002"
-    - "30003:30003"
-    - "30004:30004"
-    - "30005:30005"
-    - "30006:30006"
-  environment:
-    - INITIAL_PORT=30001
-    - DEBUG=false
+  RedisCluster:
+    restart: always
+    image: grokzen/redis-cluster:6.2.0
+    ports:
+      - "30001:30001"
+      - "30002:30002"
+      - "30003:30003"
+      - "30004:30004"
+      - "30005:30005"
+      - "30006:30006"
+    environment:
+      - INITIAL_PORT=30001
+      - DEBUG=false

modules/effects/src/main/scala/dev/profunktor/redis4cats/algebra/sets.scala

@@ -18,6 +18,7 @@ package dev.profunktor.redis4cats.algebra
 
 trait SetCommands[F[_], K, V] extends SetGetter[F, K, V] with SetSetter[F, K, V] with SetDeletion[F, K, V] {
   def sIsMember(key: K, value: V): F[Boolean]
+  def sMisMember(key: K, values: V*): F[List[Boolean]]
 }
 
 trait SetGetter[F[_], K, V] {

modules/effects/src/main/scala/dev/profunktor/redis4cats/algebra/sortedsets.scala

@@ -18,7 +18,7 @@ package dev.profunktor.redis4cats.algebra
 
 import cats.data.NonEmptyList
 import dev.profunktor.redis4cats.effects.{ RangeLimit, ScoreWithValue, ZRange }
-import io.lettuce.core.{ ZAddArgs, ZStoreArgs }
+import io.lettuce.core.{ ZAddArgs, ZAggregateArgs, ZStoreArgs }
 
 import scala.concurrent.duration.Duration
 
@@ -53,6 +53,12 @@ trait SortedSetGetter[F[_], K, V] {
   def zPopMax(key: K, count: Long): F[List[ScoreWithValue[V]]]
   def bzPopMax(timeout: Duration, keys: NonEmptyList[K]): F[Option[(K, ScoreWithValue[V])]]
   def bzPopMin(timeout: Duration, keys: NonEmptyList[K]): F[Option[(K, ScoreWithValue[V])]]
+  def zUnion(args: Option[ZAggregateArgs], keys: K*): F[List[V]]
+  def zUnionWithScores(args: Option[ZAggregateArgs], keys: K*): F[List[ScoreWithValue[V]]]
+  def zInter(args: Option[ZAggregateArgs], keys: K*): F[List[V]]
+  def zInterWithScores(args: Option[ZAggregateArgs], keys: K*): F[List[ScoreWithValue[V]]]
+  def zDiff(keys: K*): F[List[V]]
+  def zDiffWithScores(keys: K*): F[List[ScoreWithValue[V]]]
 }
 
 trait SortedSetSetter[F[_], K, V] {

modules/effects/src/main/scala/dev/profunktor/redis4cats/redis.scala

@@ -38,6 +38,7 @@ import io.lettuce.core.{
   GeoWithin,
   ScoredValue,
   ZAddArgs,
+  ZAggregateArgs,
   ZStoreArgs,
   Limit => JLimit,
   Range => JRange,
@@ -809,6 +810,12 @@ private[redis4cats] class BaseRedis[F[_]: FutureLift: MonadThrow: RedisExecutor:
       .futureLift
       .map(x => Boolean.box(x))
 
+  override def sMisMember(key: K, values: V*): F[List[Boolean]] =
+    async
+      .flatMap(c => RedisExecutor[F].delay(c.smismember(key, values: _*)))
+      .futureLift
+      .map(_.asScala.map(Boolean.unbox(_)).toList)
+
   override def sAdd(key: K, values: V*): F[Long] =
     async.flatMap(c => RedisExecutor[F].delay(c.sadd(key, values: _*))).futureLift.map(x => Long.box(x))
 
@@ -1394,6 +1401,54 @@ private[redis4cats] class BaseRedis[F[_]: FutureLift: MonadThrow: RedisExecutor:
       .futureLift
       .map(Option(_).map(kv => (kv.getKey, kv.getValue.asScoreWithValues)))
 
+  override def zUnion(args: Option[ZAggregateArgs], keys: K*): F[List[V]] = {
+    val res = args match {
+      case Some(aggArgs) => async.flatMap(c => RedisExecutor[F].delay(c.zunion(aggArgs, keys: _*)))
+      case None          => async.flatMap(c => RedisExecutor[F].delay(c.zunion(keys: _*)))
+    }
+    res.futureLift
+      .map(_.asScala.toList)
+  }
+
+  override def zUnionWithScores(args: Option[ZAggregateArgs], keys: K*): F[List[ScoreWithValue[V]]] = {
+    val res = args match {
+      case Some(aggArgs) => async.flatMap(c => RedisExecutor[F].delay(c.zunionWithScores(aggArgs, keys: _*)))
+      case None          => async.flatMap(c => RedisExecutor[F].delay(c.zunionWithScores(keys: _*)))
+    }
+    res.futureLift
+      .map(_.asScala.toList.map(_.asScoreWithValues))
+  }
+
+  override def zInter(args: Option[ZAggregateArgs], keys: K*): F[List[V]] = {
+    val res = args match {
+      case Some(aggArgs) => async.flatMap(c => RedisExecutor[F].delay(c.zinter(aggArgs, keys: _*)))
+      case None          => async.flatMap(c => RedisExecutor[F].delay(c.zinter(keys: _*)))
+    }
+    res.futureLift
+      .map(_.asScala.toList)
+  }
+
+  override def zInterWithScores(args: Option[ZAggregateArgs], keys: K*): F[List[ScoreWithValue[V]]] = {
+    val res = args match {
+      case Some(aggArgs) => async.flatMap(c => RedisExecutor[F].delay(c.zinterWithScores(aggArgs, keys: _*)))
+      case None          => async.flatMap(c => RedisExecutor[F].delay(c.zinterWithScores(keys: _*)))
+    }
+    res.futureLift
+      .map(_.asScala.toList.map(_.asScoreWithValues))
+  }
+
+  override def zDiff(keys: K*): F[List[V]] =
+    async
+      .flatMap(c => RedisExecutor[F].delay(c.zdiff(keys: _*)))
+      .futureLift
+      .map(_.asScala.toList)
+
+  override def zDiffWithScores(keys: K*): F[List[ScoreWithValue[V]]] =
+    async
+      .flatMap(c => RedisExecutor[F].delay(c.zdiffWithScores(keys: _*)))
+      .futureLift
+      .map(_.asScala.toList.map(_.asScoreWithValues))
+
   /******************************* Connection API **********************************/
   override val ping: F[String] =
     async.flatMap(c => RedisExecutor[F].lift(c.ping())).futureLift

modules/tests/src/test/scala/dev/profunktor/redis4cats/TestScenarios.scala

@@ -29,7 +29,7 @@ import dev.profunktor.redis4cats.effects._
 import dev.profunktor.redis4cats.hlist._
 import dev.profunktor.redis4cats.pipeline.{ PipelineError, RedisPipeline }
 import dev.profunktor.redis4cats.transactions.RedisTransaction
-import io.lettuce.core.GeoArgs
+import io.lettuce.core.{ GeoArgs, ZAggregateArgs }
 import munit.FunSuite
 
 import scala.concurrent.duration._
@@ -151,13 +151,18 @@ trait TestScenarios { self: FunSuite =>
       _ <- IO(assert(z.isEmpty))
       t <- cmd.sCard(testKey)
       _ <- IO(assertEquals(t, 0L))
+      _ <- cmd.sAdd(testKey, "value 1", "value 2")
+      r <- cmd.sMisMember(testKey, "value 1", "random", "value 2")
+      _ <- IO(assertEquals(r, List(true, false, true)))
     } yield ()
   }
 
   def sortedSetsScenario(cmd: RedisCommands[IO, String, Long]): IO[Unit] = {
-    val testKey         = "zztop"
+    val testKey         = "{same_hash_slot}:zztop"
+    val otherTestKey    = "{same_hash_slot}:sharp:dressed:man"
     val scoreWithValue1 = ScoreWithValue(Score(1), 1L)
     val scoreWithValue2 = ScoreWithValue(Score(3), 2L)
+    val scoreWithValue3 = ScoreWithValue(Score(5), 3L)
     val timeout         = 1.second
     for {
       minPop1 <- cmd.zPopMin(testKey, 1)
@@ -190,6 +195,22 @@ trait TestScenarios { self: FunSuite =>
       _ <- IO(assert(y.contains(2)))
       z <- cmd.zCount(testKey, ZRange(0, 1))
       _ <- IO(assert(z.contains(1)))
+      _ <- cmd.zAdd(otherTestKey, args = None, scoreWithValue1, scoreWithValue3)
+      zUnion <- cmd.zUnion(args = None, testKey, otherTestKey)
+      _ <- IO(assertEquals(zUnion, List(1L, 2L, 3L)))
+      aggregateArgs = ZAggregateArgs.Builder.sum().weights(10L, 20L)
+      zUnionWithScoreAndArgs <- cmd.zUnionWithScores(Some(aggregateArgs), testKey, otherTestKey)
+      _ <- IO(
+            assertEquals(
+              zUnionWithScoreAndArgs,
+              // scores for each element: 1 -> 10*1 + 20*1; 2 -> 10*3; 3 -> 20*5
+              List(ScoreWithValue(Score(30), 1L), ScoreWithValue(Score(30), 2L), ScoreWithValue(Score(100), 3L))
+            )
+          )
+      zInter <- cmd.zInter(args = None, testKey, otherTestKey)
+      _ <- IO(assertEquals(zInter, List(1L)))
+      zDiff <- cmd.zDiff(testKey, otherTestKey)
+      _ <- IO(assertEquals(zDiff, List(2L)))
       r <- cmd.zRemRangeByScore(testKey, ZRange(1, 3))
       _ <- IO(assertEquals(r, 2L))
     } yield ()