task_queue

README

Local Task Queue - use a postgres database and depends on go channel based custom implementation. Remote Task Queue - can be configured to use it with either RabbitMQ or Redis (It's better for reliability and scalability)

Documentation

Index

• type ArgumentType
• type Client
  • func NewClient(options Options) (Client, error)
• type EnqueuedTask
• type Options
• type RemoteQueueType
• type ServiceType
• type WorkerFunctionType

Types

type ArgumentType

type ArgumentType interface{}

type Client

type Client interface {
	// RegisterFunction registers a consumer function for a queue
	RegisterFunction(queueName string, function WorkerFunctionType) error
	// EnqueueTask enqueues a task to a queue
	EnqueueTask(queueName string, argument ArgumentType) error
	// StartConsumers is a blocking function that starts the consumers for all the registered queues
	StartConsumers(nowait bool) error
	// WaitForConsumers is a blocking function that waits for all the consumers to finish
	WaitForConsumers()
	// EnqueueProcessingQueueExpiredTask enqueues tasks from processing queue to the original queue
	EnqueueProcessingQueueExpiredTask() error
	// PurgeQueue purges all the messages from a queue
	PurgeQueue(queueName string) error
	// ListMessages returns the messages of a queue
	// Note: Should be called when no consumers are running
	ListMessages(queueName string) ([]string, error)
}

func NewClient

func NewClient(options Options) (Client, error)

type EnqueuedTask

type EnqueuedTask struct {
	ID        int `gorm:"primaryKey"`
	QueueName string
	Body      string
	Hash      string
}

EnqueuedTask holds the task details Add EnqueuedTask to gorm migration

type Options

type Options struct {
	Type                ServiceType
	NoOfWorkersPerQueue int
	MaxMessagesPerQueue int      // only applicable for local task queue
	DbClient            *gorm.DB // only applicable for local task queue
	// Extra options for remote task queue
	RemoteQueueType RemoteQueueType
	// Redis specific options
	RedisClient *redis.Client
	// AMQP specific options
	AMQPUri        string
	AMQPVhost      string
	AMQPClientName string
}

type RemoteQueueType

type RemoteQueueType string

const (
	AmqpQueue       RemoteQueueType = "amqp"
	RedisQueue      RemoteQueueType = "redis"
	NoneRemoteQueue RemoteQueueType = "none"
)

type ServiceType

type ServiceType string

const (
	Local  ServiceType = "local"
	Remote ServiceType = "remote"
)

type WorkerFunctionType

type WorkerFunctionType interface{}