package push import ( "context" "errors" "github.com/redis/go-redis/v9/internal" "github.com/redis/go-redis/v9/internal/proto" ) // NotificationProcessor defines the interface for push notification processors. type NotificationProcessor interface { // GetHandler returns the handler for a specific push notification name. GetHandler(pushNotificationName string) NotificationHandler // ProcessPendingNotifications checks for and processes any pending push notifications. // To be used when it is known that there are notifications on the socket. // It will try to read from the socket and if it is empty - it may block. ProcessPendingNotifications(ctx context.Context, handlerCtx NotificationHandlerContext, rd *proto.Reader) error // RegisterHandler registers a handler for a specific push notification name. RegisterHandler(pushNotificationName string, handler NotificationHandler, protected bool) error // UnregisterHandler removes a handler for a specific push notification name. UnregisterHandler(pushNotificationName string) error } // Processor handles push notifications with a registry of handlers type Processor struct { registry *Registry } type timeoutError interface { Timeout() bool } func isTimeoutError(err error) bool { var timeoutErr timeoutError return errors.As(err, &timeoutErr) && timeoutErr.Timeout() } // NewProcessor creates a new push notification processor func NewProcessor() *Processor { return &Processor{ registry: NewRegistry(), } } // GetHandler returns the handler for a specific push notification name func (p *Processor) GetHandler(pushNotificationName string) NotificationHandler { return p.registry.GetHandler(pushNotificationName) } // RegisterHandler registers a handler for a specific push notification name func (p *Processor) RegisterHandler(pushNotificationName string, handler NotificationHandler, protected bool) error { return p.registry.RegisterHandler(pushNotificationName, handler, protected) } // UnregisterHandler removes a handler for a specific push notification name func (p *Processor) UnregisterHandler(pushNotificationName string) error { return p.registry.UnregisterHandler(pushNotificationName) } // ProcessPendingNotifications checks for and processes any pending push notifications // This method should be called by the client in WithReader before reading the reply // It will try to read from the socket and if it is empty - it may block. func (p *Processor) ProcessPendingNotifications(ctx context.Context, handlerCtx NotificationHandlerContext, rd *proto.Reader) error { return p.processPendingNotifications(ctx, handlerCtx, rd, false) } // ProcessPendingNotificationsBuffered processes one pending push notification // and then continues only through frames already buffered by that read. It is // used by callers that have already established socket readiness and must not // wait for another frame after draining the current batch. func (p *Processor) ProcessPendingNotificationsBuffered( ctx context.Context, handlerCtx NotificationHandlerContext, rd *proto.Reader, ) error { return p.processPendingNotifications(ctx, handlerCtx, rd, true) } func (p *Processor) processPendingNotifications( ctx context.Context, handlerCtx NotificationHandlerContext, rd *proto.Reader, bufferedContinuation bool, ) error { if rd == nil { return nil } processed := false for !bufferedContinuation || !processed || rd.Buffered() > 0 { // Check if there's data available to read var replyType byte if bufferedContinuation { for { b, err := rd.Peek(1) if err != nil { if isTimeoutError(err) { return nil } return err } replyType = b[0] if replyType != proto.RespAttr { break } // Unlike Peek, DiscardNext consumes bytes. Any error here is // fatal because the reader may be left mid-frame. if err := rd.DiscardNext(); err != nil { return err } } } else { var err error replyType, err = rd.PeekReplyType() if err != nil { // No more data available or error reading. break } } // Only process push notifications (arrays starting with >) if replyType != proto.RespPush { break } // see if we should skip this notification notificationName, err := rd.PeekPushNotificationName() if err != nil { // A buffered probe stops cleanly when no full frame is available yet. if bufferedContinuation && isTimeoutError(err) { return nil } // The frame is a CONFIRMED push (peeked above) but its name cannot be // peeked — too long, or a non-string / malformed name. Fall through to // ReadReply to CONSUME it. Do NOT break: that leaves the push at the // buffer head for the caller's reply read to eat as the command value // (cache poison / one-frame reply shift). A genuine mid-frame read error // surfaces at ReadReply below (fatal → the reply-expected drain retires // the conn); a non-string name is gracefully ignored (the type assertion // below fails, so no handler runs). } else if willHandleNotificationInClient(notificationName) { break } // Surface a ReadReply error (unlike the boundary peek errors above, // which consumed nothing): it happens mid-frame after bytes are // consumed, so the conn is desynced and the CSC drainer must remove it. // Normal reply-read callers log-and-ignore this and let their own read fail. reply, err := rd.ReadReply() if err != nil { internal.Logger.Printf(ctx, "push: error reading push notification: %v", err) return err } processed = true // Convert to slice of interfaces notification, ok := reply.([]interface{}) if !ok { break } // Handle the notification directly if len(notification) > 0 { // Extract the notification type (first element) if notificationType, ok := notification[0].(string); ok { // Get the handler for this notification type if handler := p.registry.GetHandler(notificationType); handler != nil { // Handle the notification err := handler.HandlePushNotification(ctx, handlerCtx, notification) if err != nil { internal.Logger.Printf(ctx, "push: error handling push notification: %v", err) } } } } } return nil } // VoidProcessor discards all push notifications without processing them type VoidProcessor struct{} // NewVoidProcessor creates a new void push notification processor func NewVoidProcessor() *VoidProcessor { return &VoidProcessor{} } // GetHandler returns nil for void processor since it doesn't maintain handlers func (v *VoidProcessor) GetHandler(_ string) NotificationHandler { return nil } // RegisterHandler returns an error for void processor since it doesn't maintain handlers func (v *VoidProcessor) RegisterHandler(pushNotificationName string, _ NotificationHandler, _ bool) error { return ErrVoidProcessorRegister(pushNotificationName) } // UnregisterHandler returns an error for void processor since it doesn't maintain handlers func (v *VoidProcessor) UnregisterHandler(pushNotificationName string) error { return ErrVoidProcessorUnregister(pushNotificationName) } // ProcessPendingNotifications for VoidProcessor does nothing since push notifications // are only available in RESP3 and this processor is used for RESP2 connections. // This avoids unnecessary buffer scanning overhead. // It does however read and discard all push notifications from the buffer to avoid // them being interpreted as a reply. // This method should be called by the client in WithReader before reading the reply // to be sure there are no buffered push notifications. // It will try to read from the socket and if it is empty - it may block. func (v *VoidProcessor) ProcessPendingNotifications(_ context.Context, handlerCtx NotificationHandlerContext, rd *proto.Reader) error { // read and discard all push notifications if rd == nil { return nil } for { // Check if there's data available to read replyType, err := rd.PeekReplyType() if err != nil { // No more data available or error reading // if timeout, it will be handled by the caller break } // Only process push notifications (arrays starting with >) if replyType != proto.RespPush { break } // see if we should skip this notification notificationName, err := rd.PeekPushNotificationName() if err != nil { // Name too long to peek: still consume the frame below so it isn't // misread as a reply. if !errors.Is(err, proto.ErrPushNotificationNameTooLong) { break } } else if willHandleNotificationInClient(notificationName) { break } // Read the push notification _, err = rd.ReadReply() if err != nil { internal.Logger.Printf(context.Background(), "push: error reading push notification: %v", err) return nil } } return nil } // willHandleNotificationInClient checks if a notification type should be ignored by the push notification // processor and handled by other specialized systems instead (pub/sub, streams, keyspace, etc.). func willHandleNotificationInClient(notificationType string) bool { switch notificationType { // Pub/Sub notifications - handled by pub/sub system case "message", // Regular pub/sub message "pmessage", // Pattern pub/sub message "subscribe", // Subscription confirmation "unsubscribe", // Unsubscription confirmation "psubscribe", // Pattern subscription confirmation "punsubscribe", // Pattern unsubscription confirmation "smessage", // Sharded pub/sub message (Redis 7.0+) "ssubscribe", // Sharded subscription confirmation "sunsubscribe": // Sharded unsubscription confirmation return true default: return false } }