redis

package
v0.3.2 Latest Latest
Warning

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

 Go to latest Published: Oct 31, 2026 License: MIT Imports: 45 Imported by: 0

Documentation

Overview

Package redis owns an explicitly configured Redis connection pool. Feature packages consume its focused capabilities; application code uses typed handles.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Module 

func Module(name foundation.ProviderID, key foundation.Key[*Client], config Config) foundation.Module

Module prepares a distinct client for each application and starts it at boot. Dependents resolve the typed key during construction; cleanup is installed before the connectivity check and participates in reverse-order shutdown.

Types

type Client 

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

Client owns its connections and bounded in-flight operations. A cache.Store borrows this client; it does not close it. Use Module for application ownership.

The per-operation path uses atomics and a FIFO admission semaphore only; the mutex guards lifecycle transitions and the subscription registry.

func Open 

func Open(ctx context.Context, config Config) (*Client, error)

Open prepares and starts a client, closing all owned resources on startup error.

func Prepare 

func Prepare(config Config) (*Client, error)

Prepare snapshots validated configuration without I/O or background goroutines.

func (*Client) Add 

func (c *Client) Add(ctx context.Context, key cache.EntryKey, data []byte, ttl cache.TTL) (bool, error)

func (*Client) AddSnapshot 

func (c *Client) AddSnapshot(ctx context.Context, base cache.EntryKey, tags []cache.EntryKey, data []byte, ttl cache.TTL) (bool, error)

AddSnapshot fills an absent or invalidated entry under the current snapshot.

func (*Client) AddTagged 

func (c *Client) AddTagged(ctx context.Context, key cache.TaggedKey, data []byte, ttl cache.TTL) (bool, error)

AddTagged atomically fills an absent or invalidated entry under this snapshot. An existing current value retains its bytes and expiry.

func (*Client) AdvanceScheduleCursor 

func (c *Client) AdvanceScheduleCursor(ctx context.Context, key lease.Key, at time.Time, ttl time.Duration) error

AdvanceScheduleCursor raises one persisted cursor and refreshes its expiry.

func (*Client) ClearRateLimit 

func (c *Client) ClearRateLimit(ctx context.Context, key ratelimit.Key) (bool, error)

ClearRateLimit deletes the key's bucket, including unreadable state at its exact address. A lost reply leaves the deletion unconfirmed; it is not retried.

func (*Client) Close 

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

Close rejects new work and drains owned operations before closing connections. Caller cancellation only stops waiting; cleanup continues. In-flight commands retain their own bounded contexts. Close is safe to repeat concurrently.

func (*Client) DataCount 

func (c *Client) DataCount(ctx context.Context, key data.Key, l data.Limits) (uint64, error)

func (*Client) DataDeleteMany 

func (c *Client) DataDeleteMany(ctx context.Context, keys []data.Key) (uint64, error)

func (*Client) DataExists 

func (c *Client) DataExists(ctx context.Context, key data.Key) (bool, error)

func (*Client) DataExpire 

func (c *Client) DataExpire(ctx context.Context, key data.Key, ttl cache.TTL) (bool, error)

func (*Client) Done 

func (c *Client) Done() <-chan struct{}

Done closes after all operations have drained and connections have closed. The zero client has no lifecycle and returns nil.

func (*Client) ExecuteRaw 

func (c *Client) ExecuteRaw(ctx context.Context, request raw.Request, limits raw.Limits) (raw.Reply, error)

ExecuteRaw performs one attempt. Missing Redis replies remain explicit nil values for a nullable decoder; server and transport errors never become misses.

func (*Client) ExecuteRawBatch 

func (c *Client) ExecuteRawBatch(ctx context.Context, requests []raw.Request, mode raw.Mode, limits raw.Limits) ([]raw.Reply, error)

func (*Client) Exists 

func (c *Client) Exists(ctx context.Context, key cache.EntryKey) (bool, error)

Exists checks the stored type and byte bound without transferring the value.

func (*Client) ExistsTagged 

func (c *Client) ExistsTagged(ctx context.Context, key cache.TaggedKey) (bool, error)

func (*Client) Expire 

func (c *Client) Expire(ctx context.Context, key cache.EntryKey, ttl cache.TTL) (bool, error)

