redisq

README

Redisq

Redisq is a queue-over-redis that provides simple way to works with queues stored in Redis.

Lib works over redis lib.

Advantages

• Simple - just launch redis-server on persistent mode
• Enough persistence
• Good for periodic poll model
• No need for mature message broker on simple queue task, use if you already have redis-server
• Distributed write

Disadvantages

• No broadcasting
• No distributed read
• One transaction per time

Warnings

• Use noeviction mode of redis-server for persist your data

Requirements

• Redis-server >= 6.2.0

Installation

go get github.com/go-redis/redis@latest

Features:

• BeginRead locks queue for reading
• Push into queue without blocking
• Scan from queue without removing
• Commit scanned elements
• Rollback transaction with moving elements into end of queue
• Cancel breaks transaction, remains elements in queue

On BeginRead queue locks for read and unlocks after Commit, Rollback, or Cancel.

For generic usage with any type lib provides StringConverter interface

Example

type intStringer int // implements redisq.StringConverter

setter := intStringer(1)

q, _ := redisq.NewSeqQueue(cfg)

q.Push(ctx, setter)

getter := intStringer(0)

q.BeginRead(ctx)
q.Scan(ctx, getter)
q.Commit(ctx)

// getter will be intStringer(1)

See queue_test.go for more examples.

Roadmap

• Distributed read
• Queue with priority
• Non-blocking transactions

Run tests

Warning: tests clear the storage after every test.

• Set TEST_REDIS_QUEUE_NETWORK and TEST_REDIS_QUEUE_ADDR, or just launch redis-server on localhost:6379
• Run go test ./...

Documentation

Index

• Variables
• type BeginMode
• type Queue
  • func NewSeqQueue(cfg *QueueConfig) (Queue, error)
• type QueueConfig
• type StringConverter

Variables

var (
	ErrWrongType = errors.New("type doesn't match")
	ErrNoTx      = errors.New("no transaction in progress")
	ErrAlreadyTx = errors.New("transaction already opened")
)

Types

type BeginMode

type BeginMode int64

BeginMode defines behavior of Queue.BeginRead.

const (
	// BlockBeginMode defines waiting while queue unlocks.
	BlockBeginMode BeginMode = iota
	// ErrBeginMode defines error if queue locked on Queue.BeginRead.
	ErrBeginMode
)

type Queue

type Queue interface {
	// BeginRead locks queue for Scan.
	// Must be called before Scan.
	BeginRead(ctx context.Context) error
	// Push calls StringConverter.ToString and adds result to the queue.
	// Non-blocking.
	Push(ctx context.Context, t StringConverter) error
	// Scan gets next queue element and put it into StringConverter.FromString.
	// Element remain in queue.
	Scan(ctx context.Context, t StringConverter) error
	// Commit removes elements from queue and unlocks it.
	// Must call after sequence of Scan.
	Commit(ctx context.Context) error
	// Rollback move element into end of queue and unlocks it.
	// Must call after sequence of Scan.
	Rollback(ctx context.Context) error
	// Cancel breaks transaction. Elements remains in queue.
	// Must call after sequence of Scan.
	Cancel(ctx context.Context) error
}

func NewSeqQueue

func NewSeqQueue(cfg *QueueConfig) (Queue, error)

type QueueConfig

type QueueConfig struct {
	Conn redis.UniversalClient
	// Name is the name of queue.
	// Don't use same name in some queues.
	Name string
	// Typ is the type of queue
	Typ       StringConverter
	BeginMode BeginMode
}

type StringConverter

type StringConverter interface {
	ToString() (string, error)
	FromString(s string) error
}