redis

Documentation

Index

• Constants
• func ClientHasJSONSupport(c RedisClient) bool
• func GetConnectedSlaves(ctx context.Context, c RedisClient) (int, error)
• func GetServerVersion(c RedisClient) (string, error)
• func ParseClientFromProperties(properties map[string]string, componentType metadata.ComponentType, ...) (RedisClient, *Settings, error)
• func StartEntraIDTokenRefreshBackgroundRoutine(client RedisClient, username string, nextExpiration time.Time, ...)
• type ConfigurationSubscribeArgs
• type Duration
  • func (r *Duration) DecodeString(value string) error
• type RedisClient
  • func ClientFromV8Client(client v8.UniversalClient) RedisClient
• type RedisError
  • func (e RedisError) Error() string
  • func (RedisError) RedisError()
• type RedisPipeliner
• type RedisXMessage
• type RedisXPendingExt
• type RedisXStream
• type Settings
  • func (s *Settings) Decode(in interface{}) error
  • func (s *Settings) GetEntraIDCredentialAndSetInitialTokenAsPassword(ctx context.Context, properties *map[string]string) (*time.Time, *azcore.TokenCredential, error)
  • func (s *Settings) SetCertificate(fn func(cert *tls.Certificate)) error

Constants

const (
	ClusterType = "cluster"
	NodeType    = "node"
)

Functions

func ClientHasJSONSupport

func ClientHasJSONSupport(c RedisClient) bool

func GetConnectedSlaves

func GetConnectedSlaves(ctx context.Context, c RedisClient) (int, error)

GetConnectedSlaves returns the number of slaves connected to the Redis master.

func GetServerVersion

func GetServerVersion(c RedisClient) (string, error)

func ParseClientFromProperties

func ParseClientFromProperties(properties map[string]string, componentType metadata.ComponentType, ctx context.Context, logger *kitlogger.Logger) (RedisClient, *Settings, error)

func StartEntraIDTokenRefreshBackgroundRoutine

func StartEntraIDTokenRefreshBackgroundRoutine(client RedisClient, username string, nextExpiration time.Time, cred *azcore.TokenCredential, logger *kitlogger.Logger)

Types

type ConfigurationSubscribeArgs

type ConfigurationSubscribeArgs struct {
	HandleSubscribedChange func(ctx context.Context, req *configuration.SubscribeRequest, handler configuration.UpdateHandler, channel string, id string)
	Req                    *configuration.SubscribeRequest
	Handler                configuration.UpdateHandler
	RedisChannel           string
	IsAllKeysChannel       bool
	ID                     string
}

type Duration

type Duration time.Duration

func (*Duration) DecodeString

func (r *Duration) DecodeString(value string) error

type RedisClient

type RedisClient interface {
	GetNilValueError() RedisError
	Context() context.Context
	DoRead(ctx context.Context, args ...interface{}) (interface{}, error)
	DoWrite(ctx context.Context, args ...interface{}) error
	Del(ctx context.Context, keys ...string) error
	Get(ctx context.Context, key string) (string, error)
	GetDel(ctx context.Context, key string) (string, error)
	Close() error
	PingResult(ctx context.Context) (string, error)
	ConfigurationSubscribe(ctx context.Context, args *ConfigurationSubscribeArgs)
	SetNX(ctx context.Context, key string, value interface{}, expiration time.Duration) (*bool, error)
	EvalInt(ctx context.Context, script string, keys []string, args ...interface{}) (*int, error, error)
	XAdd(ctx context.Context, stream string, maxLenApprox int64, values map[string]interface{}) (string, error)
	XGroupCreateMkStream(ctx context.Context, stream string, group string, start string) error
	XAck(ctx context.Context, stream string, group string, messageID string) error
	XReadGroupResult(ctx context.Context, group string, consumer string, streams []string, count int64, block time.Duration) ([]RedisXStream, error)
	XPendingExtResult(ctx context.Context, stream string, group string, start string, end string, count int64) ([]RedisXPendingExt, error)
	XClaimResult(ctx context.Context, stream string, group string, consumer string, minIdleTime time.Duration, messageIDs []string) ([]RedisXMessage, error)
	TxPipeline() RedisPipeliner
	TTLResult(ctx context.Context, key string) (time.Duration, error)
	AuthACL(ctx context.Context, username, password string) error
}

func ClientFromV8Client

func ClientFromV8Client(client v8.UniversalClient) RedisClient

type RedisError

type RedisError string

func (RedisError) Error

func (e RedisError) Error() string

func (RedisError) RedisError

func (RedisError) RedisError()

type RedisPipeliner

type RedisPipeliner interface {
	Exec(ctx context.Context) error
	Do(ctx context.Context, args ...interface{})
}

type RedisXMessage

type RedisXMessage struct {
	ID     string
	Values map[string]interface{}
}

type RedisXPendingExt

type RedisXPendingExt struct {
	ID         string
	Consumer   string
	Idle       time.Duration
	RetryCount int64
}

