push.go

package quickredis

import (
	"context"
	"strings"
)

// Push is a RESP3 out-of-band message (the '>' type). It carries the payload
// array plus the raw Value for advanced inspection.
type Push struct {
	// Type is the first element of the push, e.g. "message", "invalidate".
	Type string
	// Values are the remaining elements of the push payload.
	Values []string
	// Raw is the fully parsed push value. Its lifetime depends on the read path
	// that produced it. The owned path (plain command replies, and plain
	// auto-pipelined commands) allocates it and it is safe to retain; the
	// borrowed paths — Client.Borrow and the borrow-routed branch of the
	// auto-pipeline pump — hand over a Value whose Str, Bytes and nested
	// elements alias the connection's reusable read buffer, which a later read
	// on that connection overwrites. How soon differs: the pump resets the
	// buffer and reads the command's own reply at offset 0, so it is overwritten
	// by that reply; under Client.Borrow the reply is appended after the push
	// bytes instead, so the overwrite comes with the next command. A Raw
	// observed on a borrowed path is therefore valid only for the duration of
	// the handler call; copy out of it what must outlive the call. Type and
	// Values are cloned by parsePush and can be retained either way.
	Raw Value
}

// PushHandler receives RESP3 push messages as they arrive on the connection.
// Handlers run on the goroutine that triggered the command; they should be
// fast and must not call the same client (deadlock).
type PushHandler interface {
	HandlePush(ctx context.Context, p Push)
}

// PushFunc adapts a function to PushHandler.
type PushFunc func(ctx context.Context, p Push)

func (f PushFunc) HandlePush(ctx context.Context, p Push) { f(ctx, p) }

func (c *Client) dispatchPush(v Value) {
	p := parsePush(v)
	if p.Type == "" {
		return
	}
	// Snapshot the handlers under the lock, then run them outside it: the
	// reader dispatches on its own goroutine, so it can reach here while the
	// caller is still in OnPush — and a handler may itself call OnPush.
	c.mu.RLock()
	handlers := make([]PushHandler, len(c.push))
	copy(handlers, c.push)
	c.mu.RUnlock()
	for _, h := range handlers {
		h.HandlePush(context.Background(), p)
	}
}

// parsePush converts a push Value into the user-facing Push. Type and Values
// are cloned rather than aliased: on the borrowed read paths (Client.Borrow and
// the auto-pipeline pump) valueAsString returns bytesToString of the
// connection's scratch buffer, which is reset and rewritten by the very next
// read — and the auto-pipeline does exactly that between a dispatched push and
// the command's own reply. Handing that aliasing string to a handler would make
// a retained Push decay into a mix of the push and the following reply. The
// owned path already hands over freshly allocated strings, so cloning there
// costs a small copy and keeps the two paths indistinguishable.
func parsePush(v Value) Push {
	if v.Kind != KindPush || len(v.Arr) == 0 {
		return Push{Raw: v}
	}
	typ, _ := valueAsString(v.Arr[0])
	p := Push{Type: strings.Clone(typ), Raw: v}
	for _, el := range v.Arr[1:] {
		s, err := valueAsString(el)
		if err != nil {
			continue
		}
		p.Values = append(p.Values, strings.Clone(s))
	}
	return p
}

// OnPush registers a push handler. Handlers are invoked serially with each
// other, on the connection's reader goroutine. Registration is safe at any
// time, including while pushes are being dispatched.
func (c *Client) OnPush(h PushHandler) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.push = append(c.push, h)
}

// OnInvalidate registers a handler for client-side-caching invalidation pushes
// (keys that changed). Requires RESP3 and client-side caching enabled.
func (c *Client) OnInvalidate(fn func(ctx context.Context, keys []string)) {
	c.OnPush(PushFunc(func(ctx context.Context, p Push) {
		if p.Type == "invalidate" {
			fn(ctx, p.Values)
		}
	}))
}

// OnPubSub registers a handler for pub/sub message pushes.
func (c *Client) OnPubSub(fn func(ctx context.Context, channel, message string)) {
	c.OnPush(PushFunc(func(ctx context.Context, p Push) {
		if p.Type == "message" && len(p.Values) >= 2 {
			fn(ctx, p.Values[0], p.Values[1])
		}
	}))
}