Expire changes a live entry's expiry without replacing its bytes. Existing persistent entries return true when Forever is requested again.

func (*Client) ExpireRaw 

func (c *Client) ExpireRaw(ctx context.Context, key raw.Key, ttl cache.TTL) (bool, error)

func (*Client) ExpireSnapshot 

func (c *Client) ExpireSnapshot(ctx context.Context, base cache.EntryKey, tags []cache.EntryKey, ttl cache.TTL) (bool, error)

ExpireSnapshot changes the expiry of the current snapshot's entry.

func (*Client) ExpireTagged 

func (c *Client) ExpireTagged(ctx context.Context, key cache.TaggedKey, ttl cache.TTL) (bool, error)

func (*Client) Forget 

func (c *Client) Forget(ctx context.Context, key cache.EntryKey) (bool, error)

func (*Client) ForgetMany 

func (c *Client) ForgetMany(ctx context.Context, keys []cache.EntryKey) (uint64, error)

ForgetMany atomically deletes a bounded canonical batch after validating every stored entry. A lost reply is an uncertain outcome, never a reason to retry.

func (*Client) ForgetManyTagged 

func (c *Client) ForgetManyTagged(ctx context.Context, keys []cache.TaggedKey) (uint64, error)

ForgetManyTagged checks one shared tag snapshot and every selected payload before one exact DEL. Stale physical entries are excluded from the live count.

func (*Client) ForgetSnapshot 

func (c *Client) ForgetSnapshot(ctx context.Context, base cache.EntryKey, tags []cache.EntryKey) (bool, error)

ForgetSnapshot removes the entry of the current snapshot; obsolete payloads are reclaimed but not counted.

func (*Client) ForgetTagged 

func (c *Client) ForgetTagged(ctx context.Context, key cache.TaggedKey) (bool, error)

ForgetTagged conditionally removes this tagged entry. A stale caller cannot remove a replacement written under a newer snapshot.

func (*Client) FoundryAdapter 

func (*Client) FoundryAdapter(frameworkadapter.Seal)

FoundryAdapter marks the client as framework-owned adapter I/O.

func (*Client) Get 

func (c *Client) Get(ctx context.Context, key cache.EntryKey) ([]byte, bool, error)

Get returns an owned value, distinguishing a present empty value from a miss.

func (*Client) GetMany 

func (c *Client) GetMany(ctx context.Context, keys []cache.EntryKey) ([]cache.BatchValue, error)

GetMany reads a bounded canonical batch in one script; results follow input order. Missing, expired, over-bound and wrong-type entries are misses.

func (*Client) GetManyTagged 

func (c *Client) GetManyTagged(ctx context.Context, keys []cache.TaggedKey) ([]cache.BatchValue, error)

GetManyTagged checks one shared snapshot and reads every payload in the same script. Obsolete payloads are misses and are reclaimed.

func (*Client) GetTagged 

func (c *Client) GetTagged(ctx context.Context, key cache.TaggedKey) ([]byte, bool, error)

GetTagged returns owned bytes under a current tag snapshot. A changed/missing snapshot returns Conflict; an older stored payload is removed conditionally.

func (*Client) HashDelete 

func (c *Client) HashDelete(ctx context.Context, key data.Key, field string, l data.Limits) (bool, error)

func (*Client) HashGet 

func (c *Client) HashGet(ctx context.Context, key data.Key, field string, l data.Limits) (string, bool, error)

func (*Client) HashGetAll 

func (c *Client) HashGetAll(ctx context.Context, key data.Key, l data.Limits) ([]data.StoredField, error)

HashGetAll returns every field sorted by field bytes, after bounding cardinality, field sizes and the value reply budget on the server. The script returns fields unordered (Lua comparison follows the server locale); the adapter sorts them bytewise and rejects duplicates.

func (*Client) HashGetMany 

func (c *Client) HashGetMany(ctx context.Context, key data.Key, fields []string, l data.Limits) ([]value.Optional[string], error)

HashGetMany reads the requested fields atomically after checking every stored size against the value and reply bounds.

func (*Client) HashIncrement 

func (c *Client) HashIncrement(ctx context.Context, key data.Key, field string, delta int64, l data.Limits) (int64, error)

HashIncrement uses Redis signed 64-bit arithmetic on a canonical integer field.

