feat: add general push notification system

- Add PushNotificationRegistry for managing notification handlers
- Add PushNotificationProcessor for processing RESP3 push notifications
- Add client methods for registering push notification handlers
- Add PubSub integration for handling generic push notifications
- Add comprehensive test suite with 100% coverage
- Add push notification demo example

This system allows handling any arbitrary RESP3 push notification
with registered handlers, not just specific notification types.

Author: ndyakov

example/push-notification-demo/main.go

@@ -0,0 +1,262 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	"github.com/redis/go-redis/v9"
+)
+
+func main() {
+	fmt.Println("Redis Go Client - General Push Notification System Demo")
+	fmt.Println("======================================================")
+
+	// Example 1: Basic push notification setup
+	basicPushNotificationExample()
+
+	// Example 2: Custom push notification handlers
+	customHandlersExample()
+
+	// Example 3: Global push notification handlers
+	globalHandlersExample()
+
+	// Example 4: Custom push notifications
+	customPushNotificationExample()
+
+	// Example 5: Multiple notification types
+	multipleNotificationTypesExample()
+
+	// Example 6: Processor API demonstration
+	demonstrateProcessorAPI()
+}
+
+func basicPushNotificationExample() {
+	fmt.Println("\n=== Basic Push Notification Example ===")
+
+	// Create a Redis client with push notifications enabled
+	client := redis.NewClient(&redis.Options{
+		Addr:              "localhost:6379",
+		Protocol:          3,    // RESP3 required for push notifications
+		PushNotifications: true, // Enable general push notification processing
+	})
+	defer client.Close()
+
+	// Register a handler for custom notifications
+	client.RegisterPushNotificationHandlerFunc("CUSTOM_EVENT", func(ctx context.Context, notification []interface{}) bool {
+		fmt.Printf("Received CUSTOM_EVENT: %v\n", notification)
+		return true
+	})
+
+	fmt.Println("✅ Push notifications enabled and handler registered")
+	fmt.Println("   The client will now process any CUSTOM_EVENT push notifications")
+}
+
+func customHandlersExample() {
+	fmt.Println("\n=== Custom Push Notification Handlers Example ===")
+
+	client := redis.NewClient(&redis.Options{
+		Addr:              "localhost:6379",
+		Protocol:          3,
+		PushNotifications: true,
+	})
+	defer client.Close()
+
+	// Register handlers for different notification types
+	client.RegisterPushNotificationHandlerFunc("USER_LOGIN", func(ctx context.Context, notification []interface{}) bool {
+		if len(notification) >= 3 {
+			username := notification[1]
+			timestamp := notification[2]
+			fmt.Printf("🔐 User login: %v at %v\n", username, timestamp)
+		}
+		return true
+	})
+
+	client.RegisterPushNotificationHandlerFunc("CACHE_INVALIDATION", func(ctx context.Context, notification []interface{}) bool {
+		if len(notification) >= 2 {
+			cacheKey := notification[1]
+			fmt.Printf("🗑️  Cache invalidated: %v\n", cacheKey)
+		}
+		return true
+	})
+
+	client.RegisterPushNotificationHandlerFunc("SYSTEM_ALERT", func(ctx context.Context, notification []interface{}) bool {
+		if len(notification) >= 3 {
+			alertLevel := notification[1]
+			message := notification[2]
+			fmt.Printf("🚨 System alert [%v]: %v\n", alertLevel, message)
+		}
+		return true
+	})
+
+	fmt.Println("✅ Multiple custom handlers registered:")
+	fmt.Println("   - USER_LOGIN: Handles user authentication events")
+	fmt.Println("   - CACHE_INVALIDATION: Handles cache invalidation events")
+	fmt.Println("   - SYSTEM_ALERT: Handles system alert notifications")
+}
+
+func globalHandlersExample() {
+	fmt.Println("\n=== Global Push Notification Handler Example ===")
+
+	client := redis.NewClient(&redis.Options{
+		Addr:              "localhost:6379",
+		Protocol:          3,
+		PushNotifications: true,
+	})
+	defer client.Close()
+
+	// Register a global handler that receives ALL push notifications
+	client.RegisterGlobalPushNotificationHandlerFunc(func(ctx context.Context, notification []interface{}) bool {
+		if len(notification) > 0 {
+			command := notification[0]
+			fmt.Printf("📡 Global handler received: %v (args: %d)\n", command, len(notification)-1)
+		}
+		return true
+	})
+
+	// Register specific handlers as well
+	client.RegisterPushNotificationHandlerFunc("SPECIFIC_EVENT", func(ctx context.Context, notification []interface{}) bool {
+		fmt.Printf("🎯 Specific handler for SPECIFIC_EVENT: %v\n", notification)
+		return true
+	})
+
+	fmt.Println("✅ Global and specific handlers registered:")
+	fmt.Println("   - Global handler will receive ALL push notifications")
+	fmt.Println("   - Specific handler will receive only SPECIFIC_EVENT notifications")
+	fmt.Println("   - Both handlers will be called for SPECIFIC_EVENT notifications")
+}
+
+func customPushNotificationExample() {
+	fmt.Println("\n=== Custom Push Notifications Example ===")
+
+	// Create a client with custom push notifications
+	client := redis.NewClient(&redis.Options{
+		Addr:              "localhost:6379",
+		Protocol:          3,    // RESP3 required
+		PushNotifications: true, // Enable general push notifications
+	})
+	defer client.Close()
+
+	// Register custom handlers for application events
+	client.RegisterPushNotificationHandlerFunc("APPLICATION_EVENT", func(ctx context.Context, notification []interface{}) bool {
+		fmt.Printf("📱 Application event: %v\n", notification)
+		return true
+	})
+
+	// Register a global handler to monitor all notifications
+	client.RegisterGlobalPushNotificationHandlerFunc(func(ctx context.Context, notification []interface{}) bool {
+		if len(notification) > 0 {
+			command := notification[0]
+			switch command {
+			case "MOVING", "MIGRATING", "MIGRATED":
+				fmt.Printf("🔄 Cluster notification: %v\n", command)
+			default:
+				fmt.Printf("📨 Other notification: %v\n", command)
+			}
+		}
+		return true
+	})
+
+	fmt.Println("✅ Custom push notifications enabled:")
+	fmt.Println("   - MOVING, MIGRATING, MIGRATED notifications → Cluster handlers")
+	fmt.Println("   - APPLICATION_EVENT notifications → Custom handler")
+	fmt.Println("   - All notifications → Global monitoring handler")
+}
+
+func multipleNotificationTypesExample() {
+	fmt.Println("\n=== Multiple Notification Types Example ===")
+
+	client := redis.NewClient(&redis.Options{
+		Addr:              "localhost:6379",
+		Protocol:          3,
+		PushNotifications: true,
+	})
+	defer client.Close()
+
+	// Register handlers for Redis built-in notification types
+	client.RegisterPushNotificationHandlerFunc(redis.PushNotificationPubSubMessage, func(ctx context.Context, notification []interface{}) bool {
+		fmt.Printf("💬 Pub/Sub message: %v\n", notification)
+		return true
+	})
+
+	client.RegisterPushNotificationHandlerFunc(redis.PushNotificationKeyspace, func(ctx context.Context, notification []interface{}) bool {
+		fmt.Printf("🔑 Keyspace notification: %v\n", notification)
+		return true
+	})
+
+	client.RegisterPushNotificationHandlerFunc(redis.PushNotificationKeyevent, func(ctx context.Context, notification []interface{}) bool {
+		fmt.Printf("⚡ Key event notification: %v\n", notification)
+		return true
+	})
+
+	// Register handlers for cluster notifications
+	client.RegisterPushNotificationHandlerFunc(redis.PushNotificationMoving, func(ctx context.Context, notification []interface{}) bool {
+		fmt.Printf("🚚 Cluster MOVING notification: %v\n", notification)
+		return true
+	})
+
+	// Register handlers for custom application notifications
+	client.RegisterPushNotificationHandlerFunc("METRICS_UPDATE", func(ctx context.Context, notification []interface{}) bool {
+		fmt.Printf("📊 Metrics update: %v\n", notification)
+		return true
+	})
+
+	client.RegisterPushNotificationHandlerFunc("CONFIG_CHANGE", func(ctx context.Context, notification []interface{}) bool {
+		fmt.Printf("⚙️  Configuration change: %v\n", notification)
+		return true
+	})
+
+	fmt.Println("✅ Multiple notification type handlers registered:")
+	fmt.Println("   Redis built-in notifications:")
+	fmt.Printf("   - %s: Pub/Sub messages\n", redis.PushNotificationPubSubMessage)
+	fmt.Printf("   - %s: Keyspace notifications\n", redis.PushNotificationKeyspace)
+	fmt.Printf("   - %s: Key event notifications\n", redis.PushNotificationKeyevent)
+	fmt.Println("   Cluster notifications:")
+	fmt.Printf("   - %s: Cluster slot migration\n", redis.PushNotificationMoving)
+	fmt.Println("   Custom application notifications:")
+	fmt.Println("   - METRICS_UPDATE: Application metrics")
+	fmt.Println("   - CONFIG_CHANGE: Configuration updates")
+}
+
+func demonstrateProcessorAPI() {
+	fmt.Println("\n=== Push Notification Processor API Example ===")
+
+	client := redis.NewClient(&redis.Options{
+		Addr:              "localhost:6379",
+		Protocol:          3,
+		PushNotifications: true,
+	})
+	defer client.Close()
+
+	// Get the push notification processor
+	processor := client.GetPushNotificationProcessor()
+	if processor == nil {
+		log.Println("Push notification processor not available")
+		return
+	}
+
+	fmt.Printf("✅ Push notification processor status: enabled=%v\n", processor.IsEnabled())
+
+	// Get the registry to inspect registered handlers
+	registry := processor.GetRegistry()
+	commands := registry.GetRegisteredCommands()
+	fmt.Printf("📋 Registered commands: %v\n", commands)
+
+	// Register a handler using the processor directly
+	processor.RegisterHandlerFunc("DIRECT_REGISTRATION", func(ctx context.Context, notification []interface{}) bool {
+		fmt.Printf("🎯 Direct registration handler: %v\n", notification)
+		return true
+	})
+
+	// Check if handlers are registered
+	if registry.HasHandlers() {
+		fmt.Println("✅ Push notification handlers are registered and ready")
+	}
+
+	// Demonstrate notification info parsing
+	sampleNotification := []interface{}{"SAMPLE_EVENT", "arg1", "arg2", 123}
+	info := redis.ParsePushNotificationInfo(sampleNotification)
+	if info != nil {
+		fmt.Printf("📄 Notification info - Command: %s, Args: %d\n", info.Command, len(info.Args))
+	}
+}