type RedisXStream

type RedisXStream struct {
	Stream   string
	Messages []RedisXMessage
}

type Settings

type Settings struct {
	// The Redis host
	Host string `mapstructure:"redisHost"`
	// The Redis password
	Password string `mapstructure:"redisPassword"`
	// The Redis username
	Username string `mapstructure:"redisUsername"`
	// Database to be selected after connecting to the server.
	DB int `mapstructure:"redisDB"`
	// The redis type node or cluster
	RedisType string `mapstructure:"redisType"`
	// Maximum number of retries before giving up.
	// A value of -1 (not 0) disables retries
	// Default is 3 retries
	RedisMaxRetries int `mapstructure:"redisMaxRetries"`
	// Minimum backoff between each retry.
	// Default is 8 milliseconds; -1 disables backoff.
	RedisMinRetryInterval Duration `mapstructure:"redisMinRetryInterval"`
	// Maximum backoff between each retry.
	// Default is 512 milliseconds; -1 disables backoff.
	RedisMaxRetryInterval Duration `mapstructure:"redisMaxRetryInterval"`
	// Dial timeout for establishing new connections.
	DialTimeout Duration `mapstructure:"dialTimeout"`
	// Timeout for socket reads. If reached, commands will fail
	// with a timeout instead of blocking. Use value -1 for no timeout and 0 for default.
	ReadTimeout Duration `mapstructure:"readTimeout"`
	// Timeout for socket writes. If reached, commands will fail
	WriteTimeout Duration `mapstructure:"writeTimeout"`
	// Maximum number of socket connections.
	PoolSize int `mapstructure:"poolSize"`
	// Minimum number of idle connections which is useful when establishing
	// new connection is slow.
	MinIdleConns int `mapstructure:"minIdleConns"`
	// Connection age at which client retires (closes) the connection.
	// Default is to not close aged connections.
	MaxConnAge Duration `mapstructure:"maxConnAge"`
	// Amount of time client waits for connection if all connections
	// are busy before returning an error.
	// Default is ReadTimeout + 1 second.
	PoolTimeout Duration `mapstructure:"poolTimeout"`
	// Amount of time after which client closes idle connections.
	// Should be less than server's timeout.
	// Default is 5 minutes. -1 disables idle timeout check.
	IdleTimeout Duration `mapstructure:"idleTimeout"`
	// Frequency of idle checks made by idle connections reaper.
	// Default is 1 minute. -1 disables idle connections reaper,
	// but idle connections are still discarded by the client
	// if IdleTimeout is set.
	IdleCheckFrequency Duration `mapstructure:"idleCheckFrequency"`
	// The master name
	SentinelMasterName string `mapstructure:"sentinelMasterName"`
	// Use Redis Sentinel for automatic failover.
	Failover bool `mapstructure:"failover"`

	// A flag to enables TLS by setting InsecureSkipVerify to true
	EnableTLS bool `mapstructure:"enableTLS"`

	// Client certificate and key
	ClientCert string `mapstructure:"clientCert"`
	ClientKey  string `mapstructure:"clientKey"`

	// == state only properties ==
	TTLInSeconds *int   `mapstructure:"ttlInSeconds" mdonly:"state"`
	QueryIndexes string `mapstructure:"queryIndexes" mdonly:"state"`

	// == pubsub only properties ==
	// The consumer identifier
	ConsumerID string `mapstructure:"consumerID" mdonly:"pubsub"`
	// The interval between checking for pending messages to redelivery (0 disables redelivery)
	RedeliverInterval time.Duration `mapstructure:"-" mdonly:"pubsub"`
	// The amount time a message must be pending before attempting to redeliver it (0 disables redelivery)
	ProcessingTimeout time.Duration `mapstructure:"processingTimeout" mdonly:"pubsub"`
	// The size of the message queue for processing
	QueueDepth uint `mapstructure:"queueDepth" mdonly:"pubsub"`
	// The number of concurrent workers that are processing messages
	Concurrency uint `mapstructure:"concurrency" mdonly:"pubsub"`

	// The max len of stream
	MaxLenApprox int64 `mapstructure:"maxLenApprox" mdonly:"pubsub"`

	// EntraID / AzureAD Authentication based on the shared code which essentially uses the DefaultAzureCredential
	// from the official Azure Identity SDK for Go
	UseEntraID bool `mapstructure:"useEntraID" mapstructurealiases:"useAzureAD"`
}

func (*Settings) Decode

func (s *Settings) Decode(in interface{}) error

func (*Settings) GetEntraIDCredentialAndSetInitialTokenAsPassword

func (s *Settings) GetEntraIDCredentialAndSetInitialTokenAsPassword(ctx context.Context, properties *map[string]string) (*time.Time, *azcore.TokenCredential, error)

func (*Settings) SetCertificate

func (s *Settings) SetCertificate(fn func(cert *tls.Certificate)) error