func (*Client) HashSet 

func (c *Client) HashSet(ctx context.Context, key data.Key, field, value string, l data.Limits) (bool, error)

func (*Client) Increment 

func (c *Client) Increment(ctx context.Context, key cache.EntryKey, delta int64, initialTTL cache.TTL) (int64, error)

Increment uses server integer arithmetic and retains an existing entry's TTL. Missing entries begin at zero and receive initialTTL. No mutation is retried.

func (*Client) IncrementSnapshot 

func (c *Client) IncrementSnapshot(ctx context.Context, base cache.EntryKey, tags []cache.EntryKey, delta int64, ttl cache.TTL) (int64, error)

IncrementSnapshot applies exact signed arithmetic under the current snapshot.

func (*Client) IncrementTagged 

func (c *Client) IncrementTagged(ctx context.Context, key cache.TaggedKey, delta int64, ttl cache.TTL) (int64, error)

IncrementTagged combines snapshot validation with exact signed arithmetic. Existing expiry is retained; an absent/invalidated counter starts at zero.

func (*Client) InvalidateTags 

func (c *Client) InvalidateTags(ctx context.Context, keys []cache.EntryKey) error

InvalidateTags atomically rotates the entire selected set. An uncertain remote error may hide an applied rotation; callers must not infer rollback or retry.

func (*Client) LeaseAcquire 

func (c *Client) LeaseAcquire(ctx context.Context, key lease.Key, owner lease.Owner, ttl time.Duration) (bool, error)

LeaseAcquire attempts once against this Redis authority. No mutation is retried.

func (*Client) LeaseForceRelease 

func (c *Client) LeaseForceRelease(ctx context.Context, key lease.Key) (bool, error)

LeaseForceRelease administratively deletes one exact lease address whatever its owner, type or expiry, repairing metadata that ordinary operations reject as corrupt (such as a lease key without TTL). It is never used implicitly.

func (*Client) LeaseRelease 

func (c *Client) LeaseRelease(ctx context.Context, key lease.Key, owner lease.Owner) (bool, error)

LeaseRelease deletes only the matching live owner. Expiry/replacement returns false.

func (*Client) LeaseRenew 

func (c *Client) LeaseRenew(ctx context.Context, key lease.Key, owner lease.Owner, ttl time.Duration) (bool, error)

LeaseRenew extends only the matching live owner; it cannot recreate an expired key.

func (*Client) LeaseTransfer 

func (c *Client) LeaseTransfer(ctx context.Context, key lease.Key, from, to lease.Owner, ttl time.Duration) (bool, error)

LeaseTransfer atomically replaces the live owner from with to and renews the lease; a different, expired or absent owner returns false without change.

func (*Client) ListPop 

func (c *Client) ListPop(ctx context.Context, key data.Key, end data.End, l data.Limits) (string, bool, error)

ListPop removes one element after checking its size.

func (*Client) ListPush 

func (c *Client) ListPush(ctx context.Context, key data.Key, values []string, end data.End, l data.Limits) (uint64, error)

ListPush inserts every value in one command after checking sizes and the resulting length, and returns the new length.

func (*Client) ListRange 

func (c *Client) ListRange(ctx context.Context, key data.Key, start, stop int64, l data.Limits) ([]string, error)

ListRange returns elements start..stop inclusive within the reply bounds.

func (*Client) ListTrim 

func (c *Client) ListTrim(ctx context.Context, key data.Key, start, stop int64, l data.Limits) error

ListTrim keeps elements start..stop inclusive (Redis LTRIM semantics).

func (*Client) LockoutBegin 

func (c *Client) LockoutBegin(ctx context.Context, key lockout.Key, policy lockout.Policy, candidate lockout.Generation) (lockout.Admission, error)

LockoutBegin atomically checks the shared failure window using Redis TIME. This client owns no process-local account state and never retries a lost reply.

func (*Client) LockoutFinish 

func (c *Client) LockoutFinish(ctx context.Context, key lockout.Key, policy lockout.Policy, snapshot lockout.Snapshot, outcome lockout.Outcome) (lockout.Decision, error)

func (*Client) LockoutReset 

func (c *Client) LockoutReset(ctx context.Context, key lockout.Key, policy lockout.Policy) (bool, error)

