Add watch subscription checking for event flush when not subscribing

Author: krapie

client/client.go

@@ -28,6 +28,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync/atomic"
 
 	"connectrpc.com/connect"
 	"github.com/rs/xid"
@@ -72,13 +73,18 @@ var (
 
 	// ErrInitializationNotReceived occurs when the first response of the watch stream is not received.
 	ErrInitializationNotReceived = errors.New("initialization is not received")
+
+	// ErrAlreadySubscribed occurs when the client is already subscribed to the document.
+	ErrAlreadySubscribed = errors.New("already subscribed")
 )
 
 // Attachment represents the document attached.
 type Attachment struct {
 	doc   *document.Document
 	docID types.ID
 
+	isSubscribed atomic.Bool
+
 	rch        <-chan WatchResponse
 	ctx        context.Context
 	cancelFunc context.CancelFunc
@@ -395,10 +401,8 @@ func (c *Client) Detach(ctx context.Context, doc *document.Document, options ...
 	if doc.Status() != document.StatusRemoved {
 		doc.SetStatus(document.StatusDetached)
 	}
-	delete(c.attachments, doc.Key())
-
 	attachment.cancelFunc()
-	attachment.rch = nil
+	delete(c.attachments, doc.Key())
 
 	return nil
 }
@@ -432,6 +436,11 @@ func (c *Client) Subscribe(
 		return nil, nil, ErrDocumentNotAttached
 	}
 
+	if attachment.isSubscribed.Load() {
+		return nil, nil, ErrAlreadySubscribed
+	}
+	attachment.isSubscribed.Store(true)
+
 	return attachment.rch, attachment.cancelFunc, nil
 }
 
@@ -485,13 +494,14 @@ func (c *Client) runWatchLoop(
 				close(rch)
 				return
 			}
-			if resp == nil {
+			if resp == nil || !attachment.isSubscribed.Load() {
 				continue
 			}
 
 			rch <- *resp
 		}
 		if err = stream.Err(); err != nil {
+			attachment.isSubscribed.Store(false)
 			rch <- WatchResponse{Err: err}
 			close(rch)
 			return

test/integration/document_test.go

@@ -166,7 +166,6 @@ func TestDocument(t *testing.T) {
 		rch, _, err := c1.Subscribe(d1)
 		assert.NoError(t, err)
 
-		isPresenceChangedEvent := true
 		go func() {
 			defer wg.Done()
 
@@ -178,16 +177,7 @@ func TestDocument(t *testing.T) {
 				}
 				assert.NoError(t, resp.Err)
 
-				// TODO(krapie): event includes presence changes in the document too.
-				// therefore, there will be
-				// 1. DocumentChanged from client 2 (Presence change)
-				// 2. DocumentWatched from client 2 (Watch event)
-				// 3. DocumentChanged from client 2 (Document change)
 				if resp.Type == client.DocumentChanged {
-					if isPresenceChangedEvent {
-						isPresenceChangedEvent = false
-						continue
-					}
 					err := c1.Sync(ctx, client.WithDocKey(d1.Key()))
 					assert.NoError(t, err)
 					return