Implement PubSub api by 0pg · Pull Request #714 · zio/zio-redis

0pg · 2023-01-04T16:24:11Z

resolves #160

Added RedisSubscription and SubscriptionExecutor

Distinct between Publish and Subscribe commands
Execute Subscribe commands on SubscriptionExecutor
Subscribe commands with multiple parameters at once return Stream[RedisError, (String, A)] that tuple is (key, value) pair

CLAassistant · 2023-01-04T16:24:17Z

All committers have signed the CLA.

0pg · 2023-01-09T13:29:38Z

It might be better to get the subscription state for connection recovery 🤔

0pg · 2023-01-16T16:45:02Z

@mijicd I think I've done for the first step. please give me a feedback about this PR when you are ready 😃

anatolysergeev · 2023-01-22T11:08:26Z

-) extends RedisExecutor {
+  scope: Scope.Closeable,
+  codec: BinaryCodec
+) extends RedisExecutor


Don't think that cluster executor should inherit RedisPubSub.
I think it's more logic if it would be other way, there are would be some RedisPubSubClusterExecutor that is going to inherit ClusterExecuter, like SingleNodeRedisPubSubExecutor
And there is going to be some consistency.

mijicd · 2023-01-30T08:16:40Z

@0pg please rebase and address the change requested by @anatolysergeev 🙏

anovakovic01 · 2023-01-30T13:52:21Z

+        }
+    }
+
+  final def subscribe(channel: String, channels: String*): ZStream[Redis, RedisError, PushProtocol] =


In case of multiple channels passed, we should return a List/Chunk of streams. Also, I think that we should hide PushProtocol from the end user.

In case of multiple channels passed, we should return a List/Chunk of streams.

okay then it'd be better to get api with a single param and the callers have to handle multiple params using ZIO.foreach

I think that we should hide PushProtocol from the end user.

hmm... meaning that defines separate data types containing pushed message content for only end users or registers callbacks for each pubsub command for the users who want to handle subscribe/unsubscribe messages?

or using ResultBuilder that ask to give callbacks of sub/unsub when calls returning