func (*Client) PeekRateLimit 

func (c *Client) PeekRateLimit(ctx context.Context, key ratelimit.Key, limit ratelimit.Limit, cost uint32) (ratelimit.Decision, error)

PeekRateLimit runs the same atomic script read-only: it reports the decision cost would receive now without consuming capacity or changing expiry.

func (*Client) Ping 

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

Ping verifies the running client under the same operation bounds as features.

func (*Client) Publish 

func (c *Client) Publish(ctx context.Context, channel pubsub.Channel, data []byte) (uint64, error)

Publish attempts one ephemeral publication. The count is server subscriptions, not delivery/handler completion. Uncertain publications are never retried.

func (*Client) Put 

func (c *Client) Put(ctx context.Context, key cache.EntryKey, data []byte, ttl cache.TTL) error

func (*Client) PutLeased 

func (c *Client) PutLeased(ctx context.Context, key cache.EntryKey, data []byte, ttl cache.TTL, proof lease.Proof) error

PutLeased atomically checks this fill's current lease and publishes the value. Old owners cannot write after a successor has acquired the same fill address.

func (*Client) PutSnapshot 

func (c *Client) PutSnapshot(ctx context.Context, base cache.EntryKey, tags []cache.EntryKey, data []byte, ttl cache.TTL) error

PutSnapshot resolves tag metadata (creating fresh versions for missing metadata) and writes under the resolved snapshot in one atomic script.

func (*Client) PutTagged 

func (c *Client) PutTagged(ctx context.Context, key cache.TaggedKey, data []byte, ttl cache.TTL) error

PutTagged validates the snapshot and writes an owned value with explicit expiry. Corruption, stale snapshots and denied mutation permissions preserve live data.

func (*Client) PutTaggedLeased 

func (c *Client) PutTaggedLeased(ctx context.Context, key cache.TaggedKey, data []byte, ttl cache.TTL, proof lease.Proof) error

PutTaggedLeased checks the lease and tag snapshot in the same atomic operation. It reuses tagged payload/expiry validation and permission preflight.

func (*Client) RateLimit 

func (c *Client) RateLimit(ctx context.Context, key ratelimit.Key, limit ratelimit.Limit, cost uint32) (ratelimit.Decision, error)

RateLimit makes one atomic decision using Redis TIME and the key's phased fixed windows. A live bucket with a different policy is converted, keeping its admitted usage; corrupt metadata fails without mutation. Lost replies never trigger retry; their capacity consumption remains unknown to the caller.

func (*Client) ReadSnapshot 

func (c *Client) ReadSnapshot(ctx context.Context, base cache.EntryKey, tags []cache.EntryKey, payload bool) (cache.TaggedKey, []byte, bool, error)

ReadSnapshot resolves tag/namespace metadata (creating fresh versions for missing metadata) and reads the tagged entry in one atomic script, so every typed read costs one round trip. Obsolete, over-bound or malformed payloads are misses and are reclaimed. The reply never includes a stale value.

func (*Client) ResolveTags 

func (c *Client) ResolveTags(ctx context.Context, keys []cache.EntryKey) ([]cache.TagVersion, error)

ResolveTags atomically creates fresh versions for missing metadata and returns the current versions in canonical input order. Redis tags require Redis 7+.

func (*Client) ScheduleCursor 

func (c *Client) ScheduleCursor(ctx context.Context, key lease.Key) (time.Time, bool, error)

ScheduleCursor reads one persisted catch-up cursor.

func (*Client) SetAdd 

func (c *Client) SetAdd(ctx context.Context, key data.Key, member string, l data.Limits) (bool, error)

func (*Client) SetContains 

func (c *Client) SetContains(ctx context.Context, key data.Key, member string, l data.Limits) (bool, error)

func (*Client) SetMembers 

func (c *Client) SetMembers(ctx context.Context, key data.Key, l data.Limits) ([]string, error)

func (*Client) SetRemove 

func (c *Client) SetRemove(ctx context.Context, key data.Key, member string, l data.Limits) (bool, error)

func (*Client) SortedSetAdd 

func (c *Client) SortedSetAdd(ctx context.Context, key data.Key, member string, score float64, l data.Limits) (bool, error)

