redis

package module
v1.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 6, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Overview

Package redis provides isolated go-redis integration and ownership-token locks.

Index

Constants

View Source 
const KeepTTL time.Duration = r.KeepTTL

KeepTTL preserves an existing key lifetime on Set, following Redis KEEPTTL semantics.

Variables

View Source 
var ErrLockBusy = errors.New("redis: lock busy")
View Source 
var ErrNotFound = r.Nil
View Source 
var ErrOwnership = errors.New("redis: lock ownership lost")

Functions

func Classify 

func Classify(e error) error

Classify maps Redis misses and optimistic transaction conflicts while preserving errors.Is behavior.

Types

type Client 

type Client struct {
	// contains filtered or unexported fields
}

Client owns its configured infrastructure client and supports concurrent operations.

func New 

func New(ctx context.Context, c Config) (*Client, error)

New constructs and validates an explicitly owned instance. See package documentation for defaults and cleanup.

func (*Client) Acquire 

func (c *Client) Acquire(ctx context.Context, key string, ttl time.Duration) (*Lock, error)

Acquire attempts to acquire an expiring lock with a cryptographic ownership token.

func (*Client) CacheAside 

func (c *Client) CacheAside(ctx context.Context, key string, dst any, ttl time.Duration, load func(context.Context) (any, error)) error

CacheAside may run load concurrently for misses. Use external coordination for expensive duplicate loads.

func (*Client) Close 

func (c *Client) Close() error

Close releases owned resources. See package documentation for shutdown ordering.

func (*Client) ConsumeGroup 

func (c *Client) ConsumeGroup(ctx context.Context, cfg StreamConsumerConfig, handler func(context.Context, StreamMessage) error) error

ConsumeGroup reads new group messages serially and acknowledges only after successful processing. A handler failure/panic leaves the entry pending; callers choose retry or reclaim policy explicitly.

func (*Client) Decrement 

func (c *Client) Decrement(ctx context.Context, key string) (int64, error)

Decrement atomically decrements a key.

func (*Client) Delete 

func (c *Client) Delete(ctx context.Context, keys ...string) (int64, error)

Delete deletes the specified keys.

func (*Client) Eval 

func (c *Client) Eval(ctx context.Context, script string, keys []string, args ...any) (any, error)

Eval executes Lua with namespaced keys and bound arguments.

func (*Client) EvalSHA 

func (c *Client) EvalSHA(ctx context.Context, sha string, keys []string, args ...any) (any, error)

EvalSHA executes a cached Lua script with namespaced keys.

func (*Client) Exists 

func (c *Client) Exists(ctx context.Context, keys ...string) (int64, error)

Exists counts existing keys.

func (*Client) Expire 

func (c *Client) Expire(ctx context.Context, key string, ttl time.Duration) (bool, error)

Expire updates key expiration.

func (*Client) Get 

func (c *Client) Get(ctx context.Context, key string) (string, error)

Get reads one value into the requested representation.

func (*Client) GetAndDelete 

func (c *Client) GetAndDelete(ctx context.Context, key string) (string, error)

GetAndDelete atomically reads and removes a key.

func (*Client) GetBytes 

func (c *Client) GetBytes(ctx context.Context, key string) ([]byte, error)

GetBytes returns the stored value as bytes, or ErrNotFound.

func (*Client) GetJSON 

func (c *Client) GetJSON(ctx context.Context, key string, dst any) error

GetJSON reads and deserializes a stored value.

func (*Client) HDel 

func (c *Client) HDel(ctx context.Context, key string, fields ...string) (int64, error)

HDel removes hash fields.

func (*Client) HExists 

func (c *Client) HExists(ctx context.Context, key, field string) (bool, error)

HExists checks whether a hash field exists.

func (*Client) HGet 

func (c *Client) HGet(ctx context.Context, key, field string) (string, error)

HGet reads a hash field.

func (*Client) HGetAll 

func (c *Client) HGetAll(ctx context.Context, key string) (map[string]string, error)

HGetAll reads all hash fields.

func (*Client) HIncrBy 

func (c *Client) HIncrBy(ctx context.Context, key, field string, n int64) (int64, error)

HIncrBy atomically increments a hash field.

func (*Client) HSet 

func (c *Client) HSet(ctx context.Context, key string, values ...any) (int64, error)

HSet sets hash fields.

func (*Client) Increment 

func (c *Client) Increment(ctx context.Context, key string) (int64, error)

Increment atomically increments a key.

func (*Client) IncrementBy 

func (c *Client) IncrementBy(ctx context.Context, key string, n int64) (int64, error)

IncrementBy atomically adds the requested integer.

func (*Client) LPop 

func (c *Client) LPop(ctx context.Context, key string) (string, error)

LPop removes and returns the first list value.

func (*Client) LPush 

func (c *Client) LPush(ctx context.Context, key string, values ...any) (int64, error)

LPush prepends list values.

func (*Client) LRange 

func (c *Client) LRange(ctx context.Context, key string, start, stop int64) ([]string, error)

LRange reads an inclusive list range.

func (*Client) LTrim 

