goredisqueue

package module
v0.0.0-...-8b31767 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jun 22, 2025 License: MIT Imports: 6 Imported by: 0

README

Redis-Queue

A lightweight Go library to use Redis as a message queue. Supports structured and primitive data types using customizable serialization.

Features

  • Blocking Enqueue / Dequeue
  • Generic Payload interface
  • Built-in support for structs and primitives
  • Minimal API surface with helper constructors

Install

go get github.com/prakashpandey/goredisqueue

Examples

Queue Initialization

queue := goredisqueue.NewWithOptions(goredisqueue.Options{
    RedisAddr: "localhost:6379",
    QueueName: "queue-1",
    Timeout:   3 * time.Second,
})

Enqueue a Struct:

type Event struct {
    ID string `json:"id"`
}
e := Event{ID: "abc123"}
_ = queue.Enqueue(ctx, goredisqueue.NewPayloadFromValue(e))

Dequeue a Struct

var out Event
_ = queue.Dequeue(ctx, goredisqueue.NewPayloadFromPtr(&out))

Enqueue/Dequeue a String

_ = queue.Enqueue(ctx, goredisqueue.NewPayloadFromValue("hello"))

var msg string
_ = queue.Dequeue(ctx, goredisqueue.NewPayloadFromPtr(&msg))

Testing

Run: go test ./...

Helpers

Interfaces

type Payload interface {
    Marshal() ([]byte, error)
    Unmarshal([]byte) error
}

Convert any supported type to Payload type

NewPayloadFromValue(value T) *PrimitivePayload[T]
NewPayloadFromPtr(ptr *T) *PrimitivePayload[T]

Supported types: Any type tha can be supported by json.Marshal/Unmarshal can be used.

Documentation

Index

Constants

This section is empty.

Variables

View Source
var ErrInvalidQueueName = errors.New("invalid queue name")

ErrInvalidQueueName is returned when the queue name is invalid.

View Source
var ErrInvalidRedisAddr = errors.New("invalid Redis address")

ErrInvalidRedisAddr is returned when the Redis address is invalid.

View Source
var ErrInvalidTimeout = errors.New("invalid timeout")

ErrInvalidTimeout is returned when the timeout is invalid.

View Source
var (
	ErrTimeout = errors.New("no message received within the timeout")
)

Functions

This section is empty.

Types

type Options

type Options struct {
	RedisAddr     string
	RedisPassword string
	DB            int
	QueueName     string
	Timeout       time.Duration
}

func NewDefaultOptions

func NewDefaultOptions() *Options

func (*Options) GetDB

func (o *Options) GetDB() int

func (*Options) GetQueueName

func (o *Options) GetQueueName() string

func (*Options) GetRedisAddr

func (o *Options) GetRedisAddr() string

func (*Options) GetTimeout

func (o *Options) GetTimeout() time.Duration

func (*Options) SetDB

func (o *Options) SetDB(db int) *Options

func (*Options) SetQueueName

func (o *Options) SetQueueName(name string) *Options

func (*Options) SetRedisAddr

func (o *Options) SetRedisAddr(addr string) *Options

func (*Options) SetRedisPassword

func (o *Options) SetRedisPassword(password string) *Options

func (*Options) SetTimeout

func (o *Options) SetTimeout(timeout time.Duration) *Options

func (*Options) Validate

func (o *Options) Validate() error

type Payload

type Payload interface {
	Marshal() ([]byte, error)
	Unmarshal(data []byte) error
}

type PrimitivePayload

type PrimitivePayload[T any] struct {
	Value *T
}

PrimitivePayload wraps a primitive type or struct to make it a Payload.

func NewPayloadFromPtr

func NewPayloadFromPtr[T any](value *T) *PrimitivePayload[T]

NewPayloadFromPtr wraps an existing pointer in a PrimitivePayload.

func NewPayloadFromValue

func NewPayloadFromValue[T any](value T) *PrimitivePayload[T]

NewPayloadFromValue creates a new PrimitivePayload with a value. The value will be wrapped in a pointer.

func (*PrimitivePayload[T]) Marshal

func (p *PrimitivePayload[T]) Marshal() ([]byte, error)

func (*PrimitivePayload[T]) Unmarshal

func (p *PrimitivePayload[T]) Unmarshal(data []byte) error

type Queue

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

func New

func New(client *redis.Client, queueName string, timeout time.Duration) *Queue

func NewWithOptions

func NewWithOptions(opt Options) *Queue

func (*Queue) Dequeue

func (q *Queue) Dequeue(ctx context.Context, p Payload) error

func (*Queue) Enqueue

func (q *Queue) Enqueue(ctx context.Context, p Payload) error

type Queuer

type Queuer interface {
	Enqueue(ctx context.Context, p Payload) error
	Dequeue(ctx context.Context, p Payload) error
}

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL