add integration test

alicebob · Jun 7, 2024 · ab408dd · ab408dd
1 parent 368f067
commit ab408dd
Show file tree

Hide file tree

Showing 4 changed files with 58 additions and 3 deletions.
diff --git a/cmd_stream_test.go b/cmd_stream_test.go
@@ -945,6 +945,36 @@ func TestStreamDelete(t *testing.T) {
 			),
 		),
 	)
+
+	mustDo(t, c,
+		"XREADGROUP", "GROUP", "processing", "alice", "STREAMS", "planets", ">",
+		proto.Array(
+			proto.Array(
+				proto.String("planets"),
+				proto.Array(
+					proto.Array(
+						proto.String("0-4"),
+						proto.Strings("name", "Jupiter"),
+					),
+				),
+			),
+		),
+	)
+
+	mustDo(t, c,
+		"XINFO", "GROUPS", "planets",
+		proto.Array(
+			proto.Array(
+				proto.String("name"), proto.String("processing"),
+				proto.String("consumers"), proto.Int(1),
+				proto.String("pending"), proto.Int(1),
+				proto.String("last-delivered-id"), proto.String("0-4"),
+				proto.String("entries-read"), proto.Int(4),
+				proto.String("lag"), proto.Int(0),
+			),
+		),
+	)
+
 }
 
 // Test XACK

diff --git a/integration/stream_test.go b/integration/stream_test.go
@@ -531,6 +531,30 @@ func TestStreamGroup(t *testing.T) {
 		})
 	})
 
+	t.Run("XREADGROUP XDEL XINFO lag", func(t *testing.T) {
+		testRaw(t, func(c *client) {
+			// TODO ALL INFO needs to be uncommented, even DoLoosely is not enough
+			//  for some of them becase Nil is returned instead of Int
+			c.Do("XGROUP", "CREATE", "planets", "processing", "$", "MKSTREAM")
+			c.Do("XADD", "planets", "0-1", "name", "Mercury")
+			c.Do("XADD", "planets", "0-2", "name", "Venus")
+			c.Do("XREADGROUP", "GROUP", "processing", "alice", "STREAMS", "planets", ">")
+			//c.Do("XINFO", "GROUPS", "planets") //SPEC entries-added 2,  entries-read = 2, lag = 0
+			c.Do("XADD", "planets", "0-3", "name", "Earth")
+			c.Do("XADD", "planets", "0-4", "name", "Jupiter")
+			//c.Do("XINFO", "GROUPS", "planets") //SPEC entries-added 4,  entries-read = 2, lag = 2
+
+			c.Do("XDEL", "planets", "0-1")
+			//c.Do("XINFO", "GROUPS", "planets") // SPEC entries-added 4, entries-read = 2, lag = 2
+			c.Do("XDEL", "planets", "0-2")
+			//c.Do("XINFO", "GROUPS", "planets") // SPEC entries-added 4,  entries-read = 2, lag = 2
+			c.Do("XDEL", "planets", "0-3")
+			//c.Do("XINFO", "GROUPS", "planets") // SPEC entries-added 4,  entries-read = 2, lag = nil
+			c.Do("XREADGROUP", "GROUP", "processing", "alice", "STREAMS", "planets", ">")
+			//c.Do("XINFO", "GROUPS", "planets") // SPEC entries-added 4,  entries-read = 4,  lag = 0
+		})
+	})
+
 	t.Run("XACK", func(t *testing.T) {
 		testRaw(t, func(c *client) {
 			c.Do("XGROUP", "CREATE", "planets", "processing", "$", "MKSTREAM")

diff --git a/integration/test.go b/integration/test.go
@@ -346,7 +346,7 @@ func (c *client) Do(cmd string, args ...string) {
 	// c.t.Logf("real:%q mini:%q", string(resReal), string(resMini))
 
 	if resReal != resMini {
-		c.t.Errorf("real: %q mini: %q", string(resReal), string(resMini))
+		c.t.Errorf("\nreal: %q \nmini: %q", string(resReal), string(resMini))
 		return
 	}
 

diff --git a/stream.go b/stream.go
@@ -233,7 +233,7 @@ func (s *streamKey) createGroup(group, id string) error {
 	var entriesReadValid = true
 	if id == "$" {
 		id = s.lastIDUnlocked()
-		entriesRead = len(s.entries)
+		entriesRead = s.entriesAdded
 	} else if id == "0" || id == "0-0" || id == s.firstIDUnlocked() {
 		entriesRead = 0
 	} else {
@@ -565,6 +565,7 @@ func (g *streamGroup) updateEntriesRead(msgs []StreamEntry) {
 		g.entriesRead += len(msgs)
 	} else if g.lastID == g.stream.lastIDUnlocked() {
 		// reset entries read as we catch up to the last ID
-		g.entriesRead = len(g.stream.entries)
+		g.entriesRead = g.stream.entriesAdded
+		g.entriesReadValid = true
 	}
 }