package compatibility import ( "fmt" "sync" "testing" "time" "github.com/gomodule/redigo/redis" ) // dialRedis creates a new redigo connection for testing func dialRedis(t *testing.T, addr string) redis.Conn { t.Helper() conn, err := redis.Dial("tcp", addr, redis.DialConnectTimeout(5*time.Second), redis.DialReadTimeout(5*time.Second), redis.DialWriteTimeout(5*time.Second), ) if err != nil { t.Fatalf("Failed to dial: %v", err) } return conn } // TestRedigoBasicSubscribe tests basic SUBSCRIBE functionality func TestRedigoBasicSubscribe(t *testing.T) { ts := startTestServer(t) defer ts.Close() conn := dialRedis(t, ts.Addr()) defer func() { _ = conn.Close() }() // Send SUBSCRIBE err := conn.Send("SUBSCRIBE", "rigo-test-channel") if err != nil { t.Fatalf("Send failed: %v", err) } _ = conn.Flush() // Read subscription confirmation msg, err := redis.Values(conn.Receive()) if err != nil { t.Fatalf("Receive failed: %v", err) } // Should be ["subscribe", "channel", count] if len(msg) != 3 { t.Fatalf("Expected 3 elements, got %d", len(msg)) } msgType, _ := redis.String(msg[0], nil) if msgType != "subscribe" { t.Errorf("Expected 'subscribe', got '%s'", msgType) } channel, _ := redis.String(msg[1], nil) if channel != "rigo-test-channel" { t.Errorf("Expected 'rigo-test-channel', got '%s'", channel) } } // TestRedigoBasicPublish tests basic PUBLISH functionality func TestRedigoBasicPublish(t *testing.T) { ts := startTestServer(t) defer ts.Close() // Subscriber subConn := dialRedis(t, ts.Addr()) defer func() { _ = subConn.Close() }() err := subConn.Send("SUBSCRIBE", "rigo-pub-test") if err != nil { t.Fatalf("Send failed: %v", err) } _ = subConn.Flush() // Consume subscription confirmation _, err = redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive subscription failed: %v", err) } // Publisher pubConn := dialRedis(t, ts.Addr()) defer func() { _ = pubConn.Close() }() reply, err := redis.Int(pubConn.Do("PUBLISH", "rigo-pub-test", "hello redigo")) if err != nil { t.Fatalf("Publish failed: %v", err) } t.Logf("Published to %d subscribers", reply) // Receive message msg, err := redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive message failed: %v", err) } // Should be ["message", "channel", payload] if len(msg) != 3 { t.Fatalf("Expected 3 elements, got %d", len(msg)) } msgType, _ := redis.String(msg[0], nil) if msgType != "message" { t.Errorf("Expected 'message', got '%s'", msgType) } payload, _ := redis.String(msg[2], nil) if payload != "hello redigo" { t.Errorf("Expected 'hello redigo', got '%s'", payload) } } // TestRedigoMultipleSubscribers tests multiple subscribers func TestRedigoMultipleSubscribers(t *testing.T) { ts := startTestServer(t) defer ts.Close() numSubs := 3 conns := make([]redis.Conn, numSubs) // Create subscribers for i := 0; i < numSubs; i++ { conn := dialRedis(t, ts.Addr()) conns[i] = conn err := conn.Send("SUBSCRIBE", "rigo-multi") if err != nil { t.Fatalf("Send failed for %d: %v", i, err) } _ = conn.Flush() // Consume confirmation _, err = redis.Values(conn.Receive()) if err != nil { t.Fatalf("Receive confirmation failed for %d: %v", i, err) } } defer func() { for _, c := range conns { _ = c.Close() } }() // Publisher pubConn := dialRedis(t, ts.Addr()) defer func() { _ = pubConn.Close() }() _, err := pubConn.Do("PUBLISH", "rigo-multi", "broadcast") if err != nil { t.Fatalf("Publish failed: %v", err) } // All should receive for i, conn := range conns { msg, err := redis.Values(conn.Receive()) if err != nil { t.Errorf("Subscriber %d failed to receive: %v", i, err) continue } payload, _ := redis.String(msg[2], nil) if payload != "broadcast" { t.Errorf("Subscriber %d got wrong payload: %s", i, payload) } } } // TestRedigoUnsubscribe tests UNSUBSCRIBE functionality func TestRedigoUnsubscribe(t *testing.T) { ts := startTestServer(t) defer ts.Close() conn := dialRedis(t, ts.Addr()) defer func() { _ = conn.Close() }() // Subscribe err := conn.Send("SUBSCRIBE", "rigo-unsub-test") if err != nil { t.Fatalf("Send failed: %v", err) } _ = conn.Flush() // Consume confirmation _, err = redis.Values(conn.Receive()) if err != nil { t.Fatalf("Receive failed: %v", err) } // Unsubscribe err = conn.Send("UNSUBSCRIBE", "rigo-unsub-test") if err != nil { t.Fatalf("Send unsubscribe failed: %v", err) } _ = conn.Flush() // Receive unsubscribe confirmation msg, err := redis.Values(conn.Receive()) if err != nil { t.Fatalf("Receive unsubscribe confirmation failed: %v", err) } msgType, _ := redis.String(msg[0], nil) if msgType != "unsubscribe" { t.Errorf("Expected 'unsubscribe', got '%s'", msgType) } } // TestRedigoPatternSubscribe tests PSUBSCRIBE functionality func TestRedigoPatternSubscribe(t *testing.T) { ts := startTestServer(t) defer ts.Close() conn := dialRedis(t, ts.Addr()) defer func() { _ = conn.Close() }() // PSubscribe err := conn.Send("PSUBSCRIBE", "rigo-news.*") if err != nil { t.Fatalf("Send failed: %v", err) } _ = conn.Flush() // Receive confirmation msg, err := redis.Values(conn.Receive()) if err != nil { t.Fatalf("Receive failed: %v", err) } msgType, _ := redis.String(msg[0], nil) if msgType != "psubscribe" { t.Errorf("Expected 'psubscribe', got '%s'", msgType) } pattern, _ := redis.String(msg[1], nil) if pattern != "rigo-news.*" { t.Errorf("Expected 'rigo-news.*', got '%s'", pattern) } } // TestRedigoPatternMatch tests pattern matching with PUBLISH func TestRedigoPatternMatch(t *testing.T) { ts := startTestServer(t) defer ts.Close() // Subscriber with pattern subConn := dialRedis(t, ts.Addr()) defer func() { _ = subConn.Close() }() err := subConn.Send("PSUBSCRIBE", "rigo-pattern.*") if err != nil { t.Fatalf("Send failed: %v", err) } _ = subConn.Flush() // Consume confirmation _, err = redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive confirmation failed: %v", err) } t.Log("Pattern subscription confirmed, now publishing...") // Publisher pubConn := dialRedis(t, ts.Addr()) defer func() { _ = pubConn.Close() }() reply, err := redis.Int(pubConn.Do("PUBLISH", "rigo-pattern.match", "pattern message")) if err != nil { t.Fatalf("Publish failed: %v", err) } t.Logf("Publish returned: %d", reply) // Receive pmessage msg, err := redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive failed: %v", err) } // Should be ["pmessage", "pattern", "channel", "payload"] if len(msg) != 4 { t.Fatalf("Expected 4 elements, got %d", len(msg)) } msgType, _ := redis.String(msg[0], nil) if msgType != "pmessage" { t.Errorf("Expected 'pmessage', got '%s'", msgType) } pattern, _ := redis.String(msg[1], nil) if pattern != "rigo-pattern.*" { t.Errorf("Expected pattern 'rigo-pattern.*', got '%s'", pattern) } channel, _ := redis.String(msg[2], nil) if channel != "rigo-pattern.match" { t.Errorf("Expected channel 'rigo-pattern.match', got '%s'", channel) } payload, _ := redis.String(msg[3], nil) if payload != "pattern message" { t.Errorf("Expected 'pattern message', got '%s'", payload) } } // TestRedigoMultiplePatterns tests subscribing to multiple patterns func TestRedigoMultiplePatterns(t *testing.T) { ts := startTestServer(t) defer ts.Close() conn := dialRedis(t, ts.Addr()) defer func() { _ = conn.Close() }() patterns := []string{"rigo-pat1.*", "rigo-pat2.*", "rigo-pat3.*"} for _, pattern := range patterns { err := conn.Send("PSUBSCRIBE", pattern) if err != nil { t.Fatalf("Send failed: %v", err) } _ = conn.Flush() // Consume confirmation _, err = redis.Values(conn.Receive()) if err != nil { t.Fatalf("Receive confirmation failed: %v", err) } } } // TestRedigoPing tests PING command func TestRedigoPing(t *testing.T) { ts := startTestServer(t) defer ts.Close() conn := dialRedis(t, ts.Addr()) defer func() { _ = conn.Close() }() result, err := redis.String(conn.Do("PING")) if err != nil { t.Fatalf("Ping failed: %v", err) } if result != "PONG" { t.Errorf("Expected 'PONG', got '%s'", result) } } // TestRedigoPingWithMessage tests PING with message func TestRedigoPingWithMessage(t *testing.T) { ts := startTestServer(t) defer ts.Close() conn := dialRedis(t, ts.Addr()) defer func() { _ = conn.Close() }() result, err := redis.String(conn.Do("PING", "hello")) if err != nil { t.Fatalf("Ping failed: %v", err) } if result != "hello" { t.Errorf("Expected 'hello', got '%s'", result) } } // TestRedigoBinarySafe tests binary-safe messages func TestRedigoBinarySafe(t *testing.T) { ts := startTestServer(t) defer ts.Close() // Subscriber subConn := dialRedis(t, ts.Addr()) defer func() { _ = subConn.Close() }() err := subConn.Send("SUBSCRIBE", "rigo-binary") if err != nil { t.Fatalf("Send failed: %v", err) } _ = subConn.Flush() _, err = redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive confirmation failed: %v", err) } // Publisher pubConn := dialRedis(t, ts.Addr()) defer func() { _ = pubConn.Close() }() binaryMsg := []byte{0x00, 0x01, 0x02, 0xFF, 0xFE, '\r', '\n', 0x00} _, err = pubConn.Do("PUBLISH", "rigo-binary", binaryMsg) if err != nil { t.Fatalf("Publish failed: %v", err) } // Receive message msg, err := redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive failed: %v", err) } payload, _ := redis.String(msg[2], nil) if []byte(payload)[0] != binaryMsg[0] { t.Errorf("Binary message mismatch") } } // TestRedigisEmptyMessage tests empty message func TestRedigisEmptyMessage(t *testing.T) { ts := startTestServer(t) defer ts.Close() // Subscriber subConn := dialRedis(t, ts.Addr()) defer func() { _ = subConn.Close() }() err := subConn.Send("SUBSCRIBE", "rigo-empty") if err != nil { t.Fatalf("Send failed: %v", err) } _ = subConn.Flush() _, err = redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive confirmation failed: %v", err) } // Publisher pubConn := dialRedis(t, ts.Addr()) defer func() { _ = pubConn.Close() }() _, err = pubConn.Do("PUBLISH", "rigo-empty", "") if err != nil { t.Fatalf("Publish failed: %v", err) } // Receive message msg, err := redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive failed: %v", err) } payload, _ := redis.String(msg[2], nil) if payload != "" { t.Errorf("Expected empty payload, got '%s'", payload) } } // TestRedigoLongMessage tests long message func TestRedigoLongMessage(t *testing.T) { ts := startTestServer(t) defer ts.Close() // Subscriber subConn := dialRedis(t, ts.Addr()) defer func() { _ = subConn.Close() }() err := subConn.Send("SUBSCRIBE", "rigo-long") if err != nil { t.Fatalf("Send failed: %v", err) } _ = subConn.Flush() _, err = redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive confirmation failed: %v", err) } // Publisher pubConn := dialRedis(t, ts.Addr()) defer func() { _ = pubConn.Close() }() // 100KB message longMsg := make([]byte, 100*1024) for i := range longMsg { longMsg[i] = byte(i % 256) } _, err = pubConn.Do("PUBLISH", "rigo-long", longMsg) if err != nil { t.Fatalf("Publish failed: %v", err) } // Receive message msg, err := redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive failed: %v", err) } payload, _ := redis.String(msg[2], nil) if len(payload) != len(longMsg) { t.Errorf("Message length mismatch: got %d, want %d", len(payload), len(longMsg)) } } // TestRedigoConcurrentSubscribers tests concurrent operations func TestRedigoConcurrentSubscribers(t *testing.T) { ts := startTestServer(t) defer ts.Close() var wg sync.WaitGroup for i := 0; i < 10; i++ { wg.Add(1) go func(id int) { defer wg.Done() conn := dialRedis(t, ts.Addr()) defer func() { _ = conn.Close() }() channel := fmt.Sprintf("rigo-concurrent-%d", id%3) err := conn.Send("SUBSCRIBE", channel) if err != nil { t.Errorf("Send failed: %v", err) return } _ = conn.Flush() _, err = redis.Values(conn.Receive()) if err != nil { t.Errorf("Receive failed: %v", err) return } time.Sleep(50 * time.Millisecond) }(i) } wg.Wait() } // TestRedigoPublishNoSubscribers tests PUBLISH with no subscribers func TestRedigoPublishNoSubscribers(t *testing.T) { ts := startTestServer(t) defer ts.Close() conn := dialRedis(t, ts.Addr()) defer func() { _ = conn.Close() }() count, err := redis.Int(conn.Do("PUBLISH", "rigo-no-subs", "message")) if err != nil { t.Fatalf("Publish failed: %v", err) } if count != 0 { t.Errorf("Expected 0 receivers, got %d", count) } } // TestRedigoUnicodeMessages tests Unicode messages func TestRedigoUnicodeMessages(t *testing.T) { ts := startTestServer(t) defer ts.Close() messages := []string{ "Hello, 世界!", "Emoji: 🎉", "Cyrillic: Привет", } for _, testMsg := range messages { t.Run(testMsg[:min(20, len(testMsg))], func(t *testing.T) { // Subscriber subConn := dialRedis(t, ts.Addr()) defer func() { _ = subConn.Close() }() err := subConn.Send("SUBSCRIBE", "rigo-unicode") if err != nil { t.Fatalf("Send failed: %v", err) } _ = subConn.Flush() _, err = redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive confirmation failed: %v", err) } // Publisher pubConn := dialRedis(t, ts.Addr()) defer func() { _ = pubConn.Close() }() _, err = pubConn.Do("PUBLISH", "rigo-unicode", testMsg) if err != nil { t.Fatalf("Publish failed: %v", err) } // Receive msg, err := redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive failed: %v", err) } payload, _ := redis.String(msg[2], nil) if payload != testMsg { t.Errorf("Unicode mismatch: got %q, want %q", payload, testMsg) } }) } } // TestRedigoChannelWithSpecialChars tests channels with special characters func TestRedigoChannelWithSpecialChars(t *testing.T) { ts := startTestServer(t) defer ts.Close() channels := []string{ "rigo-with-dash", "rigo_with_underscore", "rigo.with.dots", "rigo:with:colons", } for _, channelName := range channels { t.Run(channelName, func(t *testing.T) { // Subscriber subConn := dialRedis(t, ts.Addr()) defer func() { _ = subConn.Close() }() err := subConn.Send("SUBSCRIBE", channelName) if err != nil { t.Fatalf("Send failed: %v", err) } _ = subConn.Flush() _, err = redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive confirmation failed: %v", err) } // Publisher pubConn := dialRedis(t, ts.Addr()) defer func() { _ = pubConn.Close() }() _, err = pubConn.Do("PUBLISH", channelName, "test") if err != nil { t.Fatalf("Publish failed: %v", err) } // Receive msg, err := redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive failed: %v", err) } channel, _ := redis.String(msg[1], nil) if channel != channelName { t.Errorf("Channel mismatch: got %s, want %s", channel, channelName) } }) } } // TestRedigoMixedSubscribePSubscribe tests mixing SUBSCRIBE and PSUBSCRIBE // nolint: cyclop func TestRedigoMixedSubscribePSubscribe(t *testing.T) { ts := startTestServer(t) defer ts.Close() conn := dialRedis(t, ts.Addr()) defer func() { _ = conn.Close() }() // Subscribe to channel err := conn.Send("SUBSCRIBE", "rigo-mixed-ch") if err != nil { t.Fatalf("Send subscribe failed: %v", err) } _ = conn.Flush() _, err = redis.Values(conn.Receive()) if err != nil { t.Fatalf("Receive subscribe confirmation failed: %v", err) } // Also subscribe to pattern err = conn.Send("PSUBSCRIBE", "rigo-mixed-*") if err != nil { t.Fatalf("Send psubscribe failed: %v", err) } _ = conn.Flush() _, err = redis.Values(conn.Receive()) if err != nil { t.Fatalf("Receive psubscribe confirmation failed: %v", err) } // Publish to channel - should receive via channel subscription pubConn := dialRedis(t, ts.Addr()) defer func() { _ = pubConn.Close() }() _, err = pubConn.Do("PUBLISH", "rigo-mixed-ch", "channel msg") if err != nil { t.Fatalf("Publish failed: %v", err) } msg, err := redis.Values(conn.Receive()) if err != nil { t.Fatalf("Receive failed: %v", err) } msgType, _ := redis.String(msg[0], nil) if msgType != "message" { t.Errorf("Expected 'message', got '%s'", msgType) } // Publish to pattern-matching channel - should receive via pattern subscription _, err = pubConn.Do("PUBLISH", "rigo-mixed-pattern", "pattern msg") if err != nil { t.Fatalf("Publish failed: %v", err) } msg, err = redis.Values(conn.Receive()) if err != nil { t.Fatalf("Receive failed: %v", err) } msgType, _ = redis.String(msg[0], nil) if msgType != "pmessage" { t.Errorf("Expected 'pmessage', got '%s'", msgType) } } // TestRedigoPunsubscribe tests PUNSUBSCRIBE functionality func TestRedigoPunsubscribe(t *testing.T) { ts := startTestServer(t) defer ts.Close() conn := dialRedis(t, ts.Addr()) defer func() { _ = conn.Close() }() // PSubscribe err := conn.Send("PSUBSCRIBE", "rigo-punsub.*") if err != nil { t.Fatalf("Send failed: %v", err) } _ = conn.Flush() _, err = redis.Values(conn.Receive()) if err != nil { t.Fatalf("Receive confirmation failed: %v", err) } // Punsubscribe err = conn.Send("PUNSUBSCRIBE", "rigo-punsub.*") if err != nil { t.Fatalf("Send punsubscribe failed: %v", err) } _ = conn.Flush() msg, err := redis.Values(conn.Receive()) if err != nil { t.Fatalf("Receive punsubscribe confirmation failed: %v", err) } msgType, _ := redis.String(msg[0], nil) if msgType != "punsubscribe" { t.Errorf("Expected 'punsubscribe', got '%s'", msgType) } } // TestRedigoConcurrentPublishSubscribe tests concurrent pub/sub func TestRedigoConcurrentPublishSubscribe(t *testing.T) { ts := startTestServer(t) defer ts.Close() numMessages := 50 // Subscriber subConn := dialRedis(t, ts.Addr()) defer func() { _ = subConn.Close() }() err := subConn.Send("SUBSCRIBE", "rigo-concurrent-pub") if err != nil { t.Fatalf("Send failed: %v", err) } _ = subConn.Flush() _, err = redis.Values(subConn.Receive()) if err != nil { t.Fatalf("Receive confirmation failed: %v", err) } // Publisher in goroutine var wg sync.WaitGroup wg.Add(1) go func() { defer wg.Done() pubConn := dialRedis(t, ts.Addr()) defer func() { _ = pubConn.Close() }() for i := 0; i < numMessages; i++ { _, err := pubConn.Do("PUBLISH", "rigo-concurrent-pub", fmt.Sprintf("msg-%d", i)) if err != nil { t.Errorf("Publish %d failed: %v", i, err) return } time.Sleep(time.Millisecond) } }() // Receive messages received := 0 timeout := time.After(10 * time.Second) for received < numMessages { select { case <-timeout: t.Logf("Received %d/%d messages", received, numMessages) wg.Wait() return default: } msg, err := redis.Values(subConn.Receive()) if err != nil { t.Errorf("Receive failed: %v", err) break } msgType, _ := redis.String(msg[0], nil) if msgType == "message" { received++ } } wg.Wait() t.Logf("Received all %d messages", received) }