SortedSetAdd inserts or re-scores one member, rejecting growth when full.

func (*Client) SortedSetCountByScore 

func (c *Client) SortedSetCountByScore(ctx context.Context, key data.Key, scores data.ScoreRange, l data.Limits) (uint64, error)

func (*Client) SortedSetIncrement 

func (c *Client) SortedSetIncrement(ctx context.Context, key data.Key, member string, delta float64, l data.Limits) (float64, error)

SortedSetIncrement adds delta to one member's score (missing starts at zero).

func (*Client) SortedSetRange 

func (c *Client) SortedSetRange(ctx context.Context, key data.Key, window data.Window, l data.Limits) ([]data.StoredMember, error)

SortedSetRange returns a rank window with scores.

func (*Client) SortedSetRangeByScore 

func (c *Client) SortedSetRangeByScore(ctx context.Context, key data.Key, scores data.ScoreRange, window data.Window, l data.Limits) ([]data.StoredMember, error)

SortedSetRangeByScore returns a score range, windowed with Redis LIMIT.

func (*Client) SortedSetRank 

func (c *Client) SortedSetRank(ctx context.Context, key data.Key, member string, order data.Order, l data.Limits) (uint64, bool, error)

func (*Client) SortedSetRemove 

func (c *Client) SortedSetRemove(ctx context.Context, key data.Key, member string, l data.Limits) (bool, error)

func (*Client) SortedSetScore 

func (c *Client) SortedSetScore(ctx context.Context, key data.Key, member string, l data.Limits) (float64, bool, error)

func (*Client) Start 

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

Start checks connectivity once. The first caller owns that attempt; concurrent callers can cancel their wait. A failed start is terminal for this instance.

func (*Client) Stats 

func (c *Client) Stats() Stats

func (*Client) Subscribe 

func (c *Client) Subscribe(ctx context.Context, channels []pubsub.Channel, limits pubsub.Limits) (pubsub.Stream, error)

Subscribe confirms every exact channel before returning. All subscriptions of one Client share one dedicated connection with one reader and one heartbeat; a channel already confirmed on that connection is ready immediately, and a new channel waits for its own SUBSCRIBE acknowledgement. Establishment context cancellation never leaves an unowned reader. Any connection failure (receive error, heartbeat loss or unexpected protocol frame) terminates EVERY stream on that connection with pubsub.ErrDisconnected; vendor reconnect attempts never imply continuous delivery or cause automatic public resubscription.

type Config 

type Config struct {
	Host           string
	Port           uint16
	Database       int
	User           string
	Password       secret.String
	TLS            TLSMode
	TLSConfig      *tls.Config
	MaxConnections int
	MaxOperations  int
	// MaxSubscriptions separately bounds dedicated pub/sub connections.
	MaxSubscriptions int
	ConnectTimeout   time.Duration
	OperationTimeout time.Duration
	PoolTimeout      time.Duration
	MaxIdleTime      time.Duration
	MaxLifetime      time.Duration
	// MaxValueBytes bounds cache writes and server-side reads before GET returns
	// the value. It is not a bound on an untrusted server's RESP parser allocations.
	MaxValueBytes int
}

Config addresses one standalone Redis server. It performs no environment or credential-file discovery. Cluster and Sentinel routing are not implemented. TLS certificate/key material must remain immutable after Prepare.

func DefaultConfig 

func DefaultConfig() Config

DefaultConfig sizes the pool like go-redis's own default (10 connections per CPU) on a typical 4–8 core host, with a fixed value so configuration validation stays deterministic across machines. MaxOperations bounds admitted commands, including those waiting for a pooled connection; a burst beyond it queues briefly and then reports fault.Overloaded.

func (Config) Validate 

func (c Config) Validate() error

type Connection 

type Connection struct {
	Name  ConnectionName
	Value *Client
}

Connection binds a typed name to an existing borrowed instance.

type ConnectionName 

type ConnectionName string

ConnectionName identifies one configured redis instance, distinct from other service families.

func (ConnectionName) Validate 

func (n ConnectionName) Validate() error

type Connections 

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

Connections is immutable. Default and named access alias the same owned instance. Resolve dependencies during construction; pass concrete services to domain code.

func NewConnections 

func NewConnections(selected ConnectionName, entries ...Connection) (*Connections, error)

