redis

Documentation

Index

• Variables
• type CmdableExtension
• type Config
  • func Configure(key ...string) (*Config, error)
  • func Defaults() *Config
• type Entry
• type KVStore
  • func NewKVStore(client Redis) KVStore
• type Redis
  • func New(cfg *Config) (Redis, error)
• type SortedSet
  • func NewZSortedSet(client Redis, name string) SortedSet
• type StreamMessageHandler
• type StreamMessageReq
  • func (msg *StreamMessageReq) Decode(v interface{}) error
• type StreamMessageRes
• type Streamer
  • func NewStreamer(client Redis) Streamer

Variables

var (
	ErrConn               = errors.New("cannot connect to address")
	ErrShutdown           = errors.New("cannot shutdown redisCluster client")
	ErrZSetEntryAdd       = errors.New("cannot add entry to zset")
	ErrZSetEntryDelete    = errors.New("cannot delete entry from zset")
	ErrStreamNotExist     = errors.New("ERR no such key")
	ErrConsumerGroupExist = errors.New("BUSYGROUP Consumer Group name already exists")
)

Types

type CmdableExtension

type CmdableExtension interface {
	goRedisV9.Cmdable
}

CmdableExtension interface extension to pass to various Redis extensions in this package

type Config

type Config struct {
	Addrs []string `mapstructure:"redisAddrs" envVar:"ATMK_REDIS_ADDRS"`
	Auth  string   `mapstructure:"redisAuth" envVar:"ATMK_REDIS_AUTH"`
	DB    int      `mapstructure:"redisDb" envVar:"ATMK_REDIS_DB"`
}

func Configure

func Configure(key ...string) (*Config, error)

func Defaults

func Defaults() *Config

type Entry

type Entry struct {
	ID        string
	ExecuteAt time.Time
}

type KVStore

type KVStore interface {
	Set(ctx context.Context, key string, value interface{}, expiry time.Duration) error
	Get(ctx context.Context, key string) (string, error)
	Delete(ctx context.Context, key string) error
}

func NewKVStore

func NewKVStore(client Redis) KVStore

NewKVStore creates a new instance of kvStore.

type Redis

type Redis interface {
	atomika.Service
	Client() CmdableExtension
	Log() log.Ctx
	Close() error
}

func New

func New(cfg *Config) (Redis, error)

New create a new Redis client/cluster connection handler

type SortedSet

type SortedSet interface {
	C() <-chan Entry
	Start(ctx context.Context, interval time.Duration)
	Stop()
	AddEntry(ctx context.Context, entry Entry) error
	DeleteEntry(ctx context.Context, entryID string) error
	GetEntries(ctx context.Context, min, max string) ([]Entry, error)
}

func NewZSortedSet

func NewZSortedSet(client Redis, name string) SortedSet

NewZSortedSet start a new ZSet internal ticker with a name and a ticker in seconds

type StreamMessageHandler

type StreamMessageHandler interface {
	HandleMessage(ctx context.Context, req StreamMessageReq) StreamMessageRes
}

type StreamMessageReq

type StreamMessageReq struct {
	ID     string
	Stream string
	Values map[string]interface{}
}

func (*StreamMessageReq) Decode

func (msg *StreamMessageReq) Decode(v interface{}) error

type StreamMessageRes

type StreamMessageRes struct {
	Processed bool
	Error     error
}

type Streamer

type Streamer interface {
	CreateStream(ctx context.Context, streamName, groupName string) error
	CleanupStream(ctx context.Context, streamName string, retentionPeriod time.Duration)
	Publish(ctx context.Context, streamName string, message map[string]interface{}) error
	Subscribe(ctx context.Context, streamName, groupName, consumerName string, handler StreamMessageHandler) error
}

func NewStreamer

func NewStreamer(client Redis) Streamer

NewStreamer create a new streamer instance