func (c *Client) LTrim(ctx context.Context, key string, start, stop int64) error

LTrim retains an inclusive list range.

func (*Client) Native 

func (c *Client) Native() r.UniversalClient

Native provides an explicit advanced driver escape hatch; preserve SDK ownership rules.

func (*Client) Ping 

func (c *Client) Ping(ctx context.Context) error

Ping checks connectivity using the supplied context.

func (*Client) Pipeline 

func (c *Client) Pipeline(ctx context.Context, fn func(r.Pipeliner) error) error

Pipeline is an advanced native callback. Keys inside the callback are not automatically namespaced.

func (*Client) Publish 

func (c *Client) Publish(ctx context.Context, channel string, value any) (int64, error)

Publish publishes with the configured acknowledgment policy.

func (*Client) RPop 

func (c *Client) RPop(ctx context.Context, key string) (string, error)

RPop removes and returns the last list value.

func (*Client) RPush 

func (c *Client) RPush(ctx context.Context, key string, values ...any) (int64, error)

RPush appends list values.

func (*Client) RateLimit 

func (c *Client) RateLimit(ctx context.Context, key string, limit int64, window time.Duration) (allowed bool, remaining int64, retry time.Duration, err error)

RateLimit is an atomic fixed-window counter. A key must describe one policy/window.

func (*Client) SAdd 

func (c *Client) SAdd(ctx context.Context, key string, members ...any) (int64, error)

SAdd adds set members.

func (*Client) SCard 

func (c *Client) SCard(ctx context.Context, key string) (int64, error)

SCard returns set cardinality.

func (*Client) SIsMember 

func (c *Client) SIsMember(ctx context.Context, key string, member any) (bool, error)

SIsMember tests set membership.

func (*Client) SMembers 

func (c *Client) SMembers(ctx context.Context, key string) ([]string, error)

SMembers returns set members.

func (*Client) SRem 

func (c *Client) SRem(ctx context.Context, key string, members ...any) (int64, error)

SRem removes set members.

func (*Client) ScriptLoad 

func (c *Client) ScriptLoad(ctx context.Context, script string) (string, error)

ScriptLoad loads a script into the Redis script cache.

func (*Client) Set 

func (c *Client) Set(ctx context.Context, key string, value any, ttl time.Duration) error

Set stores a value with the requested expiration.

func (*Client) SetJSON 

func (c *Client) SetJSON(ctx context.Context, key string, value any, ttl time.Duration) error

SetJSON serializes and stores a value with expiration.

func (*Client) SetNX 

func (c *Client) SetNX(ctx context.Context, key string, value any, ttl time.Duration) (bool, error)

SetNX sets a value only when its key is absent.

func (*Client) Stats 

func (c *Client) Stats() *PoolStats

Stats returns a concurrency-safe statistics snapshot.

func (*Client) Subscribe 

func (c *Client) Subscribe(ctx context.Context, patterns bool, handler func(context.Context, Message) error, channels ...string) error

Subscribe owns its subscription until ctx cancellation or handler error. It waits for the subscription acknowledgment.

func (*Client) TTL 

func (c *Client) TTL(ctx context.Context, key string) (time.Duration, error)

TTL returns remaining key lifetime using Redis duration semantics.

func (*Client) Transaction 

func (c *Client) Transaction(ctx context.Context, fn func(r.Pipeliner) error) error

Transaction executes an advanced native MULTI/EXEC pipeline callback.

func (*Client) Watch 

func (c *Client) Watch(ctx context.Context, fn func(*r.Tx) error, keys ...string) error

Watch executes an advanced native optimistic transaction callback.

func (*Client) XAck 

func (c *Client) XAck(ctx context.Context, key, group string, ids ...string) (int64, error)

XAck acknowledges stream entries for a consumer group.

func (*Client) XAdd 

func (c *Client) XAdd(ctx context.Context, key string, values map[string]any) (string, error)

XAdd appends a stream entry.

func (*Client) XAutoClaim 

func (c *Client) XAutoClaim(ctx context.Context, key, group, consumer, start string, idle time.Duration, count int64) ([]StreamMessage, string, error)

XAutoClaim claims idle pending entries and returns the next scan position.

func (*Client) XClaim 

func (c *Client) XClaim(ctx context.Context, key, group, consumer string, idle time.Duration, ids ...string) ([]StreamMessage, error)

XClaim transfers ownership of sufficiently idle pending entries.

func (*Client) XDelete 

func (c *Client) XDelete(ctx context.Context, key string, ids ...string) (int64, error)

XDelete deletes stream entries.

func (*Client) XGroupCreate 

func (c *Client) XGroupCreate(ctx context.Context, key, group, start string) error

XGroupCreate creates a stream consumer group, creating the stream if needed.

func (*Client) XGroupDestroy 

func (c *Client) XGroupDestroy(ctx context.Context, key, group string) (int64, error)

XGroupDestroy removes a stream consumer group.

func (*Client) XPending 

func (c *Client) XPending(ctx context.Context, key, group string) (*Pending, error)

XPending summarizes pending consumer-group entries.

func (*Client) XRead 