func (*Connections) Connection 

func (r *Connections) Connection(name ConnectionName) (*Client, error)

func (*Connections) Default 

func (r *Connections) Default() (*Client, error)

func (*Connections) DefaultName 

func (r *Connections) DefaultName() ConnectionName

func (*Connections) Names 

func (r *Connections) Names() []ConnectionName

type JobBackend 

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

JobBackend borrows the existing Redis client and owns no connections. Limits apply per queue and must match across processes. Accepted work persists until terminal retention expires or retained-record eviction; durability depends on Redis persistence/failover. Stop workers and publication before closing the client's owning module.

Queues use storage layout 2: the immutable envelope is stored apart from small mutable state. A layout-1 queue is never migrated implicitly: every operation returns jobs.ErrLegacyLayout until an operator stops or drains the previous release and runs JobMigrateLayout (`jobs migrate-layout`). The previous release cannot read a migrated queue.

func NewJobBackend 

func NewJobBackend(client *Client, config jobs.QueueConfig) (*JobBackend, error)

func (*JobBackend) DurableAcceptance 

func (*JobBackend) DurableAcceptance() bool

DurableAcceptance describes Redis acknowledgement under the deployment's persistence/failover policy; it is not an fsync or replication guarantee.

func (*JobBackend) JobCancel 

func (b *JobBackend) JobCancel(ctx context.Context, key jobs.Key, target jobs.Target) (bool, error)

func (*JobBackend) JobCancelWorkflow 

func (b *JobBackend) JobCancelWorkflow(ctx context.Context, key jobs.Key, id jobs.WorkflowID) (bool, error)

func (*JobBackend) JobEnqueue 

func (b *JobBackend) JobEnqueue(ctx context.Context, key jobs.Key, envelope jobs.Envelope) (bool, error)

func (*JobBackend) JobFinish 

func (b *JobBackend) JobFinish(ctx context.Context, key jobs.Key, proof jobs.Ownership, result jobs.Result) (bool, error)

func (*JobBackend) JobForget 

func (b *JobBackend) JobForget(ctx context.Context, key jobs.Key, target jobs.Target) (bool, error)

JobForget removes one retained terminal independent record, its envelope and its deduplication identity. Live and workflow records are never removed.

func (*JobBackend) JobInspect 

func (b *JobBackend) JobInspect(ctx context.Context, key jobs.Key, id jobs.ExecutionID) (value.Optional[jobs.Record], error)

func (*JobBackend) JobList 

func (b *JobBackend) JobList(ctx context.Context, key jobs.Key, options jobs.ListOptions) (jobs.Page, error)

func (*JobBackend) JobMigrateLayout 

func (b *JobBackend) JobMigrateLayout(ctx context.Context, key jobs.Key) (bool, error)

JobMigrateLayout moves a layout-1 queue to layout 2 in one atomic script. Records keep their IDs, state and history; embedded envelopes move on each record's next write. It is idempotent.

func (*JobBackend) JobRenew 

func (b *JobBackend) JobRenew(ctx context.Context, key jobs.Key, proof jobs.Ownership, ttl time.Duration) (jobs.LeaseStatus, error)

func (*JobBackend) JobReserve 

func (b *JobBackend) JobReserve(ctx context.Context, key jobs.Key, owner lease.Owner, ttl time.Duration) (value.Optional[jobs.Reservation], error)

func (*JobBackend) JobRetry 

func (b *JobBackend) JobRetry(ctx context.Context, key jobs.Key, request jobs.RetryRequest) (bool, error)

func (*JobBackend) JobStart 

func (b *JobBackend) JobStart(ctx context.Context, key jobs.Key, proof jobs.Ownership) (uint32, error)

func (*JobBackend) JobStats 

func (b *JobBackend) JobStats(ctx context.Context, key jobs.Key) (jobs.QueueStats, error)

JobStats reports queue depth from indexes and counters, without payloads.

func (*JobBackend) JobWakeup 

func (b *JobBackend) JobWakeup(key jobs.Key) <-chan struct{}

JobWakeup wakes idle workers of this process subscribed to key after this process's own accepted enqueue in that queue. Workers in other processes rely on their adaptive idle polling.

func (*JobBackend) JobWorkflow 