options.go

@@ -216,6 +216,17 @@ type Options struct {
 	// UnstableResp3 enables Unstable mode for Redis Search module with RESP3.
 	// When unstable mode is enabled, the client will use RESP3 protocol and only be able to use RawResult
 	UnstableResp3 bool
+
+	// PushNotifications enables general push notification processing.
+	// When enabled, the client will process RESP3 push notifications and
+	// route them to registered handlers.
+	//
+	// default: false
+	PushNotifications bool
+
+	// PushNotificationProcessor is the processor for handling push notifications.
+	// If nil, a default processor will be created when PushNotifications is enabled.
+	PushNotificationProcessor *PushNotificationProcessor
 }
 
 func (opt *Options) init() {

pubsub.go

@@ -38,12 +38,21 @@ type PubSub struct {
 	chOnce sync.Once
 	msgCh  *channel
 	allCh  *channel
+
+	// Push notification processor for handling generic push notifications
+	pushProcessor *PushNotificationProcessor
 }
 
 func (c *PubSub) init() {
 	c.exit = make(chan struct{})
 }
 
+// SetPushNotificationProcessor sets the push notification processor for handling
+// generic push notifications received on this PubSub connection.
+func (c *PubSub) SetPushNotificationProcessor(processor *PushNotificationProcessor) {
+	c.pushProcessor = processor
+}
+
 func (c *PubSub) String() string {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -367,6 +376,18 @@ func (p *Pong) String() string {
 	return "Pong"
 }
 
+// PushNotificationMessage represents a generic push notification received on a PubSub connection.
+type PushNotificationMessage struct {
+	// Command is the push notification command (e.g., "MOVING", "CUSTOM_EVENT").
+	Command string
+	// Args are the arguments following the command.
+	Args []interface{}
+}
+
+func (m *PushNotificationMessage) String() string {
+	return fmt.Sprintf("push: %s", m.Command)
+}
+
 func (c *PubSub) newMessage(reply interface{}) (interface{}, error) {
 	switch reply := reply.(type) {
 	case string:
@@ -413,6 +434,18 @@ func (c *PubSub) newMessage(reply interface{}) (interface{}, error) {
 				Payload: reply[1].(string),
 			}, nil
 		default:
+			// Try to handle as generic push notification
+			if c.pushProcessor != nil && c.pushProcessor.IsEnabled() {
+				ctx := c.getContext()
+				handled := c.pushProcessor.GetRegistry().HandleNotification(ctx, reply)
+				if handled {
+					// Return a special message type to indicate it was handled
+					return &PushNotificationMessage{
+						Command: kind,
+						Args:    reply[1:],
+					}, nil
+				}
+			}
 			return nil, fmt.Errorf("redis: unsupported pubsub message: %q", kind)
 		}
 	default:
@@ -658,6 +691,9 @@ func (c *channel) initMsgChan() {
 				// Ignore.
 			case *Pong:
 				// Ignore.
+			case *PushNotificationMessage:
+				// Ignore push notifications in message-only channel
+				// They are already handled by the push notification processor
 			case *Message:
 				timer.Reset(c.chanSendTimeout)
 				select {
@@ -712,7 +748,7 @@ func (c *channel) initAllChan() {
 			switch msg := msg.(type) {
 			case *Pong:
 				// Ignore.
-			case *Subscription, *Message:
+			case *Subscription, *Message, *PushNotificationMessage:
 				timer.Reset(c.chanSendTimeout)
 				select {
 				case c.allCh <- msg: