taskqueue

Documentation

Index

• type GetOptions
• type MessageData
• type RedisTaskQueue
  • func NewRedisQueue(opts RedisTaskQueueOptions, logger *zap.SugaredLogger, ...) *RedisTaskQueue
  • func (q *RedisTaskQueue) Add(ctx context.Context, tasks []*TaskMessage) error
  • func (q *RedisTaskQueue) Delete(ctx context.Context, task *TaskMessage) error
  • func (q *RedisTaskQueue) Get(ctx context.Context, opts *GetOptions) ([]*TaskMessage, error)
  • func (q *RedisTaskQueue) Schedule(ctx context.Context, task *TaskMessage) error
  • func (q *RedisTaskQueue) Size(ctx context.Context) (int64, error)
  • func (q *RedisTaskQueue) Stats() map[string]interface{}
• type RedisTaskQueueOptions
• type TaskMessage
  • func (t *TaskMessage) MarshalData() ([]byte, error)
  • func (t *TaskMessage) String() string
  • func (t *TaskMessage) UnmarshalData(v interface{}) error
• type TaskQueue

Types

type GetOptions

type GetOptions struct {
	Count int64
}

type MessageData

type MessageData struct {
	EventID    string `json:"event_id"`
	EndpointId string `json:"endpoint_id"`
	Attempt    int    `json:"attempt"`
	Event      string `json:"event"`
}

type RedisTaskQueue

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

RedisTaskQueue use redis as queue implementation

func NewRedisQueue

func NewRedisQueue(opts RedisTaskQueueOptions, logger *zap.SugaredLogger, metrics *metrics.Metrics) *RedisTaskQueue

func (*RedisTaskQueue) Add

func (q *RedisTaskQueue) Add(ctx context.Context, tasks []*TaskMessage) error

func (*RedisTaskQueue) Delete

func (q *RedisTaskQueue) Delete(ctx context.Context, task *TaskMessage) error

func (*RedisTaskQueue) Get

func (q *RedisTaskQueue) Get(ctx context.Context, opts *GetOptions) ([]*TaskMessage, error)

func (*RedisTaskQueue) Schedule

func (q *RedisTaskQueue) Schedule(ctx context.Context, task *TaskMessage) error

func (*RedisTaskQueue) Size

func (q *RedisTaskQueue) Size(ctx context.Context) (int64, error)

func (*RedisTaskQueue) Stats

func (q *RedisTaskQueue) Stats() map[string]interface{}

type RedisTaskQueueOptions

type RedisTaskQueueOptions struct {
	QueueName         string
	QueueDataName     string
	VisibilityTimeout time.Duration
	Client            *redis.Client
}

type TaskMessage

type TaskMessage struct {
	ID          string
	ScheduledAt time.Time
	Data        interface{}
	// contains filtered or unexported fields
}

func (*TaskMessage) MarshalData

func (t *TaskMessage) MarshalData() ([]byte, error)

func (*TaskMessage) String

func (t *TaskMessage) String() string

func (*TaskMessage) UnmarshalData

func (t *TaskMessage) UnmarshalData(v interface{}) error

type TaskQueue

type TaskQueue interface {
	Add(ctx context.Context, tasks []*TaskMessage) error
	Get(ctx context.Context, opts *GetOptions) (tasks []*TaskMessage, err error)
	Delete(ctx context.Context, task *TaskMessage) error
	Size(ctx context.Context) (int64, error)
	Schedule(ctx context.Context, task *TaskMessage) error
	Stats() map[string]interface{}
}