Add test cases

test/integration/document_test.go

@@ -37,8 +37,8 @@ import (
 )
 
 func TestDocument(t *testing.T) {
-	clients := activeClients(t, 2)
-	c1, c2 := clients[0], clients[1]
+	clients := activeClients(t, 3)
+	c1, c2, c3 := clients[0], clients[1], clients[2]
 	defer deactivateAndCloseClients(t, clients)
 
 	t.Run("attach/detach test", func(t *testing.T) {
@@ -421,12 +421,14 @@ func TestDocument(t *testing.T) {
 		assert.ErrorIs(t, err, client.ErrDocumentNotAttached)
 	})
 
-	t.Run("broadcast to subscriber except publisher test", func(t *testing.T) {
+	t.Run("broadcast to subscribers except publisher test", func(t *testing.T) {
 		bch := make(chan string)
 		ctx := context.Background()
 		handler := func(eventType, publisher string, payload []byte) error {
 			var mentionedBy string
 			assert.NoError(t, gojson.Unmarshal(payload, &mentionedBy))
+			// Send the unmarshaled payload to the channel to notify that this
+			// subscriber receives the event.
 			bch <- mentionedBy
 			return nil
 		}
@@ -445,6 +447,150 @@ func TestDocument(t *testing.T) {
 		err = d2.SubscribeBroadcastEvent("mention", handler)
 		assert.NoError(t, err)
 
+		d3 := document.New(helper.TestDocKey(t))
+		assert.NoError(t, c3.Attach(ctx, d3))
+		rch3, err := c3.Watch(ctx, d3)
+		assert.NoError(t, err)
+		err = d3.SubscribeBroadcastEvent("mention", handler)
+		assert.NoError(t, err)
+
+		err = d3.Broadcast("mention", "yorkie")
+		assert.NoError(t, err)
+
+		wg := sync.WaitGroup{}
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			rcv := 0
+			for {
+				select {
+				case <-rch1:
+				case <-rch2:
+				case <-rch3:
+				case m := <-bch:
+					assert.Equal(t, "yorkie", m)
+					rcv++
+				case <-time.After(1 * time.Second):
+					// Assuming that all subscribers can receive the broadcast
+					// event within this timeout period, check if the subscribers,
+					// except the publisher, successfully receive the event.
+					assert.Equal(t, 2, rcv)
+					return
+				case <-ctx.Done():
+					return
+				}
+			}
+		}()
+
+		wg.Wait()
+	})
+
+	t.Run("refuse to subscribe document type for broadcasting", func(t *testing.T) {
+		ctx := context.Background()
+
+		d1 := document.New(helper.TestDocKey(t))
+		assert.NoError(t, c1.Attach(ctx, d1))
+		_, err := c1.Watch(ctx, d1)
+		assert.NoError(t, err)
+		err = d1.SubscribeBroadcastEvent("document", nil)
+		assert.ErrorIs(t, err, document.ErrReservedEventType)
+	})
+
+	t.Run("no broadcasts to unsubscribers", func(t *testing.T) {
+		bch := make(chan string)
+		ctx := context.Background()
+		handler := func(eventType, publisher string, payload []byte) error {
+			var mentionedBy string
+			assert.NoError(t, gojson.Unmarshal(payload, &mentionedBy))
+			// Send the unmarshaled payload to the channel to notify that this
+			// subscriber receives the event.
+			bch <- mentionedBy
+			return nil
+		}
+
+		d1 := document.New(helper.TestDocKey(t))
+		assert.NoError(t, c1.Attach(ctx, d1))
+		rch1, err := c1.Watch(ctx, d1)
+		assert.NoError(t, err)
+		err = d1.SubscribeBroadcastEvent("mention", handler)
+		assert.NoError(t, err)
+
+		d2 := document.New(helper.TestDocKey(t))
+		assert.NoError(t, c2.Attach(ctx, d2))
+		rch2, err := c2.Watch(ctx, d2)
+		assert.NoError(t, err)
+		err = d2.SubscribeBroadcastEvent("mention", handler)
+		assert.NoError(t, err)
+
+		// d1 unsubscribes to the broadcast event.
+		err = d1.UnsubscribeBroadcastEvent("mention")
+		assert.NoError(t, err)
+
+		err = d2.Broadcast("mention", "yorkie")
+		assert.NoError(t, err)
+
+		wg := sync.WaitGroup{}
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			rcv := 0
+			for {
+				select {
+				case <-rch1:
+				case <-rch2:
+				case m := <-bch:
+					assert.Equal(t, "yorkie", m)
+					rcv++
+				case <-time.After(1 * time.Second):
+					// Assuming that all subscribers can receive the broadcast
+					// event within this timeout period, check if the unsubscriber
+					// doesn't receive the event.
+					assert.Equal(t, 0, rcv)
+					return
+				case <-ctx.Done():
+					return
+				}
+			}
+		}()
+
+		wg.Wait()
+	})
+
+	t.Run("no duplicate broadcasts for duplicate subscriptions", func(t *testing.T) {
+		// t.Skip()
+		bch := make(chan string)
+		ctx := context.Background()
+		handler := func(eventType, publisher string, payload []byte) error {
+			var mentionedBy string
+			assert.NoError(t, gojson.Unmarshal(payload, &mentionedBy))
+			// Send the unmarshaled payload to the channel to notify that this
+			// subscriber receives the event.
+			bch <- mentionedBy
+			return nil
+		}
+
+		d1 := document.New(helper.TestDocKey(t))
+		assert.NoError(t, c1.Attach(ctx, d1))
+		rch1, err := c1.Watch(ctx, d1)
+		assert.NoError(t, err)
+		err = d1.SubscribeBroadcastEvent("mention", handler)
+		assert.NoError(t, err)
+
+		println("second sub start")
+
+		// d1 subscribes to the broadcast event twice.
+		err = d1.SubscribeBroadcastEvent("mention", handler)
+		assert.NoError(t, err)
+
+		println("second sub end")
+
+		d2 := document.New(helper.TestDocKey(t))
+		assert.NoError(t, c2.Attach(ctx, d2))
+		rch2, err := c2.Watch(ctx, d2)
+		assert.NoError(t, err)
+		err = d2.SubscribeBroadcastEvent("mention", handler)
+		assert.NoError(t, err)
+
 		err = d2.Broadcast("mention", "yorkie")
 		assert.NoError(t, err)
 
@@ -460,7 +606,10 @@ func TestDocument(t *testing.T) {
 				case m := <-bch:
 					assert.Equal(t, "yorkie", m)
 					rcv++
-				case <-time.After(3 * time.Second):
+				case <-time.After(1 * time.Second):
+					// Assuming that all subscribers can receive the broadcast
+					// event within this timeout period, check if the unsubscriber
+					// doesn't receive the event.
 					assert.Equal(t, 1, rcv)
 					return
 				case <-ctx.Done():
@@ -472,9 +621,6 @@ func TestDocument(t *testing.T) {
 		wg.Wait()
 	})
 
-	// TODO(sejongk): broadcast to multiple subscribers
-	// TODO(sejongk): dont broadcast to unsubscribers
-	// TODO(sejongk): reject subscribing document for broadcast
 }
 
 func TestDocumentWithProjects(t *testing.T) {