def subscribe(channel: String): ResultOutputStreamBuilder = new ResultOutputStreamBuilder( override def returning[R: Schema]: ZStream[Redis, RedisError, R] = ZStream.serviceWithStream[Redis] { redis => RedisPubSubCommand .run(RedisPubSubCommand.Subscribe(channel)) .collect { case t: PushProtocol.Message => t.message } .mapZIO { resp => ZIO .attempt(ArbitraryOutput[R]().unsafeDecode(resp)(redis.codec)) .refineToOrDie[RedisError] } } ) def subscribe(channels: Chunk[String]): ResultOutputMultiStreamBuilder = new ResultOutputMultiStreamBuilder( override def returning[R: Schema]: UIO[Chunk[ZStream[Redis, RedisError, R]]] = ZIO.serviceWithZIO[Redis] { redis => RedisPubSubCommand .run(RedisPubSubCommand.Subscribe(channels.head, channels.tail)) .collect { case t: PushProtocol.Message => t.message } .broadcast(channels.length, maximumLag) .map(streams => streams.zipWith(channels) { (stream, channel) => stream .filter(_.key.value == channel) .mapZIO { resp => ZIO .attempt(ArbitraryOutput[R]().unsafeDecode(resp)(redis.codec)) .refineToOrDie[RedisError] } } ) }

This is just a sketch. I hope that I managed to explain what's the base idea through this code. Of course, something like ResultOutputMultiStreamBuilder should be added for multi-stream scenarios.

yeah that's almost same what I though and mentioned above 👍

Fix RedisExecutor structure Add PubSub api Fix broken compile Add PushProtocolOutput suite Fix message field type as generic Fix broken output Fix PushProtocol output spec Add key property in PushProtocol Add test implementation Add PubSub integration test Fix formatting Refactor RedisPubSub Apply RedisPubSub refactoring Apply RedisPubSub refactoring to t/c Remove unused file Fix logic bugs Fix broken t/c Fix unsubscribe process Fix pubSubSpec Add request message broker in SingleNodeRedisPubSub Simplify RedisPubSub's public api Revert unrelated changes

0pg · 2023-02-08T03:44:32Z

@anovakovic01 Hi there are some changes please check 🙏

added ResultStreamBuilder[+F[_]
added subscription callback

Cluster PubSubExecutor will be worked by another PR

anovakovic01 · 2023-02-09T08:21:36Z

+    case class Pattern(value: String) extends SubscriptionKey
+  }
+
+  case class NumSubResponse(channel: String, subscriberCount: Long)


Maybe it would be better to name this class NumberOfSubscribers, or NumOfSubs. Also, you should place it in the options/PubSub.scala file and make it final.

I renamed and moved it inside to companion object of options/PubSub

anovakovic01 · 2023-02-09T09:53:15Z

+
+  object PushProtocol {
+    case class Subscribe(channel: String, numOfSubscription: Long) extends PushProtocol {
+      def key: SubscriptionKey = SubscriptionKey.Channel(channel)


Do you need SubscriptionKey? Can you use String instead?

you mean that gets rid of SubscriptionKey? or changes return type of def key to String?
I changed return type to String

Sorry for not being clear. I meant to remove SubscriptionKey entirely and replace it with a plain String.

ah I moved SubscriptionKey into SingleNodeRedisPubSub as a private case class

anovakovic01 · 2023-02-09T10:38:28Z

+
+  final def subscribeWithCallback(
+    channel: String,
+    channels: List[String]


You can use String* here.

anovakovic01 · 2023-02-09T10:38:43Z

+  final def subscribe(channel: String): ResultStreamBuilder[Id] =
+    subscribeWithCallback(channel)(emptyCallback)
+
+  final def subscribe(channel: String, channels: List[String]): ResultStreamBuilder[List] =


You can use String* here.

anovakovic01 · 2023-02-09T10:46:15Z

+          ZIO
+            .attempt(ArbitraryOutput[R]().unsafeDecode(msg)(codec))
+            .refineToOrDie[RedisError]
+            .asSome


I think that this can be part of the run implementation in the RedisPubSubCommand.

I made RedisPubSubCommand takes responsibilities of transforming message types and invoking callback

anovakovic01 · 2023-02-09T10:47:44Z

+sealed trait PubSubCommand
+
+object PubSubCommand {
+  case class Subscribe(
+    channel: String,
+    channels: List[String],
+    onSubscribe: PubSubCallback
+  ) extends PubSubCommand
+  case class PSubscribe(
+    pattern: String,
+    patterns: List[String],
+    onSubscribe: PubSubCallback
+  ) extends PubSubCommand
+  case class Unsubscribe(channels: List[String])  extends PubSubCommand
+  case class PUnsubscribe(patterns: List[String]) extends PubSubCommand
+}


We should hide these from the end user. These should be in a dedicated PubSubCommand file.

I moved it to its own file but do we need to add package private access modifier?

I think that we do if it's not part of the api interface.

I added package private accessor

Co-authored-by: Aleksandar Novaković <anovakovic01@gmail.com>

0pg · 2023-05-07T14:19:22Z

@anovakovic01 I fixed some cross compile issues 🙏
92457c4

mijicd · 2023-05-07T14:27:14Z

@anovakovic01 this PR needs to address a few things before merging (compilation, linting, sync with master)

0pg · 2023-05-08T07:24:32Z

Is there anything to do about test failures? it seems like non deterministic 🤔

mijicd · 2023-05-08T08:26:12Z

It looks like these aren't directly related to your change, so I'd say let's not touch any of them in this PR.

mijicd

Thank you for taking care of this. It's a massive chunk of work, and it's going in the right direction.

During the first review pass, I focused on the package layout, top-level abstractions, and protocol correctness. Once all of them are in place, I'll do a more detailed review of its internals, specifically the streaming part.

One general remark is that we can improve the test suite to use property tests and cover more than just a "happy path". Note that this remark applies to most tests, but we must start improving their state somewhere.

mijicd · 2023-05-09T12:53:35Z

+    case class Subscribe(channel: String, numOfSubs: Long)                    extends PushProtocol
+    case class PSubscribe(pattern: String, numOfSubs: Long)                   extends PushProtocol
+    case class Unsubscribe(channel: String, numOfSubs: Long)                  extends PushProtocol
+    case class PUnsubscribe(pattern: String, numOfSubs: Long)                 extends PushProtocol
+    case class Message(channel: String, message: RespValue)                   extends PushProtocol
+    case class PMessage(pattern: String, channel: String, message: RespValue) extends PushProtocol


Please make the case classes final. Besides that, types look odd, and there's a lot of duplication.

I grouped message types using new SubscriptionKey type that contains context of channel or pattern
a154683

it also changes SingleNodeSubscriptionExecutor's member Ref that manages subscription state

mijicd · 2023-05-09T12:54:37Z

+object PubSub {
+  type PubSubCallback = (String, Long) => UIO[Unit]
+
+  private[redis] sealed trait PushProtocol


Let's rename this to PushMessage. Also, this doesn't really look like an option, it's very much internal thing.

mijicd · 2023-05-09T12:55:54Z

      }
  }

+  case object PushProtocolOutput extends Output[PushProtocol] {


Please rename accordingly (see the remark about PushProtocol below). Another question is how visible it should be.

PushMessageOutput is only used in internal package so I added private[redis] to this object
a154683

mijicd · 2023-05-09T13:09:40Z

+          commandName <-
+            ZIO
+              .fromOption(command.args.collectFirst { case RespCommandArgument.CommandName(name) => name })
+              .orElseFail(RedisError.CommandNameNotFound(command.args.toString()))


Let's adjust the error constructor to avoid invoking toString explicitly.

mijicd · 2023-05-09T13:25:11Z

+    new ResultStreamBuilder1[({ type lambda[x] = (String, x) })#lambda] {
+      def returning[R: Schema]: Stream[RedisError, (String, R)] =
+        RedisSubscriptionCommand(executor).subscribe(
+          Chunk.single(channel) ++ Chunk.fromIterable(channels),


No need for chunk concatenation.

mijicd · 2023-05-09T13:26:29Z

+    pattern: String,
+    patterns: String*


Type-safety could be improved.

Are you suggesting that defines the parameter types for channels and patterns?

mijicd · 2023-05-09T13:27:21Z

+}
+
+object Subscription {
+  private lazy val emptyCallback = (_: String, _: Long) => ZIO.unit


Let's follow the constants definition guideline. Also no need to make it lazy.

I removed this value and replaced to using Option

mijicd · 2023-05-09T13:28:42Z

+import zio.schema.Schema
+import zio.schema.codec.BinaryCodec
+
+private[redis] trait SubscribeEnvironment {


We might be able to generalize the environments, though I'm fine with exploring that possibility in a follow-up.

mijicd · 2023-05-09T13:28:58Z


 import zio._
-import zio.redis.Output._
+import zio.redis.Output.{PushProtocolOutput, _}


This is an unnecessary change.

mijicd

Thank you for taking care of this. It's a massive chunk of work, and it's going in the right direction.

During the first review pass, I focused on the package layout, top-level abstractions, and protocol correctness. Once all of them are in place, I'll do a more detailed review of its internals, specifically the streaming part.

One general remark is that we can improve the test suite to use property tests and cover more than just a "happy path". Note that this remark applies to most tests, but we must start improving their state somewhere.

0pg · 2023-06-02T13:13:46Z

@mijicd Thank you for your review. I worked some parts based on comments. please to check changes and reply comments when you're fine. 😃

mijicd

@anovakovic01 @anatolysergeev Please take a look!

mijicd · 2023-06-06T13:52:27Z

+
+package zio.redis.internal
+
+object PubSub {


Just mark the whole object as private[redis]. Everything in internal is marked that way.

I marked it to them

mijicd · 2023-06-06T13:55:27Z

+
+private[redis] object RequestQueue {
+  private final val RequestQueueSize = 16
+  def create[A]: UIO[Queue[A]]       = Queue.bounded[A](RequestQueueSize)


I would drop this and move constant to package object.

I moved constant to package object and removed this object

mijicd · 2023-06-06T13:56:23Z

+  val run: IO[RedisError, AnyVal] =
+    ZIO.logTrace(s"$this sender and reader has been started") *>
+      (send.repeat(Schedule.forever) race receive)
+        .tapError(e => ZIO.logWarning(s"Reconnecting due to error: $e") *> onError(e))
+        .retryWhile(True)
+        .tapError(e => ZIO.logError(s"Executor exiting: $e"))


It should likely be private and final.

I marked it as ~~protected final~~(it broke compile on scala3) private[internal] final

mijicd · 2023-06-06T13:58:01Z

+import zio.{Chunk, ChunkBuilder, Hub, IO, Promise, Queue, Ref, Scope, UIO, URIO, ZIO}
+
+private[redis] final class SingleNodeSubscriptionExecutor private (
+  subsRef: Ref[Map[SubscriptionKey, Hub[Take[RedisError, PushMessage]]]],


It might be worth using ConcurrentMap.

I replaced Ref[Map[...]] to ConcurrentMap

mijicd · 2023-06-06T14:01:45Z

+    def getHub(key: SubscriptionKey): IO[RedisError, Hub[Take[RedisError, PushMessage]]] =
+      subsRef.get
+        .map(_.get(key))
+        .flatMap(ZIO.fromOption(_))


I think you can go without (_). Do that on all applicable places.

I changed to use .apply because 3.2.2 version requires it

…criptionExecutor.scala Co-authored-by: Dejan Mijić <dmijic@acm.org>

anovakovic01 · 2023-06-25T09:09:42Z

+      }
+  }
+
+  case object NumSubResponseOutput extends Output[Chunk[NumberOfSubscribers]] {


I know that there is an order in the response, but maybe you should use a Map instead (for easier access to the value of the specific channel).

anovakovic01 · 2023-06-25T09:25:04Z

+                         .fromOption(command.args.collectFirst { case RespCommandArgument.CommandName(name) =>
+                           name
+                         })
+                         .orElseFail(RedisError.CommandNameNotFound(command))


I think that ProtocolError should be thrown instead, but I'm not sure how you can do it from here. @mijicd maybe we can update RespCommand to receive a NonEmptyChunk because it always has to have a name (or add a name parameter). We could do this in a dedicated PR.

Thrown ProtocolError more makes sense 🤔 I'll try to this using NonEmptyChunk approach in another PR

I made PR that gives name separated field
#852

anovakovic01 · 2023-06-25T09:45:36Z

+    onSubscribe: Option[PubSubCallback],
+    onUnsubscribe: Option[PubSubCallback]


What do you think about giving default values here and removing Option by introducing NoopCallback:

Suggested change

onSubscribe: Option[PubSubCallback],

onUnsubscribe: Option[PubSubCallback]

onSubscribe: PubSubCallback = NoopCallback,

onUnsubscribe: PubSubCallback = NoopCallback

The same suggestion applies to other methods with optional callback parameters.

yeah it seems to make more ergonomic api. I changed to use NoopCallback.
I changed function name subscribe to subscribeSingle for only single channel subscription call because it's more clear to distinct multiple channels subscription call (also they have different return type signature)
5ca98e7 (#714)

anovakovic01 · 2023-06-25T10:05:03Z

+    }.map(SubscriptionKey.Pattern.apply)
+
+  def send: IO[RedisError.IOError, Unit] =
+    requests.takeAll.flatMap { reqs =>


Use takeBetween(1, RequestQueueSize) just like in the existing send implementation.

anovakovic01 · 2023-06-25T10:08:15Z

+
+trait Subscription extends SubscribeEnvironment {
+
+  final def subscribe(channel: String): ResultStreamBuilder1[Id] =


Add doc comments for the public API.

I added doc Subscription and Publishing apis

anovakovic01 · 2023-06-25T10:10:42Z

@0pg Great work! You're almost there, these are some minor changes that I've commented on, but nothing major.

0pg requested a review from a team as a code owner January 4, 2023 16:24

0pg force-pushed the impl/pubsub branch from 485d388 to b1b9cc2 Compare January 4, 2023 16:34

anatolysergeev reviewed Jan 22, 2023

View reviewed changes

anovakovic01 requested changes Jan 30, 2023

View reviewed changes

0pg force-pushed the impl/pubsub branch from 08f123b to d4aa330 Compare January 31, 2023 09:14

0pg force-pushed the impl/pubsub branch from d4aa330 to a280504 Compare January 31, 2023 09:26

0pg added 5 commits February 7, 2023 16:33

Fix PubSub api

cd4a3f3

Apply changes to t/c

a01213d

Set private accessor

9541519

Merge remote-tracking branch 'upstream/master' into impl/pubsub

ba89a59

Fix cleanup code

9723182

0pg force-pushed the impl/pubsub branch from 4691778 to 9723182 Compare February 7, 2023 09:03

0pg added 6 commits February 7, 2023 22:07

Modify callback and unsubscribe logic in SingleNodeRedisPubSub

65b524b

Apply changes to t/c

892ba20

Modify PubSub api

43edb47

Apply changes to t/c

a6f74fa

Fix typo

c7decb7

Modify accessor

b6438f0

anovakovic01 reviewed Feb 9, 2023

View reviewed changes

0pg and others added 5 commits February 9, 2023 22:27

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

5fb75e4

Co-authored-by: Aleksandar Novaković <anovakovic01@gmail.com>

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

ce97f95

Co-authored-by: Aleksandar Novaković <anovakovic01@gmail.com>

Modify responsibility to transform streams

a25805b

Apply changes to t/c

fd0b303

Move SubscriptionKey into only used place

7cb1491

Formatting

bf52e26

Fix lint

c027295

mijicd requested changes May 9, 2023

View reviewed changes

0pg added 8 commits May 11, 2023 02:16

Reduce duplications and refactor package layout

a154683

Commit suggestion

592d327

Fix broken compile

386fefd

Merge remote-tracking branch 'upstream/master' into impl/pubsub

415f75e

Use hub instead of chunks of queue

1699211

Add release on error

be11d67

Extraact common logic

552e002

Ensure order of subs/unsubs

e218b08

mijicd reviewed Jun 6, 2023

View reviewed changes

0pg and others added 4 commits June 7, 2023 22:59

Update modules/redis/src/main/scala/zio/redis/internal/SingleNodeSubs…

ac85cb9

…criptionExecutor.scala Co-authored-by: Dejan Mijić <dmijic@acm.org>

Use ConcurrentMap instead of Map of Ref

dadb5e7

Add private accessor

88c3f4e

Fix accessor

6e50d4d

anovakovic01 reviewed Jun 25, 2023

View reviewed changes

0pg added 3 commits June 27, 2023 04:18

Add doc

5ca98e7

Add missing params

539ef3f

Fix test codes

63cc263

mijicd merged commit 407b9ac into zio:master Jul 27, 2023

0pg mentioned this pull request Jul 28, 2023

Explicit resp command name #852

Draft

hcwilhelm pushed a commit to hcwilhelm/zio-redis that referenced this pull request Feb 18, 2025

Implement PubSub api (zio#714)

018da09

		onSubscribe: Option[PubSubCallback],
		onUnsubscribe: Option[PubSubCallback]


		trait Subscription extends SubscribeEnvironment {

		final def subscribe(channel: String): ResultStreamBuilder1[Id] =

Conversation

0pg commented Jan 4, 2023 • edited by mijicd Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

CLAassistant commented Jan 4, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

0pg commented Jan 9, 2023

Uh oh!

0pg commented Jan 16, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mijicd commented Jan 30, 2023

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

0pg commented Feb 8, 2023

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

0pg commented May 7, 2023

Uh oh!

mijicd commented May 7, 2023

Uh oh!

0pg commented May 8, 2023

Uh oh!

mijicd commented May 8, 2023

Uh oh!

mijicd left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

0pg May 10, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

0pg commented Jan 4, 2023 •

edited by mijicd

Loading

CLAassistant commented Jan 4, 2023 •

edited

Loading

0pg May 10, 2023 •

edited

Loading