wrapchinery

README

wrapchinery

Wrapped machinery package (v2) to send tasks more easily

Usage

Call wrapchinery.NewServer() instead of machinery.NewServer()

import (
github.com/gempages/wrapchinery
)

server := wrapchinery.NewServer(...)

Use wrapped functions:

// Create new worker
server.WrapNewWorker(concurrency int) *machinery.Worker
// Send a task
server.WrapSendTask(taskName string, delay time.Duration, retry int, args ...interface{}) (*result.AsyncResult, error)
// Send a task with context
server.WrapSendTaskWithContext(taskName string, ctx context.Context, delay time.Duration, retry int, args ...interface{}) (*result.AsyncResult, error)

Helper function to ease the pain of creating Signature:

wrapchinery.GetTaskSignature(taskName string, delay time.Duration, retry int, args ...interface{}) *tasks.Signature

Documentation

Index

• Constants
• func GetTaskSignature(cfg *TaskConfig, args ...interface{}) *tasks.Signature
• func InitTaskServer() error
• func InitTestTaskServer() error
• func NewWorker(concurrency int) *machinery.Worker
• func RegisterTasks(taskList map[string]interface{}) error
• func SendTask(ctx context.Context, task *TaskConfig, args ...interface{}) (*result.AsyncResult, error)
• func SendTaskWaitResult(ctx context.Context, task *TaskConfig, args ...interface{}) error
• func SetupLoggers()
• type TaskConfig
• type TaskServer
  • func Server() *TaskServer
  • func (m *TaskServer) WrapNewWorker(concurrency int) *machinery.Worker
  • func (m *TaskServer) WrapSendTask(cfg *TaskConfig, args ...interface{}) (*result.AsyncResult, error)
  • func (m *TaskServer) WrapSendTaskWithContext(ctx context.Context, cfg *TaskConfig, args ...interface{}) (*result.AsyncResult, error)

Constants

const ShopIDHeader = "shopID"

Functions

func GetTaskSignature

func GetTaskSignature(cfg *TaskConfig, args ...interface{}) *tasks.Signature

GetTaskSignature returns machinery's task signature object to use with SendTask and SendTaskWithContext functions

func InitTaskServer

func InitTaskServer() error

InitTaskServer initializes task server. Call this before registering tasks for workers.

func InitTestTaskServer

func InitTestTaskServer() error

InitTestTaskServer initializes task server for automation tests. Call this before registering tasks for workers.

func NewWorker

func NewWorker(concurrency int) *machinery.Worker

func RegisterTasks

func RegisterTasks(taskList map[string]interface{}) error

func SendTask

func SendTask(ctx context.Context, task *TaskConfig, args ...interface{}) (*result.AsyncResult, error)

func SendTaskWaitResult

func SendTaskWaitResult(ctx context.Context, task *TaskConfig, args ...interface{}) error

func SetupLoggers

func SetupLoggers()

Types

type TaskConfig

type TaskConfig struct {
	Name       string
	ShopID     uint64
	Delay      time.Duration
	RetryCount int
	OnSuccess  *TaskConfig
	OnError    *TaskConfig
}

type TaskServer

type TaskServer struct {
	machinery.Server
}

func Server

func Server() *TaskServer

func (*TaskServer) WrapNewWorker

func (m *TaskServer) WrapNewWorker(concurrency int) *machinery.Worker

WrapNewWorker creates a new machinery worker with a random UUID as tag

func (*TaskServer) WrapSendTask

func (m *TaskServer) WrapSendTask(cfg *TaskConfig, args ...interface{}) (*result.AsyncResult, error)

WrapSendTask calls machinery's SendTask function with task signature created using GetTaskSignature function

func (*TaskServer) WrapSendTaskWithContext

func (m *TaskServer) WrapSendTaskWithContext(ctx context.Context, cfg *TaskConfig, args ...interface{}) (*result.AsyncResult, error)