func (b *JobBackend) JobWorkflow(ctx context.Context, key jobs.Key, workflow jobs.WorkflowEnvelope) (bool, error)

func (*JobBackend) JobWorkflowStatus 

func (b *JobBackend) JobWorkflowStatus(ctx context.Context, key jobs.Key, id jobs.WorkflowID) (value.Optional[jobs.WorkflowStatus], error)

JobWorkflowStatus reports one workflow's progress without payloads.

type Stats 

type Stats struct {
	Open, Idle, Operations int
	// Subscriptions includes establishment/drain; SubscriptionConnections is the
	// driver's live dedicated connection count, separate from Open.
	Subscriptions, SubscriptionConnections int
	Ready, Closing                         bool
}

Stats is a concurrent snapshot of owned operations and connection use.

type TLSMode 

type TLSMode string

TLSMode selects verified TLS or explicitly unencrypted transport. There is no automatic downgrade or fallback to a different endpoint. 

const (
	VerifyTLS  TLSMode = "verify-full"
	DisableTLS TLSMode = "disable"
)

type WebSocketBackend 

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

WebSocketBackend borrows the existing Redis pool and pub/sub implementation. It owns no server/connection lifecycle and never retries a state mutation. Keep its Client alive until every borrowing Hub/Publisher has actually stopped.

func NewWebSocketBackend 

func NewWebSocketBackend(client *Client) (*WebSocketBackend, error)

func (*WebSocketBackend) Publish 

func (b *WebSocketBackend) Publish(ctx context.Context, channel pubsub.Channel, data []byte) (uint64, error)

func (*WebSocketBackend) Subscribe 

func (b *WebSocketBackend) Subscribe(ctx context.Context, channels []pubsub.Channel, limits pubsub.Limits) (pubsub.Stream, error)

func (*WebSocketBackend) WebSocketAppend 

func (b *WebSocketBackend) WebSocketAppend(ctx context.Context, key websocket.ClusterKey, channel websocket.ChannelID, policy websocket.ReplayConfig, data []byte) error

func (*WebSocketBackend) WebSocketCheck 

func (b *WebSocketBackend) WebSocketCheck(ctx context.Context, key websocket.ClusterKey) error

func (*WebSocketBackend) WebSocketClose 

func (b *WebSocketBackend) WebSocketClose(ctx context.Context, key websocket.ClusterKey, instance websocket.InstanceID, connection websocket.ConnectionID) error

func (*WebSocketBackend) WebSocketHistory 

func (b *WebSocketBackend) WebSocketHistory(ctx context.Context, key websocket.ClusterKey, channel websocket.ChannelID, policy websocket.ReplayConfig) ([][]byte, error)

func (*WebSocketBackend) WebSocketJoin 

func (b *WebSocketBackend) WebSocketJoin(ctx context.Context, key websocket.ClusterKey, instance websocket.InstanceID, connection websocket.ConnectionID, membership websocket.ClusterMembership) (websocket.PresenceSnapshot, error)

func (*WebSocketBackend) WebSocketLeave 

func (b *WebSocketBackend) WebSocketLeave(ctx context.Context, key websocket.ClusterKey, instance websocket.InstanceID, connection websocket.ConnectionID, scope websocket.Scope) error

func (*WebSocketBackend) WebSocketMembers 

func (b *WebSocketBackend) WebSocketMembers(ctx context.Context, key websocket.ClusterKey, scope websocket.Scope) (websocket.PresenceSnapshot, error)

func (*WebSocketBackend) WebSocketOpen 

func (b *WebSocketBackend) WebSocketOpen(ctx context.Context, key websocket.ClusterKey, instance websocket.InstanceID, connection websocket.ConnectionID) error

func (*WebSocketBackend) WebSocketTouch 

func (b *WebSocketBackend) WebSocketTouch(ctx context.Context, key websocket.ClusterKey, instance websocket.InstanceID, connection websocket.ConnectionID) error

Source Files

View all Source files

Directories

Path Synopsis
Package data provides typed Redis hashes, sets, sorted sets and lists.
 Package data provides typed Redis hashes, sets, sorted sets and lists.
Package raw is the explicit trusted Redis command boundary.
 Package raw is the explicit trusted Redis command boundary.

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.