queue

README

Universal queue library for Golang

uni-queue library defines basic Queue interface that implements queues using:

• Redis Lists - ListQueue
• Redis Streams - StreamQueue (TBD)
• AWS SQS - SQS (TBD)

Installation

go get github.com/fastforgeinc/uni-queue

Quickstart

package main

import (
	"context"
	"log"
	"time"

	"github.com/go-redis/redis/v8"
	"github.com/fastforgeinc/uni-queue"
)

type Object struct {
	Str string
	Num int
}

func main() {
	ctx := context.TODO()

	// Construct ®Redis client
	client := redis.NewClient(&redis.Options{
		Addr: "localhost:6379",
	})

	// Construct ListQueue
	q := queue.NewListQueue(client)

	// Construct ListQueue with dequeue timeout
	timeout := time.Second
	q := queue.NewListQueue(client, queue.WithDequeueTimeout(timeout))

	// Enqueue value
	input := Object{"foo", 69}
	err := q.Enqueue(ctx, "queue:name", &input)
	if err != nil {
		log.Fatal(err)
	}

	// Dequeue value
	var output Object
	err := queue.Dequeue(ctx, "queue:name", &output)
	if err != nil {
		log.Fatal(err)
	}
	log.Printf("%v", output)
}

Documentation

Index

• type ListQueue
  • func NewListQueue(redis Rediser, opts ...Option) *ListQueue
  • func (q *ListQueue) Dequeue(ctx context.Context, queue string, value interface{}) error
  • func (q *ListQueue) Enqueue(ctx context.Context, queue string, value interface{}) error
  • func (q *ListQueue) Flush(ctx context.Context, queue string) error
  • func (q *ListQueue) Len(ctx context.Context, queue string) int
• type Option
  • func WithDequeueTimeout(t time.Duration) Option
• type Queuer
• type Rediser

Types

type ListQueue

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

func NewListQueue

func NewListQueue(redis Rediser, opts ...Option) *ListQueue

func (*ListQueue) Dequeue

func (q *ListQueue) Dequeue(ctx context.Context, queue string, value interface{}) error

func (*ListQueue) Enqueue

func (q *ListQueue) Enqueue(ctx context.Context, queue string, value interface{}) error

func (*ListQueue) Flush

func (q *ListQueue) Flush(ctx context.Context, queue string) error

func (*ListQueue) Len

func (q *ListQueue) Len(ctx context.Context, queue string) int

type Option

type Option func(*options)

func WithDequeueTimeout

func WithDequeueTimeout(t time.Duration) Option

WithDequeueTimeout sets timeout of enqueue operation, default indefinitely

type Queuer

type Queuer interface {
	Enqueue(ctx context.Context, queue string, value interface{}) error
	Dequeue(ctx context.Context, queue string, value interface{}) error
	Flush(ctx context.Context, queue string) error
	Len(ctx context.Context, queue string) int
}

type Rediser

type Rediser interface {
	LPush(ctx context.Context, key string, values ...interface{}) *redis.IntCmd
	BRPop(ctx context.Context, timeout time.Duration, keys ...string) *redis.StringSliceCmd
	LLen(ctx context.Context, key string) *redis.IntCmd
	Del(ctx context.Context, keys ...string) *redis.IntCmd
}