rsmq

README

RSMQ Fork

This is a fork of github.com/semihbkgr/go-rsmq v1.3.1 (commit d0f7bbc).

Changes from Original

The main change in this fork is support for custom message IDs. This allows for:

• Deterministic message IDs based on your own identifiers
• Message deduplication
• Message overriding/rescheduling

Custom ID Notes

When using custom IDs, be aware that:

• The msg.Sent timestamp will not reflect the actual send time for overridden messages
  • This field is derived from the message ID's timestamp part
  • For custom IDs, we use a fixed timestamp to avoid timing issues
  • Use Redis sorted set scores (via the delay parameter) for actual timing
• Message timing is controlled by Redis sorted set scores, not by the ID's timestamp part
• IDs must be exactly 32 characters long and contain only alphanumeric characters
• Overriding a message with the same ID but different delay will correctly update the timing

Original License

MIT License - see original repository for details.

Documentation

Index

• Constants
• Variables
• type QueueAttributes
• type QueueMessage
• type RedisSMQ
  • func NewRedisSMQ(client *redis.Client, ns string) *RedisSMQ
  • func (rsmq *RedisSMQ) ChangeMessageVisibility(qname string, id string, vt uint) error
  • func (rsmq *RedisSMQ) CreateQueue(qname string, vt uint, delay uint, maxsize int) error
  • func (rsmq *RedisSMQ) DeleteMessage(qname string, id string) error
  • func (rsmq *RedisSMQ) DeleteQueue(qname string) error
  • func (rsmq *RedisSMQ) GetQueueAttributes(qname string) (*QueueAttributes, error)
  • func (rsmq *RedisSMQ) ListQueues() ([]string, error)
  • func (rsmq *RedisSMQ) PopMessage(qname string) (*QueueMessage, error)
  • func (rsmq *RedisSMQ) Quit() error
  • func (rsmq *RedisSMQ) ReceiveMessage(qname string, vt uint) (*QueueMessage, error)
  • func (rsmq *RedisSMQ) SendMessage(qname string, message string, delay uint, opts ...SendMessageOption) (string, error)
  • func (rsmq *RedisSMQ) SetQueueAttributes(qname string, vt uint, delay uint, maxsize int) (*QueueAttributes, error)
• type SendMessageOption
  • func WithMessageID(id string) SendMessageOption

Constants

const (
	UnsetVt      = ^uint(0)
	UnsetDelay   = ^uint(0)
	UnsetMaxsize = -(int(^uint(0)>>1) - 1)
)

Unset values are the special values to refer default values of the attributes

Variables

var (
	ErrQueueNotFound   = errors.New("queue not found")
	ErrQueueExists     = errors.New("queue exists")
	ErrMessageTooLong  = errors.New("message too long")
	ErrMessageNotFound = errors.New("message not found")
)

Errors returned on rsmq operation

var (
	ErrInvalidQname   = errors.New("queue name is in wrong pattern")
	ErrInvalidVt      = errors.New("visibility timeout is out of range [0, 9999999]")
	ErrInvalidDelay   = errors.New("delay is out of range [0, 9999999]")
	ErrInvalidMaxsize = errors.New("max size is out of range [1024, 65536] and not -1")
	ErrInvalidID      = errors.New("id is in wrong pattern")
)

Validation errors

Types

type QueueAttributes

type QueueAttributes struct {
	Vt         uint
	Delay      uint
	Maxsize    int
	TotalRecv  uint64
	TotalSent  uint64
	Created    uint64
	Modified   uint64
	Msgs       uint64
	HiddenMsgs uint64
}

QueueAttributes contains some attributes and stats of queue

type QueueMessage

type QueueMessage struct {
	ID      string
	Message string
	Rc      uint64
	Fr      time.Time
	Sent    time.Time
}

QueueMessage contains content and metadata of message received from queue

type RedisSMQ

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

RedisSMQ is the client of rsmq to execute queue and message operations

func NewRedisSMQ

func NewRedisSMQ(client *redis.Client, ns string) *RedisSMQ

NewRedisSMQ creates and returns new rsmq client

func (*RedisSMQ) ChangeMessageVisibility

func (rsmq *RedisSMQ) ChangeMessageVisibility(qname string, id string, vt uint) error

ChangeMessageVisibility changes message visibility to refer queue vt

err:=redisRsmq.ChangeMessageVisibility(qname,id,rsmq.UnsetVt)

func (*RedisSMQ) CreateQueue

func (rsmq *RedisSMQ) CreateQueue(qname string, vt uint, delay uint, maxsize int) error

CreateQueue creates a new queue with given attributes to create new queue with default attributes:

err:=redisRsmq.CreateQueue(qname,rsmq.UnsetVt,rsmq.UnsetDelay,rsmq.UnsetMaxsize)

func (*RedisSMQ) DeleteMessage

func (rsmq *RedisSMQ) DeleteMessage(qname string, id string) error

DeleteMessage deletes message in queue

func (*RedisSMQ) DeleteQueue

func (rsmq *RedisSMQ) DeleteQueue(qname string) error

DeleteQueue deletes queue

func (*RedisSMQ) GetQueueAttributes

func (rsmq *RedisSMQ) GetQueueAttributes(qname string) (*QueueAttributes, error)

GetQueueAttributes returns queue attributes

func (*RedisSMQ) ListQueues

func (rsmq *RedisSMQ) ListQueues() ([]string, error)

ListQueues returns the slice consist of the existing queues

func (*RedisSMQ) PopMessage

func (rsmq *RedisSMQ) PopMessage(qname string) (*QueueMessage, error)

PopMessage pop message from queue

func (*RedisSMQ) Quit

func (rsmq *RedisSMQ) Quit() error

Quit closes redis client

func (*RedisSMQ) ReceiveMessage

func (rsmq *RedisSMQ) ReceiveMessage(qname string, vt uint) (*QueueMessage, error)

ReceiveMessage receives message from the queue

func (*RedisSMQ) SendMessage

func (rsmq *RedisSMQ) SendMessage(qname string, message string, delay uint, opts ...SendMessageOption) (string, error)

SendMessage sends a message to the queue. If a custom ID is provided via WithMessageID option: - The message will use that ID instead of generating a new one - If a message with that ID already exists, it will be overridden - The msg.Sent timestamp will not reflect the actual send time for overridden messages - Message timing is controlled by the delay parameter, not by the ID's timestamp

func (*RedisSMQ) SetQueueAttributes

func (rsmq *RedisSMQ) SetQueueAttributes(qname string, vt uint, delay uint, maxsize int) (*QueueAttributes, error)

SetQueueAttributes sets queue attributes to not change some attributes:

queAttrib,err:=redisRsmq.CreateQueue(qname,rsmq.UnsetVt,rsmq.UnsetDelay,newMaxsize)

type SendMessageOption

type SendMessageOption func(*sendMessageOptions)

SendMessageOption is a function that configures sendMessageOptions

func WithMessageID

func WithMessageID(id string) SendMessageOption

WithMessageID returns a SendMessageOption that sets a custom message ID