task

Documentation

Index

• Constants
• func NewTaskExecuteMessage(name string, args []byte, payloadType string) message.Message
• func NewTaskExecutedMessage(id xid.ID, result any, err error) message.Message
• func NewTaskExecutingMessage(manifestID, futureID xid.ID) message.Message
• func NewWorkerAnnounceMessage(workerID xid.ID, tasks ...string) message.Message
• type Controller
  • func NewController(bus message.Bus, kv kv.KV) *Controller
  • func (c *Controller) Dispatch(ctx context.Context, manifest Manifest) error
  • func (c *Controller) Producer() async.Manifest
  • func (c *Controller) Workers() []WorkerInfo
• type ExecuteMessage
• type ExecutedMessage
• type ExecutingMessage
• type Manifest
  • func JSON(name string, args any) (*Manifest, error)
  • func MsgPack(name string, args any) (*Manifest, error)
  • func (m *Manifest) Unmarshal(v any) error
• type Options
• type Task
  • func Of[I any, O any](name string, taskFn func(ctx context.Context, input I) (O, error)) Task
• type Worker
  • func NewWorker(bus message.Bus, tasks ...Task) *Worker
  • func (w *Worker) Announce(ctx context.Context) error
  • func (w *Worker) PollingExecutor(ctx context.Context) (*async.Manifest, error)
• type WorkerAnnounceMessage
• type WorkerInfo
• type WorkerLoad

Constants

const (
	MWorkerAnnounce = "wk.work.announce"
	MTaskExecute    = "wk.task.execute"
	MTaskExecuting  = "wk.task#executing"
	MTaskExecuted   = "wk.task#executed"
)

Functions

func NewTaskExecuteMessage

func NewTaskExecuteMessage(name string, args []byte, payloadType string) message.Message

func NewTaskExecutedMessage

func NewTaskExecutedMessage(id xid.ID, result any, err error) message.Message

func NewTaskExecutingMessage

func NewTaskExecutingMessage(manifestID, futureID xid.ID) message.Message

func NewWorkerAnnounceMessage

func NewWorkerAnnounceMessage(workerID xid.ID, tasks ...string) message.Message

Types

type Controller

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

func NewController

func NewController(bus message.Bus, kv kv.KV) *Controller

func (*Controller) Dispatch

func (c *Controller) Dispatch(ctx context.Context, manifest Manifest) error

func (*Controller) Producer

func (c *Controller) Producer() async.Manifest

func (*Controller) Workers

func (c *Controller) Workers() []WorkerInfo

type ExecuteMessage

type ExecuteMessage struct {
	Manifest
}

type ExecutedMessage

type ExecutedMessage struct {
	FutureID xid.ID
	Result   any
	Error    *string
}

type ExecutingMessage

type ExecutingMessage struct {
	ManifestID xid.ID
	FutureID   xid.ID
}

type Manifest

type Manifest struct {
	Name        string  `json:"name" yaml:"name"`
	Payload     []byte  `json:"payload" yaml:"payload"`
	PayloadType string  `json:"payloadType" yaml:"payloadType"`
	Options     Options `json:"options" yaml:"options"`
}

func JSON

func JSON(name string, args any) (*Manifest, error)

func MsgPack

func MsgPack(name string, args any) (*Manifest, error)

func (*Manifest) Unmarshal

func (m *Manifest) Unmarshal(v any) error

type Options

type Options struct {
	Priority int `json:"priority" yaml:"priority"`
}

type Task

type Task struct {
	Name          string
	AsyncManifest async.Manifest
}

func Of

func Of[I any, O any](name string, taskFn func(ctx context.Context, input I) (O, error)) Task

type Worker

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

func NewWorker

func NewWorker(bus message.Bus, tasks ...Task) *Worker

func (*Worker) Announce

func (w *Worker) Announce(ctx context.Context) error

func (*Worker) PollingExecutor

func (w *Worker) PollingExecutor(ctx context.Context) (*async.Manifest, error)

type WorkerAnnounceMessage

type WorkerAnnounceMessage struct {
	WorkerID xid.ID
	Tasks    []string
}

type WorkerInfo

type WorkerInfo struct {
	ID         string
	Load       map[string]int32
	LastUpdate time.Time
}

type WorkerLoad

type WorkerLoad struct {
	ID    string
	Tasks map[string]*atomic.Int32
}