Merge pull request #224 from alicebob/dollar

allow '$' in XREAD
alicebob · Sep 16, 2021 · 90c554a · 90c554a
2 parents a31f12a + 6798993
commit 90c554a
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 1 deletion.
diff --git a/cmd_stream.go b/cmd_stream.go
@@ -646,7 +646,7 @@ parsing:
 
 			opts.streams, opts.ids = args[0:len(args)/2], args[len(args)/2:]
 			for _, id := range opts.ids {
-				if _, err := parseStreamID(id); err != nil {
+				if _, err := parseStreamID(id); id != `$` && err != nil {
 					setDirty(c)
 					c.WriteError(msgInvalidStreamID)
 					return
@@ -718,6 +718,9 @@ func xread(db *RedisDB, streams []string, ids []string, count int) map[string][]
 			if len(returnedEntries) == entryCount {
 				break
 			}
+			if id == "$" {
+				id = s.lastID()
+			}
 			if streamCmp(entry.ID, id) <= 0 {
 				continue
 			}

diff --git a/integration/stream_test.go b/integration/stream_test.go
@@ -244,6 +244,19 @@ func TestStream(t *testing.T) {
 			c.Error("not an int", "XREAD", "BLOCK", "foo", "STREAMS", "pl", "0")
 			c.Error("negative", "XREAD", "BLOCK", "-12", "STREAMS", "pl", "0")
 		})
+
+		// special '$' ID
+		testRaw2(t, func(c, c2 *client) {
+			var wg sync.WaitGroup
+			wg.Add(1)
+			go func() {
+				time.Sleep(10 * time.Millisecond)
+				c2.Do("XADD", "pl", "60-1", "name", "Mercury")
+				wg.Done()
+			}()
+			wg.Wait()
+			c.Do("XREAD", "BLOCK", "1000", "STREAMS", "pl", "$")
+		})
 	})
 }