1
0
mirror of https://github.com/redis/go-redis.git synced 2025-07-29 17:41:15 +03:00
Files
go-redis/internal/pool/conn.go
Nedyalko Dyakov b6e712b41a feat: add proactive push notification processing to WithReader
- Add push notification processing to Conn.WithReader method
- Process notifications immediately before every read operation
- Provides proactive notification handling vs reactive processing
- Add proper error handling with internal.Logger
- Non-blocking implementation that doesn't break Redis operations
- Complements existing processing in Pool.Put and isHealthyConn

Benefits:
- Immediate processing when notifications arrive
- Called before every read operation for optimal timing
- Prevents notification backlog accumulation
- More responsive to Redis cluster changes
- Better user experience during migrations
- Optimal placement for catching asynchronous notifications

Implementation:
- Type-safe interface assertion for processor
- Context-aware error handling with logging
- Maintains backward compatibility
- Consistent with existing pool patterns
- Three-layer processing strategy: WithReader (proactive) + Pool.Put + isHealthyConn (reactive)

Use cases:
- MOVING/MIGRATING/MIGRATED notifications for slot migrations
- FAILING_OVER/FAILED_OVER notifications for failover scenarios
- Real-time cluster topology change awareness
- Improved connection utilization efficiency
2025-06-27 22:56:04 +03:00

156 lines
3.1 KiB
Go

package pool
import (
"bufio"
"context"
"net"
"sync/atomic"
"time"
"github.com/redis/go-redis/v9/internal"
"github.com/redis/go-redis/v9/internal/proto"
"github.com/redis/go-redis/v9/internal/pushnotif"
)
var noDeadline = time.Time{}
type Conn struct {
usedAt int64 // atomic
netConn net.Conn
rd *proto.Reader
bw *bufio.Writer
wr *proto.Writer
Inited bool
pooled bool
createdAt time.Time
onClose func() error
// Push notification processor for handling push notifications on this connection
// This is set when the connection is created and is a reference to the processor
PushNotificationProcessor pushnotif.ProcessorInterface
}
func NewConn(netConn net.Conn) *Conn {
cn := &Conn{
netConn: netConn,
createdAt: time.Now(),
}
cn.rd = proto.NewReader(netConn)
cn.bw = bufio.NewWriter(netConn)
cn.wr = proto.NewWriter(cn.bw)
cn.SetUsedAt(time.Now())
return cn
}
func (cn *Conn) UsedAt() time.Time {
unix := atomic.LoadInt64(&cn.usedAt)
return time.Unix(unix, 0)
}
func (cn *Conn) SetUsedAt(tm time.Time) {
atomic.StoreInt64(&cn.usedAt, tm.Unix())
}
func (cn *Conn) SetOnClose(fn func() error) {
cn.onClose = fn
}
func (cn *Conn) SetNetConn(netConn net.Conn) {
cn.netConn = netConn
cn.rd.Reset(netConn)
cn.bw.Reset(netConn)
}
func (cn *Conn) Write(b []byte) (int, error) {
return cn.netConn.Write(b)
}
func (cn *Conn) RemoteAddr() net.Addr {
if cn.netConn != nil {
return cn.netConn.RemoteAddr()
}
return nil
}
func (cn *Conn) WithReader(
ctx context.Context, timeout time.Duration, fn func(rd *proto.Reader) error,
) error {
// Process any pending push notifications before executing the read function
// This ensures push notifications are handled as soon as they arrive
if cn.PushNotificationProcessor != nil {
// Type assert to the processor interface
if err := cn.PushNotificationProcessor.ProcessPendingNotifications(ctx, cn.rd); err != nil {
// Log the error but don't fail the read operation
// Push notification processing errors shouldn't break normal Redis operations
internal.Logger.Printf(ctx, "push: error processing pending notifications in WithReader: %v", err)
}
}
if timeout >= 0 {
if err := cn.netConn.SetReadDeadline(cn.deadline(ctx, timeout)); err != nil {
return err
}
}
return fn(cn.rd)
}
func (cn *Conn) WithWriter(
ctx context.Context, timeout time.Duration, fn func(wr *proto.Writer) error,
) error {
if timeout >= 0 {
if err := cn.netConn.SetWriteDeadline(cn.deadline(ctx, timeout)); err != nil {
return err
}
}
if cn.bw.Buffered() > 0 {
cn.bw.Reset(cn.netConn)
}
if err := fn(cn.wr); err != nil {
return err
}
return cn.bw.Flush()
}
func (cn *Conn) Close() error {
if cn.onClose != nil {
// ignore error
_ = cn.onClose()
}
return cn.netConn.Close()
}
func (cn *Conn) deadline(ctx context.Context, timeout time.Duration) time.Time {
tm := time.Now()
cn.SetUsedAt(tm)
if timeout > 0 {
tm = tm.Add(timeout)
}
if ctx != nil {
deadline, ok := ctx.Deadline()
if ok {
if timeout == 0 {
return deadline
}
if deadline.Before(tm) {
return deadline
}
return tm
}
}
if timeout > 0 {
return tm
}
return noDeadline
}