models

Documentation

Index

• Variables
• type FilterCriteria
• type Message
  • func (m *Message) Base64Decode() []byte
  • func (m *Message) IsB64() bool
  • func (m *Message) IsJSON() bool
  • func (m *Message) Status() MessageStatus
• type MessageStatus
  • func (s MessageStatus) String() string
• type Queue
• type QueueProperties
• type QueueStats
• type TenantManager

Variables

var ErrQueueExists = errors.New("Queue already exists")

Types

type FilterCriteria

type FilterCriteria struct {
	MessageID int64

	// 0 means unbounded
	DeliverAtStart int
	DeliverAtEnd   int

	// status is an OR filter
	Status []MessageStatus

	// kv is an AND filter
	KV map[string]string

	// Smallest message ID to return. Message IDs are Snowflake IDs
	MinMessageID int64

	// How many message IDs to return
	Limit int
}

type Message

type Message struct {
	ID       int64 `db:"id"`
	TenantID int64 `db:"tenant_id"`
	QueueID  int64 `db:"queue_id"`

	DeliverAt   int `db:"deliver_at"`
	DeliveredAt int `db:"delivered_at"`
	Tries       int `db:"tries"`
	MaxTries    int `db:"max_tries"`

	Message   []byte `db:"message"`
	KeyValues map[string]string
}

func (*Message) Base64Decode

func (m *Message) Base64Decode() []byte

func (*Message) IsB64

func (m *Message) IsB64() bool

func (*Message) IsJSON

func (m *Message) IsJSON() bool

func (*Message) Status

func (m *Message) Status() MessageStatus

type MessageStatus

type MessageStatus uint8

const (
	MessageStatusQueued   MessageStatus = 1
	MessageStatusDequeued MessageStatus = 2
	MessageStatusFailed   MessageStatus = 3
)

func (MessageStatus) String

func (s MessageStatus) String() string

type Queue

type Queue interface {
	GetQueue(tenantId int64, queueName string) (QueueProperties, error)
	CreateQueue(tenantId int64, properties QueueProperties) error
	UpdateQueue(tenantId int64, queue string, properties QueueProperties) error
	DeleteQueue(tenantId int64, queue string) error
	ListQueues(tenantId int64) ([]string, error)

	Enqueue(tenantId int64, queue string, message string, kv map[string]string, delay int) (int64, error)
	Dequeue(tenantId int64, queue string, numToDequeue int, requeueIn int) ([]*Message, error)

	Peek(tenantId int64, queue string, messageId int64) *Message
	Stats(tenantId int64, queue string) QueueStats
	Filter(tenantId int64, queue string, filterCriteria FilterCriteria) []int64

	Delete(tenantId int64, queue string, messageId int64) error

	Shutdown() error
}

type QueueProperties

type QueueProperties struct {
	Name              string
	RateLimit         float64
	MaxRetries        int
	VisibilityTimeout int
}

type QueueStats

type QueueStats struct {
	Counts        map[MessageStatus]int
	TotalMessages int
}

type TenantManager

type TenantManager interface {
	GetTenant(r *http.Request) (int64, error)
	GetAWSSecretKey(accessKey string, region string) (tenantId int64, secretKey string, err error)
}