bredis

Documentation

Index

• Variables
• func NewRdb(host, port string, password string, database, maxRetries int, ...) redis.UniversalClient
• func NewReadGroupArgs(group, consumer string, streams []string, count int64, block time.Duration) *redis.XReadGroupArgs
• func NewZAddArgs(stream, minId, Id string, maxLen, Limit int64, vals any) *redis.XAddArgs
• type Base
  • func (t *Base) Consumer(ctx context.Context, stream *public.Stream, handler public.CallBack) error
  • func (t *Base) DeadLetter(ctx context.Context, channel, topic string)
  • func (t *Base) Dequeue(ctx context.Context, channel, topic string, do public.CallBack)
  • func (t *Base) Enqueue(ctx context.Context, data map[string]any) error
• type BlockDuration
• type Log
  • func NewLog(client redis.UniversalClient, prefix string, log public.IMigrateLog) *Log
  • func (t *Log) Migrate(ctx context.Context, data []map[string]any) error
• type Normal
  • func NewNormal(client redis.UniversalClient, prefix string, maxLen int64, consumerCount int64, ...) *Normal
  • func (t *Normal) Dequeue(ctx context.Context, channel, topic string, do public.CallBack)
  • func (t *Normal) Enqueue(ctx context.Context, data map[string]any) error
• type PreWork
• type ProcessLog
  • func NewProcessLog(client redis.UniversalClient, prefix string) *ProcessLog
  • func (t *ProcessLog) AddLog(ctx context.Context, data map[string]any) error
• type RdbBroker
  • func NewBroker(client redis.UniversalClient, prefix string, maxLen, consumers int64, ...) *RdbBroker
  • func (t *RdbBroker) Mood(moodType btype.MoodType) public.IBroker
• type Schedule
  • func NewSchedule(client redis.UniversalClient, prefix string, consumerCount int64, ...) *Schedule
  • func (t *Schedule) Dequeue(ctx context.Context, channel, topic string, do public.CallBack)
  • func (t *Schedule) Enqueue(ctx context.Context, data map[string]any) error
  • func (t *Schedule) PreWork(ctx context.Context, prefix string, channel, topic string)
  • func (t *Schedule) Watcher(ctx context.Context, zsetMax string, zsetKey, streamKey string) func(tx *redis.Tx) error
• type Sequential
  • func NewSequential(client redis.UniversalClient, prefix string, consumerCount int64, ...) *Sequential
  • func (t *Sequential) Dequeue(ctx context.Context, channel, topic string, do public.CallBack)
  • func (t *Sequential) Enqueue(ctx context.Context, data map[string]any) error
• type Status
  • func NewStatus(client redis.UniversalClient, prefix string) *Status
  • func (t *Status) Status(ctx context.Context, channel, topic, id string) (map[string]string, error)
• type UITool
  • func NewUITool(client redis.UniversalClient, prefix string) *UITool
  • func (t *UITool) QueueMessage(ctx context.Context) error
• type Watcher

Variables

var (
	HashDuplicateIdScript = redis.NewScript(hashDuplicateIdLua)

	AddLogicLockScript = redis.NewScript(addLogicLockLua)

	SaveHSetScript = redis.NewScript(saveHsetLua)
)

Functions

func NewRdb

func NewRdb(host, port string, password string,
	database, maxRetries int, dialTimeout,
	readTimeout, writeTimeout, poolTimeout time.Duration, poolSize, minIdleConns int) redis.UniversalClient

func NewReadGroupArgs

func NewReadGroupArgs(group, consumer string, streams []string, count int64, block time.Duration) *redis.XReadGroupArgs

func NewZAddArgs

func NewZAddArgs(stream, minId, Id string, maxLen, Limit int64, vals any) *redis.XAddArgs

Types

type Base

type Base struct {
	public.IProcessLog
	// contains filtered or unexported fields
}

func (*Base) Consumer

func (t *Base) Consumer(ctx context.Context, stream *public.Stream, handler public.CallBack) error

func (*Base) DeadLetter

func (t *Base) DeadLetter(ctx context.Context, channel, topic string)

func (*Base) Dequeue

func (t *Base) Dequeue(ctx context.Context, channel, topic string, do public.CallBack)

func (*Base) Enqueue

func (t *Base) Enqueue(ctx context.Context, data map[string]any) error

type BlockDuration

type BlockDuration func() time.Duration

var (
	DefaultBlockDuration BlockDuration = func() time.Duration {
		return time.Duration(rand.Int63n(9)+1) * time.Second
	}
)

type Log

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

func NewLog

func NewLog(client redis.UniversalClient, prefix string, log public.IMigrateLog) *Log

func (*Log) Migrate

func (t *Log) Migrate(ctx context.Context, data []map[string]any) error

type Normal

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

func NewNormal

func NewNormal(client redis.UniversalClient, prefix string, maxLen int64, consumerCount int64, deadLetterIdle time.Duration) *Normal

func (*Normal) Dequeue

func (t *Normal) Dequeue(ctx context.Context, channel, topic string, do public.CallBack)

func (*Normal) Enqueue

func (t *Normal) Enqueue(ctx context.Context, data map[string]any) error

type PreWork

type PreWork func(ctx context.Context, prefix string, channel, topic string)

PreWork Check if there are any messages that can be consumed

type ProcessLog

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

func NewProcessLog

func NewProcessLog(client redis.UniversalClient, prefix string) *ProcessLog

func (*ProcessLog) AddLog

func (t *ProcessLog) AddLog(ctx context.Context, data map[string]any) error

type RdbBroker

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

func NewBroker

func NewBroker(client redis.UniversalClient, prefix string, maxLen, consumers int64, duration time.Duration) *RdbBroker

func (*RdbBroker) Mood

func (t *RdbBroker) Mood(moodType btype.MoodType) public.IBroker

type Schedule

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

func NewSchedule

func NewSchedule(client redis.UniversalClient, prefix string, consumerCount int64, deadLetterIdle time.Duration) *Schedule

func (*Schedule) Dequeue

func (t *Schedule) Dequeue(ctx context.Context, channel, topic string, do public.CallBack)

func (*Schedule) Enqueue

func (t *Schedule) Enqueue(ctx context.Context, data map[string]any) error

func (*Schedule) PreWork

func (t *Schedule) PreWork(ctx context.Context, prefix string, channel, topic string)

func (*Schedule) Watcher

func (t *Schedule) Watcher(ctx context.Context, zsetMax string, zsetKey, streamKey string) func(tx *redis.Tx) error

type Sequential

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

func NewSequential

func NewSequential(client redis.UniversalClient, prefix string, consumerCount int64, deadLetterIdle time.Duration) *Sequential

func (*Sequential) Dequeue

func (t *Sequential) Dequeue(ctx context.Context, channel, topic string, do public.CallBack)

func (*Sequential) Enqueue

func (t *Sequential) Enqueue(ctx context.Context, data map[string]any) error

type Status

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

func NewStatus

func NewStatus(client redis.UniversalClient, prefix string) *Status

func (*Status) Status

func (t *Status) Status(ctx context.Context, channel, topic, id string) (map[string]string, error)

type UITool

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

func NewUITool

func NewUITool(client redis.UniversalClient, prefix string) *UITool

func (*UITool) QueueMessage

func (t *UITool) QueueMessage(ctx context.Context) error

type Watcher

type Watcher func(ctx context.Context, zsetMax string, zsetKey, streamKey string) func(tx *redis.Tx) error

Watcher redis transaction