Fix monitor on go 1.19 (#2908)

* Fix monitor on go 1.19

* Remove exmaple tests when go 1.19

* Fix typo

* Fix typo

* Skip exmaple test

* Skip exmaple test

* Guard Peek call with mutex for thread safety

Author: ofekshenawa

.github/workflows/build.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        go-version: [1.20.x, 1.21.x]
+        go-version: [1.19.x, 1.20.x, 1.21.x]
 
     services:
       redis:

Makefile

@@ -1,7 +1,12 @@
 GO_MOD_DIRS := $(shell find . -type f -name 'go.mod' -exec dirname {} \; | sort)
 
 test: testdeps
+	$(eval GO_VERSION := $(shell go version | cut -d " " -f 3 | cut -d. -f2))
 	set -e; for dir in $(GO_MOD_DIRS); do \
+	  if echo "$${dir}" | grep -q "./example" && [ "$(GO_VERSION)" = "19" ]; then \
+	    echo "Skipping go test in $${dir} due to Go version 1.19 and dir contains ./example"; \
+	    continue; \
+	  fi; \
 	  echo "go test in $${dir}"; \
 	  (cd "$${dir}" && \
 	    go mod tidy -compat=1.18 && \

command.go

@@ -5454,9 +5454,12 @@ func (cmd *MonitorCmd) readMonitor(rd *proto.Reader, cancel context.CancelFunc)
 	for {
 		cmd.mu.Lock()
 		st := cmd.status
+		pk, _ := rd.Peek(1)
 		cmd.mu.Unlock()
-		if pk, _ := rd.Peek(1); len(pk) != 0 && st == monitorStatusStart {
+		if len(pk) != 0 && st == monitorStatusStart {
+			cmd.mu.Lock()
 			line, err := rd.ReadString()
+			cmd.mu.Unlock()
 			if err != nil {
 				return err
 			}

monitor_test.go

@@ -2,8 +2,11 @@ package redis_test
 
 import (
 	"context"
+	"strings"
 	"time"
 
+	"testing"
+
 	. "github.com/bsm/ginkgo/v2"
 	. "github.com/bsm/gomega"
 
@@ -46,3 +49,52 @@ var _ = Describe("Monitor command", Label("monitor"), func() {
 		Expect(lst[3]).To(ContainSubstring(`"set" "bap" "8"`))
 	})
 })
+
+func TestMonitorCommand(t *testing.T) {
+	ctx := context.TODO()
+	client := redis.NewClient(&redis.Options{Addr: ":6379"})
+	if err := client.FlushDB(ctx).Err(); err != nil {
+		t.Fatalf("FlushDB failed: %v", err)
+	}
+
+	defer func() {
+		if err := client.Close(); err != nil {
+			t.Fatalf("Close failed: %v", err)
+		}
+	}()
+
+	ress := make(chan string, 10)                             // Buffer to prevent blocking
+	client1 := redis.NewClient(&redis.Options{Addr: ":6379"}) // Adjust the Addr field as necessary
+	mn := client1.Monitor(ctx, ress)
+	mn.Start()
+	// Wait for the Redis server to be in monitoring mode.
+	time.Sleep(100 * time.Millisecond)
+	client.Set(ctx, "foo", "bar", 0)
+	client.Set(ctx, "bar", "baz", 0)
+	client.Set(ctx, "bap", 8, 0)
+	client.Get(ctx, "bap")
+	mn.Stop()
+	var lst []string
+	for i := 0; i < 5; i++ {
+		s := <-ress
+		lst = append(lst, s)
+	}
+
+	// Assertions
+	if !containsSubstring(lst[0], "OK") {
+		t.Errorf("Expected lst[0] to contain 'OK', got %s", lst[0])
+	}
+	if !containsSubstring(lst[1], `"set" "foo" "bar"`) {
+		t.Errorf(`Expected lst[1] to contain '"set" "foo" "bar"', got %s`, lst[1])
+	}
+	if !containsSubstring(lst[2], `"set" "bar" "baz"`) {
+		t.Errorf(`Expected lst[2] to contain '"set" "bar" "baz"', got %s`, lst[2])
+	}
+	if !containsSubstring(lst[3], `"set" "bap" "8"`) {
+		t.Errorf(`Expected lst[3] to contain '"set" "bap" "8"', got %s`, lst[3])
+	}
+}
+
+func containsSubstring(s, substr string) bool {
+	return strings.Contains(s, substr)
+}