redisync

README

redisync

Synchronization primitives with Redis

TODO

• Support Redis Cluster

Documentation

Index

• Variables
• func TryLock(conn redis.Conn, key string, expiration time.Duration) (string, error)
• func Unlock(conn redis.Conn, key, value string) (err error)
• type BackOffFactory
• type BackOffFactoryFunc
  • func (f BackOffFactoryFunc) Create(ctx context.Context) (bo backoff.BackOff)
• type Config
• type Monitor
  • func NewMonitor(pool Pool, opts ...Option) *Monitor
  • func (m *Monitor) Synchronize(ctx context.Context, key string, do func(context.Context) error) (err error)
• type Mutex
  • func NewMutex(pool Pool, key string, opts ...Option) *Mutex
  • func (m *Mutex) Lock(ctx context.Context) error
  • func (m *Mutex) Unlock(ctx context.Context) error
• type Once
  • func NewOnce(pool Pool, opts ...OnceOption) *Once
  • func (o *Once) Do(ctx context.Context, key string, f func(context.Context) error) error
• type OnceConfig
• type OnceOption
  • func WithOnceExpiration(d time.Duration) OnceOption
  • func WithOnceUnlockAfterError() OnceOption
• type Option
  • func WithBackOffFactory(f BackOffFactory) Option
  • func WithLockExpiration(d time.Duration) Option
• type Pool
• type ScoreFilter
  • func NewScoreFilter(pool Pool, opts ...Option) *ScoreFilter
  • func (f *ScoreFilter) Filter(ctx context.Context, key string, score int) (ok bool, err error)

Examples

• Monitor
• Mutex
• Once
• ScoreFilter

Variables

var (
	DefaultBackOffFactory BackOffFactory = BackOffFactoryFunc(func() backoff.BackOff {
		bo := backoff.NewExponentialBackOff()
		bo.MaxInterval = 1 * time.Second
		bo.InitialInterval = 5 * time.Millisecond
		return bo
	})
	DefaultLockExpiration = 120 * time.Second
	DefaultOnceExpiration = 3 * 24 * time.Second
)

var (
	ErrLocked   = errors.New("already locked")
	ErrConflict = errors.New("locked by another process")
)

Functions

func TryLock

func TryLock(conn redis.Conn, key string, expiration time.Duration) (string, error)

func Unlock

func Unlock(conn redis.Conn, key, value string) (err error)

Types

type BackOffFactory

type BackOffFactory interface {
	Create(ctx context.Context) backoff.BackOff
}

type BackOffFactoryFunc

type BackOffFactoryFunc func() backoff.BackOff

func (BackOffFactoryFunc) Create

func (f BackOffFactoryFunc) Create(ctx context.Context) (bo backoff.BackOff)

type Config

type Config struct {
	BackOffFactory BackOffFactory
	LockExpiration time.Duration
}

type Monitor

type Monitor struct {
	Config
	// contains filtered or unexported fields
}

Example

defer cleanupTestRedis()

var wg sync.WaitGroup
ctx := context.Background()

monitor := redisync.NewMonitor(pool)

for i := 0; i < 5; i++ {
	wg.Add(1)
	go func() {
		defer wg.Done()
		monitor.Synchronize(ctx, "key", func(context.Context) error {
			fmt.Println("start")
			time.Sleep(10 * time.Millisecond)
			fmt.Println("stop")
			return nil
		})
	}()
}

wg.Wait()

Output:

start
stop
start
stop
start
stop
start
stop
start
stop

func NewMonitor

func NewMonitor(pool Pool, opts ...Option) *Monitor

func (*Monitor) Synchronize

func (m *Monitor) Synchronize(ctx context.Context, key string, do func(context.Context) error) (err error)

type Mutex

type Mutex struct {
	Config
	// contains filtered or unexported fields
}

Example

defer cleanupTestRedis()

var wg sync.WaitGroup
ctx := context.Background()

mu := redisync.NewMutex(pool, "key")

for i := 0; i < 5; i++ {
	wg.Add(1)
	go func() {
		defer wg.Done()
		mu.Lock(ctx)
		defer mu.Unlock(ctx)

		fmt.Println("start")
		time.Sleep(10 * time.Millisecond)
		fmt.Println("stop")
	}()
}

wg.Wait()

Output:

start
stop
start
stop
start
stop
start
stop
start
stop

func NewMutex

func NewMutex(pool Pool, key string, opts ...Option) *Mutex

func (*Mutex) Lock

func (m *Mutex) Lock(ctx context.Context) error

func (*Mutex) Unlock

func (m *Mutex) Unlock(ctx context.Context) error

type Once

type Once struct {
	OnceConfig
	// contains filtered or unexported fields
}

Example

defer cleanupTestRedis()

var wg sync.WaitGroup
ctx := context.Background()

once := redisync.NewOnce(pool)

for i := 0; i < 10; i++ {
	wg.Add(1)
	go func() {
		defer wg.Done()
		once.Do(ctx, "key", func(context.Context) error {
			fmt.Println("Only once")
			return nil
		})
	}()
}

wg.Wait()

Output:

Only once

func NewOnce

func NewOnce(pool Pool, opts ...OnceOption) *Once

func (*Once) Do

func (o *Once) Do(ctx context.Context, key string, f func(context.Context) error) error

type OnceConfig

type OnceConfig struct {
	Expiration       time.Duration
	UnlockAfterError bool
}

type OnceOption

type OnceOption func(*OnceConfig)

func WithOnceExpiration

func WithOnceExpiration(d time.Duration) OnceOption

func WithOnceUnlockAfterError

func WithOnceUnlockAfterError() OnceOption

type Option

type Option func(*Config)

func WithBackOffFactory

func WithBackOffFactory(f BackOffFactory) Option

func WithLockExpiration

func WithLockExpiration(d time.Duration) Option

type Pool

type Pool interface {
	GetContext(context.Context) (redis.Conn, error)
}

type ScoreFilter

type ScoreFilter struct {
	Config
	// contains filtered or unexported fields
}

Example

defer cleanupTestRedis()

ctx := context.Background()

filter := redisync.NewScoreFilter(pool)

for i := 10; i > 0; i-- {
	i := i

	ok, err := filter.Filter(ctx, "key", i)
	if err != nil {
		// ...
	}
	if ok {
		fmt.Println("Only once")
	}
}

Output:

Only once

func NewScoreFilter

func NewScoreFilter(pool Pool, opts ...Option) *ScoreFilter

func (*ScoreFilter) Filter

func (f *ScoreFilter) Filter(ctx context.Context, key string, score int) (ok bool, err error)