mirror of
https://github.com/redis/go-redis.git
synced 2025-07-29 17:41:15 +03:00
- 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
156 lines
3.1 KiB
Go
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
|
|
}
|