Implement streams API (#200)

* Add all of the commands except xInfo
Signed-off-by: NAleksandar Novakovic <anovakovic01@gmail.com>

* Fix XRead and XReadGroup inputs
Signed-off-by: NAleksandar Novakovic <anovakovic01@gmail.com>

* Fix some of the commands outputs
Signed-off-by: NAleksandar Novakovic <anovakovic01@gmail.com>

* Fix some of the commands inputs
Signed-off-by: NAleksandar Novakovic <anovakovic01@gmail.com>

* Fix import order
Signed-off-by: NAleksandar Novakovic <anovakovic01@gmail.com>

* Remove unnecessary case classes
Signed-off-by: NAleksandar Novakovic <anovakovic01@gmail.com>

* Fix Scan command output

Change Scan command output from Option[Unit] to Boolean.
Signed-off-by: NAleksandar Novakovic <anovakovic01@gmail.com>

* Update pattern matching of chunk to seq
Signed-off-by: NAleksandar Novakovic <anovakovic01@gmail.com>

redis/src/main/scala/zio/redis/Input.scala

@@ -211,6 +211,99 @@ object Input {
       )
   }
 
+  case object IdleInput extends Input[Duration] {
+    def encode(data: Duration): Chunk[RespValue.BulkString] =
+      Chunk(stringEncode("IDLE"), stringEncode(data.toMillis.toString))
+  }
+
+  case object TimeInput extends Input[Duration] {
+    def encode(data: Duration): Chunk[RespValue.BulkString] =
+      Chunk(stringEncode("TIME"), stringEncode(data.toMillis.toString))
+  }
+
+  case object RetryCountInput extends Input[Long] {
+    def encode(data: Long): Chunk[RespValue.BulkString] =
+      Chunk(stringEncode("RETRYCOUNT"), stringEncode(data.toString))
+  }
+
+  case object XGroupCreateInput extends Input[XGroupCommand.Create] {
+    def encode(data: XGroupCommand.Create): Chunk[RespValue.BulkString] = {
+      val chunk =
+        Chunk(
+          stringEncode("CREATE"),
+          stringEncode(data.key),
+          stringEncode(data.group),
+          stringEncode(data.id)
+        )
+
+      if (data.mkStream)
+        chunk :+ stringEncode(MkStream.stringify)
+      else
+        chunk
+    }
+  }
+
+  case object XGroupSetIdInput extends Input[XGroupCommand.SetId] {
+    def encode(data: XGroupCommand.SetId): Chunk[RespValue.BulkString] =
+      Chunk(stringEncode("SETID"), stringEncode(data.key), stringEncode(data.group), stringEncode(data.id))
+  }
+
+  case object XGroupDestroyInput extends Input[XGroupCommand.Destroy] {
+    def encode(data: XGroupCommand.Destroy): Chunk[RespValue.BulkString] =
+      Chunk(stringEncode("DESTROY"), stringEncode(data.key), stringEncode(data.group))
+  }
+
+  case object XGroupCreateConsumerInput extends Input[XGroupCommand.CreateConsumer] {
+    def encode(data: XGroupCommand.CreateConsumer): Chunk[RespValue.BulkString] =
+      Chunk(
+        stringEncode("CREATECONSUMER"),
+        stringEncode(data.key),
+        stringEncode(data.group),
+        stringEncode(data.consumer)
+      )
+  }
+
+  case object XGroupDelConsumerInput extends Input[XGroupCommand.DelConsumer] {
+    def encode(data: XGroupCommand.DelConsumer): Chunk[RespValue.BulkString] =
+      Chunk(stringEncode("DELCONSUMER"), stringEncode(data.key), stringEncode(data.group), stringEncode(data.consumer))
+  }
+
+  case object BlockInput extends Input[Duration] {
+    def encode(data: Duration): Chunk[RespValue.BulkString] =
+      Chunk(stringEncode("BLOCK"), stringEncode(data.toMillis.toString))
+  }
+
+  case object StreamsInput extends Input[((String, String), Chunk[(String, String)])] {
+    def encode(data: ((String, String), Chunk[(String, String)])): Chunk[RespValue.BulkString] = {
+      val (keys, ids) =
+        Chunk.fromIterable(data._1 +: data._2).map(pair => (stringEncode(pair._1), stringEncode(pair._2))).unzip
+
+      Chunk.single(stringEncode("STREAMS")) ++ keys ++ ids
+    }
+  }
+
+  case object GroupInput extends Input[Group] {
+    def encode(data: Group): Chunk[RespValue.BulkString] =
+      Chunk(stringEncode("GROUP"), stringEncode(data.group), stringEncode(data.consumer))
+  }
+
+  case object NoAckInput extends Input[NoAck] {
+    def encode(data: NoAck): Chunk[RespValue.BulkString] =
+      Chunk.single(stringEncode(data.stringify))
+  }
+
+  case object MaxLenInput extends Input[MaxLen] {
+    def encode(data: MaxLen): Chunk[RespValue.BulkString] = {
+      val chunk =
+        if (data.approximate)
+          Chunk(stringEncode("MAXLEN"), stringEncode("~"))
+        else
+          Chunk.single(stringEncode("MAXLEN"))
+
+      chunk :+ stringEncode(data.count.toString)
+    }
+  }
+
   final case class Tuple2[-A, -B](_1: Input[A], _2: Input[B]) extends Input[(A, B)] {
     def encode(data: (A, B)): Chunk[RespValue.BulkString] = _1.encode(data._1) ++ _2.encode(data._2)
   }
