sqlite

Documentation

Index

• type KV
• type Message
  • func (message *Message) ToModel() *models.Message
• type Queue
• type RateLimit
• type SQLiteQueue
  • func NewSQLiteQueue(cfg config.SQLiteConfig) *SQLiteQueue
  • func (q *SQLiteQueue) CreateQueue(tenantId int64, properties models.QueueProperties) error
  • func (q *SQLiteQueue) Delete(tenantId int64, queueName string, messageId int64) error
  • func (q *SQLiteQueue) DeleteQueue(tenantId int64, queueName string) error
  • func (q *SQLiteQueue) Dequeue(tenantId int64, queueName string, numToDequeue int, requeueIn int) ([]*models.Message, error)
  • func (q *SQLiteQueue) Enqueue(tenantId int64, queueName string, message string, kv map[string]string, ...) (int64, error)
  • func (q *SQLiteQueue) Filter(tenantId int64, queueName string, filterCriteria models.FilterCriteria) []int64
  • func (q *SQLiteQueue) GetQueue(tenantId int64, queueName string) (models.QueueProperties, error)
  • func (q *SQLiteQueue) ListQueues(tenantId int64) ([]string, error)
  • func (q *SQLiteQueue) Peek(tenantId int64, queueName string, messageId int64) *models.Message
  • func (q *SQLiteQueue) Shutdown() error
  • func (q *SQLiteQueue) Stats(tenantId int64, queueName string) models.QueueStats
  • func (q *SQLiteQueue) UpdateQueue(tenantId int64, queueName string, properties models.QueueProperties) error

Types

type KV

type KV struct {
	TenantID  int64  `gorm:"not null;index:idx_kv,priority:1"`
	QueueID   int64  `gorm:"not null;index:idx_kv,priority:2"`
	MessageID int64  `gorm:"not null;index:idx_kv,priority:3"`
	K         string `gorm:"not null"`
	V         string `gorm:"not null"`
}

type Message

type Message struct {
	ID       int64 `gorm:"primaryKey;autoIncrement:false"`
	TenantID int64 `gorm:"not null;index:idx_message,priority:1;not null"`
	QueueID  int64 `gorm:"not null;index:idx_message,priority:2;not null"`

	DeliverAt   int64 `gorm:"not null;index:idx_message,priority:3;not null"`
	DeliveredAt int64 `gorm:"not null;index:idx_message,priority:4;not null"`
	Tries       int   `gorm:"not null;index:idx_message,priority:5;not null"`
	MaxTries    int   `gorm:"not null;index:idx_message,priority:6;not null"`

	Message string `gorm:"not null"`

	KV []KV `gorm:"foreignKey:TenantID,QueueID,MessageID;references:TenantID,QueueID,ID"`
}

func (*Message) ToModel

func (message *Message) ToModel() *models.Message

type Queue

type Queue struct {
	ID                int64   `gorm:"primaryKey;autoIncrement:false"`
	TenantID          int64   `gorm:"not null;index:idx_queue_name,priority:1,unique"`
	Name              string  `gorm:"not null;index:idx_queue_name,priority:2"`
	RateLimit         float64 `gorm:"not null"`
	MaxRetries        int     `gorm:"not null"`
	VisibilityTimeout int     `gorm:"not null"`

	Messages []Message `gorm:"foreignKey:QueueID;references:ID"`
}

type RateLimit

type RateLimit struct {
	TenantID int64 `gorm:"not null;index:idx_ratelimit,priority:1,unique"`
	QueueID  int64 `gorm:"not null;index:idx_ratelimit,priority:2"`
	Ts       int64 `gorm:"not null;index:idx_ratelimit,priority:3"`
	N        int   `gorm:"not null;default:0"`
}

type SQLiteQueue

type SQLiteQueue struct {
	Filename string
	DBG      *gorm.DB
	Mu       *sync.Mutex
	// contains filtered or unexported fields
}

func NewSQLiteQueue

func NewSQLiteQueue(cfg config.SQLiteConfig) *SQLiteQueue

func (*SQLiteQueue) CreateQueue

func (q *SQLiteQueue) CreateQueue(tenantId int64, properties models.QueueProperties) error

func (*SQLiteQueue) Delete

func (q *SQLiteQueue) Delete(tenantId int64, queueName string, messageId int64) error

func (*SQLiteQueue) DeleteQueue

func (q *SQLiteQueue) DeleteQueue(tenantId int64, queueName string) error

func (*SQLiteQueue) Dequeue

func (q *SQLiteQueue) Dequeue(tenantId int64, queueName string, numToDequeue int, requeueIn int) ([]*models.Message, error)

func (*SQLiteQueue) Enqueue

func (q *SQLiteQueue) Enqueue(tenantId int64, queueName string, message string, kv map[string]string, delay int) (int64, error)

func (*SQLiteQueue) Filter

func (q *SQLiteQueue) Filter(tenantId int64, queueName string, filterCriteria models.FilterCriteria) []int64

func (*SQLiteQueue) GetQueue

func (q *SQLiteQueue) GetQueue(tenantId int64, queueName string) (models.QueueProperties, error)

func (*SQLiteQueue) ListQueues

func (q *SQLiteQueue) ListQueues(tenantId int64) ([]string, error)

func (*SQLiteQueue) Peek

func (q *SQLiteQueue) Peek(tenantId int64, queueName string, messageId int64) *models.Message

func (*SQLiteQueue) Shutdown

func (q *SQLiteQueue) Shutdown() error

func (*SQLiteQueue) Stats

func (q *SQLiteQueue) Stats(tenantId int64, queueName string) models.QueueStats

func (*SQLiteQueue) UpdateQueue

func (q *SQLiteQueue) UpdateQueue(tenantId int64, queueName string, properties models.QueueProperties) error