Added Label for Watch commands #5
Conversation
There was a problem hiding this comment.
Thanks for the working on this feature @psrvere! I have left some comments.
watch_connection.go
Outdated
| label := msg.Command | ||
| _, err = uuid.Parse(label) | ||
|
|
||
| if err == nil { | ||
| select { | ||
| case c.firstUpdateCh <- msg: | ||
| if !timer.Stop() { | ||
| <-timer.C | ||
| } |
There was a problem hiding this comment.
Trying to understand the flow here. Shouldn't we have multiple firstUpdateCh's - one for each Watch invocation associated with a unique uuid? The logic here could look up the uuid in a map and send it to the correct firstUpdateCh (for the respective watch command).
There was a problem hiding this comment.
Agreed. I have fixed this by creating a new channel whenever Watch is invoked and the same is stored in watchLabelFirstMsgChMap
watch_connection.go
Outdated
| w.chOnce.Do(func() { | ||
| w.msgCh = newWatchCommandChannel(w) | ||
| // Create firstUpdateCh with buffer of 1 to avoid blocking | ||
| w.msgCh.firstUpdateCh = make(chan *WatchResult, 1) | ||
| w.msgCh.initMsgChan() | ||
| }) | ||
|
|
||
| // Subscribe to the command | ||
| err := w.watchCommand(ctx, cmdName, args...) |
There was a problem hiding this comment.
The SDK here should generate a uuid and associate it with the given watch command (maybe create a map where they key is the uuid and the value is this invocation's respective firstMsgCh.
The uuid is sent to the server by the SDK and the server simply returns it back with the first response (note that the server should also be able to handle the case where no uuid is sent and in that case it simply returns an empty value)
There was a problem hiding this comment.
Changed this, Now uuid is generated in SDK and sent to server.
|
@JyotinderSingh - worked on the comments. Please review. Here is the flow for your understanding:
|
There was a problem hiding this comment.
Thanks for addressing the reviews @psrvere! The changes look good, have just added a clarification comment.
This reverts commit 4ecce8f
What is the Problem?
SDK provides
Watchapi to subscribe Watch commands. This returns push response["GET", fingerprint, value]so that users can get fingerprint immediately.Since multiplexing is allowed i.e. users can subscribe for multiple keys, example:
GET.WATCH k1andGET.WATCH k2, there is no way for SDK to know that if a received update is the first message for sayk2or an update message fork1.What is the Solution?
PR adds watch label (uuid) in diceDB
worker.go. It is generated with every watch subscription. Label is received only in the first response by SDK.This PR adds two separate channels
firstUpdateChandupdateChfor first response and subsequent updates of Watch subscriptions.Watchapi call is followed byChannelapi call to get channel for updates. Earlier, Channel initialisedinitMsgChanroutine which would check for updates from server.Now
Watchapi call initialisesinitMsgChanandinitMsgChanfilters subscription first response and sends it tofirstUpdateChbuffered channel which is then consumed byWatch. In First response,msg.Commandis expected to be a valid uuid string. Label is not sent to the end user.All subsequent responses are sent to
updateChwhich is returned as response toChannelapi and users to consume updates on subscribed keys.