func (c *Client) XRead(ctx context.Context, key, lastID string, count int64, block time.Duration) ([]Stream, error)

XRead reads a stream from the supplied entry ID.

func (*Client) XReadGroup 

func (c *Client) XReadGroup(ctx context.Context, key, group, consumer, lastID string, count int64, block time.Duration) ([]Stream, error)

XReadGroup reads stream entries for a consumer group.

func (*Client) ZAdd 

func (c *Client) ZAdd(ctx context.Context, key string, members ...ScoredMember) (int64, error)

ZAdd adds scored sorted-set members.

func (*Client) ZCard 

func (c *Client) ZCard(ctx context.Context, key string) (int64, error)

ZCard returns sorted-set cardinality.

func (*Client) ZRange 

func (c *Client) ZRange(ctx context.Context, key string, start, stop int64) ([]string, error)

ZRange reads a sorted-set index range.

func (*Client) ZRangeByScore 

func (c *Client) ZRangeByScore(ctx context.Context, key string, min, max string) ([]string, error)

ZRangeByScore reads a sorted-set score range.

func (*Client) ZRem 

func (c *Client) ZRem(ctx context.Context, key string, members ...any) (int64, error)

ZRem removes sorted-set members.

func (*Client) ZScore 

func (c *Client) ZScore(ctx context.Context, key, member string) (float64, error)

ZScore reads a member score.

type Config 

type Config struct {
	Mode                                   Mode
	SentinelUsername, SentinelPassword     string
	AllowPlaintext                         bool
	Observer                               instrument.Hook
	Addresses                              []string
	MasterName, Username, Password, Prefix string
	DB                                     int
	// PoolSize is the hard command-connection bound per Redis node. Zero uses
	// 10*GOMAXPROCS, matching the native base size. Pipelines share this pool;
	// dedicated subscriptions are separately owned and must be closed by callers.
	PoolSize                               int
	MinIdleConnections, MaxRetries         int
	DialTimeout, ReadTimeout, WriteTimeout time.Duration
	TLS                                    *tls.Config
	Serializer                             encodingx.Serializer
}

Config contains explicit constructor settings; see package documentation for defaults.

type IdempotencyStore 

type IdempotencyStore struct{ Client *Client }

IdempotencyStore stores only fingerprints, ownership tokens and response status, never response bodies.

func (IdempotencyStore) Begin 

func (s IdempotencyStore) Begin(ctx context.Context, key, fingerprint string, lease time.Duration) (idempotency.Record, error)

Begin atomically attempts to acquire a fingerprint-bound processing lease.

func (IdempotencyStore) Complete 

func (s IdempotencyStore) Complete(ctx context.Context, key, token string, status int, ttl time.Duration) error

Complete records successful processing only for the current ownership token.

func (IdempotencyStore) Fail 

func (s IdempotencyStore) Fail(ctx context.Context, key, token string, ttl time.Duration) error

Fail records failed processing only for the current ownership token.

func (IdempotencyStore) Renew 

func (s IdempotencyStore) Renew(ctx context.Context, key, token string, ttl time.Duration) error

Renew extends a current processing lease.

type Limiter 

type Limiter struct {
	Client *Client
	Limit  int64
	Window time.Duration
}

Limiter uses atomic Redis fixed windows, unlike the in-memory token bucket.

func (Limiter) Allow 

func (l Limiter) Allow(ctx context.Context, key string) (ratelimit.Decision, error)

Allow evaluates the configured rate policy for a key.

type Lock 

type Lock struct {
	// contains filtered or unexported fields
}

Lock identifies a time-limited Redis ownership lease, without fencing guarantees.

func (*Lock) Owned 

func (l *Lock) Owned(ctx context.Context) (bool, error)

Owned checks whether this lease still owns the stored token.

func (*Lock) Release 

func (l *Lock) Release(ctx context.Context) error

Release atomically deletes a lock only if its ownership token matches.

func (*Lock) Renew 

func (l *Lock) Renew(ctx context.Context, ttl time.Duration) error

Renew extends a current processing lease.

type Message 

type Message struct{ Channel, Pattern, Payload string }

Message contains transport-independent message data.

type Mode 

type Mode string

Mode explicitly selects a universal-client topology, including cluster discovery from a single seed. 

const (
	Standalone Mode = "standalone"
	Sentinel   Mode = "sentinel"
	Cluster    Mode = "cluster"
)

type Pending 

type Pending = r.XPending

Pending summarizes pending stream deliveries.

type PoolStats 

type PoolStats = r.PoolStats

PoolStats contains the Redis connection pool statistics.

type ScoredMember 

type ScoredMember = r.Z

ScoredMember describes a sorted-set member and score.

type Stream 

type Stream = r.XStream

Stream contains Redis stream messages.

type StreamConsumerConfig 

type StreamConsumerConfig struct {
	Stream, Group, Consumer string
	Count                   int64
	Block                   time.Duration
}

StreamConsumerConfig names an existing consumer group. Pending-entry reclamation is explicitly managed with XAutoClaim.

type StreamMessage 

type StreamMessage = r.XMessage

StreamMessage contains one Redis stream entry.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.