@@ -232,6 +325,19 @@ object Input {
       _1.encode(data._1) ++ _2.encode(data._2) ++ _3.encode(data._3) ++ _4.encode(data._4) ++ _5.encode(data._5)
   }
 
+  final case class Tuple6[-A, -B, -C, -D, -E, -F](
+    _1: Input[A],
+    _2: Input[B],
+    _3: Input[C],
+    _4: Input[D],
+    _5: Input[E],
+    _6: Input[F]
+  ) extends Input[(A, B, C, D, E, F)] {
+    def encode(data: (A, B, C, D, E, F)): Chunk[RespValue.BulkString] =
+      _1.encode(data._1) ++ _2.encode(data._2) ++ _3.encode(data._3) ++ _4.encode(data._4) ++ _5.encode(data._5) ++
+        _6.encode(data._6)
+  }
+
   final case class Tuple7[-A, -B, -C, -D, -E, -F, -G](
     _1: Input[A],
     _2: Input[B],
@@ -262,6 +368,24 @@ object Input {
         _6.encode(data._6) ++ _7.encode(data._7) ++ _8.encode(data._8) ++ _9.encode(data._9)
   }
 
+  final case class Tuple10[-A, -B, -C, -D, -E, -F, -G, -H, -I, -J](
+    _1: Input[A],
+    _2: Input[B],
+    _3: Input[C],
+    _4: Input[D],
+    _5: Input[E],
+    _6: Input[F],
+    _7: Input[G],
+    _8: Input[H],
+    _9: Input[I],
+    _10: Input[J]
+  ) extends Input[(A, B, C, D, E, F, G, H, I, J)] {
+    def encode(data: (A, B, C, D, E, F, G, H, I, J)): Chunk[RespValue.BulkString] =
+      _1.encode(data._1) ++ _2.encode(data._2) ++ _3.encode(data._3) ++ _4.encode(data._4) ++
+        _5.encode(data._5) ++ _6.encode(data._6) ++ _7.encode(data._7) ++ _8.encode(data._8) ++
+        _9.encode(data._9) ++ _10.encode(data._10)
+  }
+
   final case class Tuple11[-A, -B, -C, -D, -E, -F, -G, -H, -I, -J, -K](
     _1: Input[A],
     _2: Input[B],
@@ -305,4 +429,12 @@ object Input {
   case object WithHashInput extends Input[WithHash] {
     def encode(data: WithHash): Chunk[RespValue.BulkString] = Chunk.single(stringEncode(data.stringify))
   }
+
+  case object WithForceInput extends Input[WithForce] {
+    def encode(data: WithForce): Chunk[RespValue.BulkString] = Chunk.single(stringEncode(data.stringify))
+  }
+
+  case object WithJustIdInput extends Input[WithJustId] {
+    def encode(data: WithJustId): Chunk[RespValue.BulkString] = Chunk.single(stringEncode(data.stringify))
+  }
 }

redis/src/main/scala/zio/redis/Output.scala

@@ -10,9 +10,11 @@ sealed trait Output[+A] {
   private[redis] final def unsafeDecode(respValue: RespValue): A =
     respValue match {
       case RespValue.Error(msg) if msg.startsWith("ERR")       =>
-        throw RedisError.ProtocolError(msg.drop(3).trim())
+        throw RedisError.ProtocolError(msg.drop(3).trim)
       case RespValue.Error(msg) if msg.startsWith("WRONGTYPE") =>
-        throw RedisError.WrongType(msg.drop(9).trim())
+        throw RedisError.WrongType(msg.drop(9).trim)
+      case RespValue.Error(msg) if msg.startsWith("BUSYGROUP") =>
+        throw RedisError.WrongType(msg.drop(9).trim)
       case RespValue.Error(msg)                                =>
         throw RedisError.ProtocolError(msg.trim)
       case success                                             =>
@@ -310,4 +312,119 @@ object Output {
       }
   }
 
+  case object StreamOutput extends Output[Map[String, Map[String, String]]] {
+    protected def tryDecode(respValue: RespValue): Map[String, Map[String, String]] =
+      respValue match {
+        case RespValue.Array(entities) if entities.length % 2 == 0 =>
+          val output = collection.mutable.Map.empty[String, Map[String, String]]
+          val len    = entities.length
+          var pos    = 0
+          while (pos < len) {
+            (entities(pos), entities(pos + 1)) match {
+              case (id @ RespValue.BulkString(_), RespValue.Array(elements)) if elements.length % 2 == 0 =>
+                val entity = collection.mutable.Map.empty[String, String]
+                val elen   = elements.length
+                var epos   = 0
+                while (epos < elen) {
+                  (elements(epos), elements(epos + 1)) match {
+                    case (key @ RespValue.BulkString(_), value @ RespValue.BulkString(_)) =>
+                      entity += key.asString -> value.asString
+                    case _                                                                =>
+                  }
+                  epos += 2
+                }
+                output += id.asString -> entity.toMap
+              case _ =>
+            }
+            pos += 2
+          }
+          output.toMap
+        case array @ RespValue.Array(_) =>
+          throw ProtocolError(s"$array doesn't have an even number of elements")
+        case other =>
+          throw ProtocolError(s"$other isn't an array")
+      }
+  }
+
+  case object XPendingOutput extends Output[PendingInfo] {
+    protected def tryDecode(respValue: RespValue): PendingInfo =
+      respValue match {
+        case RespValue.Array(
+              Seq(
+                RespValue.Integer(total),
+                first @ RespValue.BulkString(_),
+                last @ RespValue.BulkString(_),
+                RespValue.Array(pairs)
+              )
+            ) =>
+          val consumers = collection.mutable.Map.empty[String, Long]
+          pairs.foreach {
+            case RespValue.Array(Seq(consumer @ RespValue.BulkString(_), RespValue.Integer(total))) =>
+              consumers += (consumer.asString -> total)
+            case _                                                                                  =>
+              throw ProtocolError(s"Consumer doesn't have 2 elements")
+          }
+
+          PendingInfo(total, first.asString, last.asString, consumers.toMap)
+        case array @ RespValue.Array(_) =>
+          throw ProtocolError(s"$array doesn't have valid format")
+        case other                      =>
+          throw ProtocolError(s"$other isn't an array")
+      }
+  }
+
+  case object PendingMessagesOutput extends Output[Chunk[PendingMessage]] {
+    protected def tryDecode(respValue: RespValue): Chunk[PendingMessage] =
+      respValue match {
+        case RespValue.Array(messages) =>
+          messages.collect {
+            case RespValue.Array(
+                  Seq(
+                    id @ RespValue.BulkString(_),
+                    owner @ RespValue.BulkString(_),
+                    RespValue.Integer(lastDelivered),
+                    RespValue.Integer(counter)
+                  )
+                ) =>
+              PendingMessage(id.asString, owner.asString, lastDelivered.millis, counter)
+          }
+        case other                     =>
+          throw ProtocolError(s"$other isn't an array")
+      }
+  }
+
+  case object XReadOutput extends Output[Map[String, Map[String, Map[String, String]]]] {
+    protected def tryDecode(respValue: RespValue): Map[String, Map[String, Map[String, String]]] =
+      respValue match {
+        case RespValue.Array(streams) if streams.length % 2 == 0 =>
+          val output = collection.mutable.Map.empty[String, Map[String, Map[String, String]]]
+          val len    = streams.length
+          var pos    = 0
+          while (pos < len) {
+            streams(pos) match {
+              case stream @ RespValue.BulkString(_) =>
+                output += (stream.asString -> StreamOutput.unsafeDecode(streams(pos + 1)))
+              case _                                =>
+            }
+
+            pos += 2
+          }
+
+          output.toMap
+        case array @ RespValue.Array(_) =>
+          throw ProtocolError(s"$array doesn't have valid format")
+        case other =>
+          throw ProtocolError(s"$other isn't an array")
+      }
+  }
+
+  case object SetOutput extends Output[Boolean] {
+    protected def tryDecode(respValue: RespValue): Boolean =
+      respValue match {
+        case RespValue.NullValue       => false
+        case RespValue.SimpleString(_) => true
+        case other                     => throw ProtocolError(s"$other isn't a valid set response")
+      }
+  }
+
 }

redis/src/main/scala/zio/redis/RedisError.scala

@@ -9,6 +9,7 @@ sealed trait RedisError extends NoStackTrace
 object RedisError {
   final case class ProtocolError(message: String)  extends RedisError
   final case class WrongType(message: String)      extends RedisError
+  final case class BusyGroup(message: String)      extends RedisError
+  final case class NoGroup(message: String)        extends RedisError
   final case class IOError(exception: IOException) extends RedisError
-
 }

redis/src/main/scala/zio/redis/api/Streams.scala

+package zio.redis.api
+
+import java.time.Duration
+
+import zio.redis.Input._
+import zio.redis.Output._
+import zio.redis._
+import zio.{ Chunk, ZIO }
+
+trait Streams {
+  import Streams._
+  import XGroupCommand._
+
+  final def xAck(key: String, group: String, id: String, ids: String*): ZIO[RedisExecutor, RedisError, Long] =
+    XAck.run((key, group, (id, ids.toList)))
+
+  final def xAdd(
+    key: String,
+    id: String,
+    pair: (String, String),
+    pairs: (String, String)*
+  ): ZIO[RedisExecutor, RedisError, String] =
+    XAdd.run((key, None, id, (pair, pairs.toList)))
+
+  final def xAddWithMaxLen(
+    key: String,
+    id: String,
+    count: Long,
+    approximate: Boolean,
+    pair: (String, String),
+    pairs: (String, String)*
+  ): ZIO[RedisExecutor, RedisError, String] =
+    XAdd.run((key, Some(MaxLen(approximate, count)), id, (pair, pairs.toList)))
+
+  final def xClaim(key: String, group: String, consumer: String, minIdleTime: Duration, id: String, ids: String*)(
+    idle: Option[Duration] = None,
+    time: Option[Duration] = None,
+    retryCount: Option[Long] = None,
+    force: Boolean = false
+  ): ZIO[RedisExecutor, RedisError, Map[String, Map[String, String]]] =
+    XClaim.run(
+      (
+        key,
+        group,
+        consumer,
+        minIdleTime,
+        (id, ids.toList),
+        idle,
+        time,
+        retryCount,
+        if (force) Some(WithForce) else None
+      )
+    )
+
+  final def xClaimWithJustId(
+    key: String,
+    group: String,
+    consumer: String,
+    minIdleTime: Duration,
+    id: String,
+    ids: String*
+  )(
+    idle: Option[Duration] = None,
+    time: Option[Duration] = None,
+    retryCount: Option[Long] = None,
+    force: Boolean = false
+  ): ZIO[RedisExecutor, RedisError, Chunk[String]] =
+    XClaimWithJustId.run(
+      (
+        key,
+        group,
+        consumer,
+        minIdleTime,
+        (id, ids.toList),
+        idle,
+        time,
+        retryCount,
+        if (force) Some(WithForce) else None,
+        WithJustId
+      )
+    )
+
+  final def xDel(key: String, id: String, ids: String*): ZIO[RedisExecutor, RedisError, Long] =
+    XDel.run((key, (id, ids.toList)))
+
+  final def xGroupCreate(
+    key: String,
+    group: String,
+    id: String,
+    mkStream: Boolean = false
+  ): ZIO[RedisExecutor, RedisError, Unit] =
+    XGroupCreate.run(Create(key, group, id, mkStream))
+
+  final def xGroupSetId(key: String, group: String, id: String): ZIO[RedisExecutor, RedisError, Unit] =
+    XGroupSetId.run(SetId(key, group, id))
+
+  final def xGroupDestroy(key: String, group: String): ZIO[RedisExecutor, RedisError, Long] =
+    XGroupDestroy.run(Destroy(key, group))
+
+  final def xGroupCreateConsumer(key: String, group: String, consumer: String): ZIO[RedisExecutor, RedisError, Unit] =
+    XGroupCreateConsumer.run(CreateConsumer(key, group, consumer))
+
+  final def xGroupDelConsumer(key: String, group: String, consumer: String): ZIO[RedisExecutor, RedisError, Long] =
+    XGroupDelConsumer.run(DelConsumer(key, group, consumer))
+
+  final def xLen(key: String): ZIO[RedisExecutor, RedisError, Long] =
+    XLen.run(key)
+
+  final def xPending(key: String, group: String): ZIO[RedisExecutor, RedisError, PendingInfo] =
+    XPending.run((key, group, None))
+
+  final def xPending(key: String, group: String, idle: Duration): ZIO[RedisExecutor, RedisError, PendingInfo] =
+    XPending.run((key, group, Some(idle)))
+
+  final def xPending(
+    key: String,
+    group: String,
+    start: String,
+    end: String,
+    count: Long,
+    consumer: Option[String] = None,
+    idle: Option[Duration] = None
+  ): ZIO[RedisExecutor, RedisError, Chunk[PendingMessage]] =
+    XPendingMessages.run((key, group, start, end, count, consumer, idle))
+
+  final def xRange(
+    key: String,
+    start: String,
+    end: String
+  ): ZIO[RedisExecutor, RedisError, Map[String, Map[String, String]]] =
+    XRange.run((key, start, end, None))
+
+  final def xRange(
+    key: String,
+    start: String,
+    end: String,
+    count: Long
+  ): ZIO[RedisExecutor, RedisError, Map[String, Map[String, String]]] =
+    XRange.run((key, start, end, Some(Count(count))))
+
+  final def xRead(count: Option[Long] = None, block: Option[Duration] = None)(
+    stream: (String, String),
+    streams: (String, String)*
+  ): ZIO[RedisExecutor, RedisError, Map[String, Map[String, Map[String, String]]]] =
+    XRead.run((count.map(Count), block, (stream, Chunk.fromIterable(streams))))
+
+  final def xReadGroup(group: String, consumer: String)(
+    count: Option[Long] = None,
+    block: Option[Duration] = None,
+    noAck: Boolean = false
+  )(
+    stream: (String, String),
+    streams: (String, String)*
+  ): ZIO[RedisExecutor, RedisError, Map[String, Map[String, Map[String, String]]]] =
+    XReadGroup.run(
+      (
+        Group(group, consumer),
+        count.map(Count),
+        block,
+        if (noAck) Some(NoAck) else None,
+        (stream, Chunk.fromIterable(streams))
+      )
+    )
+
+  final def xRevRange(
+    key: String,
+    end: String,
+    start: String
+  ): ZIO[RedisExecutor, RedisError, Map[String, Map[String, String]]] =
+    XRevRange.run((key, end, start, None))
+
+  final def xRevRange(
+    key: String,
+    end: String,
+    start: String,
+    count: Long
+  ): ZIO[RedisExecutor, RedisError, Map[String, Map[String, String]]] =
+    XRevRange.run((key, end, start, Some(Count(count))))
+
+  final def xTrim(key: String, count: Long, approximate: Boolean = false): ZIO[RedisExecutor, RedisError, Long] =
+    XTrim.run((key, MaxLen(approximate, count)))
+}
+
+private object Streams {
+
+  final val XAck = RedisCommand("XACK", Tuple3(StringInput, StringInput, NonEmptyList(StringInput)), LongOutput)
+
+  final val XAdd = RedisCommand(
+    "XADD",
+    Tuple4(StringInput, OptionalInput(MaxLenInput), StringInput, NonEmptyList(Tuple2(StringInput, StringInput))),
+    MultiStringOutput
+  )
+
+  final val XClaim = RedisCommand(
+    "XCLAIM",
+    Tuple9(
+      StringInput,
+      StringInput,
+      StringInput,
+      DurationMillisecondsInput,
+      NonEmptyList(StringInput),
+      OptionalInput(IdleInput),
+      OptionalInput(TimeInput),
+      OptionalInput(RetryCountInput),
+      OptionalInput(WithForceInput)
+    ),
+    StreamOutput
+  )
+
+  final val XClaimWithJustId = RedisCommand(
+    "XCLAIM",
+    Tuple10(
+      StringInput,
+      StringInput,
+      StringInput,
+      DurationMillisecondsInput,
+      NonEmptyList(StringInput),
+      OptionalInput(IdleInput),
+      OptionalInput(TimeInput),
+      OptionalInput(RetryCountInput),
+      OptionalInput(WithForceInput),
+      WithJustIdInput
+    ),
+    ChunkOutput
+  )
+
+  final val XDel = RedisCommand("XDEL", Tuple2(StringInput, NonEmptyList(StringInput)), LongOutput)
+
+  final val XGroupCreate = RedisCommand("XGROUP", XGroupCreateInput, UnitOutput)
+
+  final val XGroupSetId = RedisCommand("XGROUP", XGroupSetIdInput, UnitOutput)
+
+  final val XGroupDestroy = RedisCommand("XGROUP", XGroupDestroyInput, LongOutput)
+
+  final val XGroupCreateConsumer = RedisCommand("XGROUP", XGroupCreateConsumerInput, UnitOutput)
+
+  final val XGroupDelConsumer = RedisCommand("XGROUP", XGroupDelConsumerInput, LongOutput)
+
+  // TODO: implement XINFO command
+
+  final val XLen = RedisCommand("XLEN", StringInput, LongOutput)
+
+  final val XPending =
+    RedisCommand("XPENDING", Tuple3(StringInput, StringInput, OptionalInput(IdleInput)), XPendingOutput)
+
+  final val XPendingMessages =
+    RedisCommand(
+      "XPENDING",
+      Tuple7(
+        StringInput,
+        StringInput,
+        StringInput,
+        StringInput,
+        LongInput,
+        OptionalInput(StringInput),
+        OptionalInput(IdleInput)
+      ),
+      PendingMessagesOutput
+    )
+
+  final val XRange =
+    RedisCommand("XRANGE", Tuple4(StringInput, StringInput, StringInput, OptionalInput(CountInput)), StreamOutput)
+
+  final val XRead =
+    RedisCommand("XREAD", Tuple3(OptionalInput(CountInput), OptionalInput(BlockInput), StreamsInput), XReadOutput)
+
+  final val XReadGroup = RedisCommand(
+    "XREADGROUP",
+    Tuple5(GroupInput, OptionalInput(CountInput), OptionalInput(BlockInput), OptionalInput(NoAckInput), StreamsInput),
+    XReadOutput
+  )
+
+  final val XRevRange =
+    RedisCommand("XREVRANGE", Tuple4(StringInput, StringInput, StringInput, OptionalInput(CountInput)), StreamOutput)
+
+  final val XTrim = RedisCommand("XTRIM", Tuple2(StringInput, MaxLenInput), LongOutput)
+}

redis/src/main/scala/zio/redis/api/Strings.scala

@@ -211,7 +211,7 @@ trait Strings {
     expireTime: Option[Duration] = None,
     update: Option[Update] = None,
     keepTtl: Option[KeepTtl] = None
-  ): ZIO[RedisExecutor, RedisError, Option[Unit]] = Set.run((key, value, expireTime, update, keepTtl))
+  ): ZIO[RedisExecutor, RedisError, Boolean] = Set.run((key, value, expireTime, update, keepTtl))
 
   /**
    * Sets or clears the bit at offset in the string value stored at key
@@ -304,7 +304,7 @@ private[redis] object Strings {
         OptionalInput(UpdateInput),
         OptionalInput(KeepTtlInput)
       ),
-      OptionalOutput(UnitOutput)
+      SetOutput
     )
 
   final val SetBit   = RedisCommand("SETBIT", Tuple3(StringInput, LongInput, BoolInput), BoolOutput)

redis/src/main/scala/zio/redis/options/Geo.scala

@@ -2,8 +2,6 @@ package zio.redis.options
 
 trait Geo {
 
-  sealed case class Count(count: Long)
-
   sealed case class LongLat(longitude: Double, latitude: Double)
 
   sealed case class GeoView(member: String, dist: Option[Double], hash: Option[Long], longLat: Option[LongLat])

redis/src/main/scala/zio/redis/options/Shared.scala

@@ -13,4 +13,6 @@ trait Shared {
     case object SetExisting extends Update
     case object SetNew      extends Update
   }
+
+  sealed case class Count(count: Long)
 }

redis/src/main/scala/zio/redis/options/Streams.scala

+package zio.redis.options
+
+import zio.duration._
+
+trait Streams {
+  case object WithForce {
+    private[redis] def stringify = "FORCE"
+  }
+
+  type WithForce = WithForce.type
+
+  case object WithJustId {
+    private[redis] def stringify = "JUSTID"
+  }
+
+  type WithJustId = WithJustId.type
+
+  sealed trait XGroupCommand
+
+  object XGroupCommand {
+    case class Create(key: String, group: String, id: String, mkStream: Boolean) extends XGroupCommand
+    case class SetId(key: String, group: String, id: String)                     extends XGroupCommand
+    case class Destroy(key: String, group: String)                               extends XGroupCommand
+    case class CreateConsumer(key: String, group: String, consumer: String)      extends XGroupCommand
+    case class DelConsumer(key: String, group: String, consumer: String)         extends XGroupCommand
+  }
+
+  case object MkStream {
+    private[redis] def stringify = "MKSTREAM"
+  }
+
+  type MkStream = MkStream.type
+
+  case class PendingInfo(
+    total: Long,
+    first: String,
+    last: String,
+    consumers: Map[String, Long]
+  )
+
+  case class PendingMessage(
+    id: String,
+    owner: String,
+    lastDelivered: Duration,
+    counter: Long
+  )
+
+  case class Group(group: String, consumer: String)
+
+  case object NoAck {
+    private[redis] def stringify: String = "NOACK"
+  }
+
+  type NoAck = NoAck.type
+
+  case class MaxLen(approximate: Boolean, count: Long)
+}

redis/src/main/scala/zio/redis/package.scala

@@ -16,4 +16,5 @@ package object redis
     with options.SortedSets
     with options.Strings
     with options.Lists
+    with options.Streams
     with Interpreter

redis/src/test/scala/zio/redis/StringsSpec.scala

@@ -1162,7 +1162,7 @@ trait StringsSpec extends BaseSpec {
             key    <- uuid
             value  <- uuid
             result <- set(key, value)
-          } yield assert(result)(isSome)
+          } yield assert(result)(isTrue)
         },
         testM("override existing string") {
           for {
@@ -1170,7 +1170,7 @@ trait StringsSpec extends BaseSpec {
             value  <- uuid
             _      <- set(key, "value")
             result <- set(key, value)
-          } yield assert(result)(isSome)
+          } yield assert(result)(isTrue)
         },
         testM("override not string") {
           for {
@@ -1178,21 +1178,21 @@ trait StringsSpec extends BaseSpec {
             value  <- uuid
             _      <- sAdd(key, "a")
             result <- set(key, value)
-          } yield assert(result)(isSome)
+          } yield assert(result)(isTrue)
         },
         testM("new value with ttl 1 second") {
           for {
             key    <- uuid
             value  <- uuid
             result <- set(key, value, Some(1.second))
-          } yield assert(result)(isSome)
+          } yield assert(result)(isTrue)
         },
         testM("new value with ttl 100 milliseconds") {
           for {
             key    <- uuid
             value  <- uuid
             result <- set(key, value, Some(100.milliseconds))
-          } yield assert(result)(isSome)
+          } yield assert(result)(isTrue)
         },
         testM("error when negative ttl") {
           for {
@@ -1206,7 +1206,7 @@ trait StringsSpec extends BaseSpec {
             key    <- uuid
             value  <- uuid
             result <- set(key, value, update = Some(Update.SetNew))
-          } yield assert(result)(isSome)
+          } yield assert(result)(isTrue)
         },
         testM("existing value with SetNew parameter") {
           for {
@@ -1214,14 +1214,14 @@ trait StringsSpec extends BaseSpec {
             value  <- uuid
             _      <- set(key, "value")
             result <- set(key, value, update = Some(Update.SetNew))
-          } yield assert(result)(isNone)
+          } yield assert(result)(isFalse)
         },
         testM("new value with SetExisting parameter") {
           for {
             key    <- uuid
             value  <- uuid
             result <- set(key, value, update = Some(Update.SetExisting))
-          } yield assert(result)(isNone)
+          } yield assert(result)(isFalse)
         },
         testM("existing value with SetExisting parameter") {
           for {
@@ -1229,7 +1229,7 @@ trait StringsSpec extends BaseSpec {
             value  <- uuid
             _      <- set(key, "value")
             result <- set(key, value, update = Some(Update.SetExisting))
-          } yield assert(result)(isSome)
+          } yield assert(result)(isTrue)
         },
         testM("existing not string value with SetExisting parameter") {
           for {
@@ -1237,7 +1237,7 @@ trait StringsSpec extends BaseSpec {
             value  <- uuid
             _      <- sAdd(key, "a")
             result <- set(key, value, update = Some(Update.SetExisting))
-          } yield assert(result)(isSome)
+          } yield assert(result)(isTrue)
         },
         // next three tests include KEEPTTL parameter that is valid for Redis version >= 6
         testM("new value with KeepTtl parameter") {
@@ -1245,7 +1245,7 @@ trait StringsSpec extends BaseSpec {
             key    <- uuid
             value  <- uuid
             result <- set(key, value, keepTtl = Some(KeepTtl))
-          } yield assert(result)(isSome)
+          } yield assert(result)(isTrue)
         } @@ ignore,
         testM("existing value with KeepTtl parameter") {
           for {
@@ -1253,7 +1253,7 @@ trait StringsSpec extends BaseSpec {
             value  <- uuid
             _      <- set(key, "value", Some(1.second))
             result <- set(key, value, keepTtl = Some(KeepTtl))
-          } yield assert(result)(isSome)
+          } yield assert(result)(isTrue)
         } @@ ignore,
         testM("existing value with both ttl and KeepTtl parameters") {
           for {
@@ -1261,7 +1261,7 @@ trait StringsSpec extends BaseSpec {
             value  <- uuid
             _      <- set(key, "value", Some(1.second))
             result <- set(key, value, Some(1.second), keepTtl = Some(KeepTtl))
-          } yield assert(result)(isSome)
+          } yield assert(result)(isTrue)
         } @@ ignore
       ),
       suite("setBit")(