worker

Documentation

Index

• Constants
• Variables
• func Main(serverURL, topic string, maxJobSize int, workerName string, workerPort int) error
• type Client
  • func NewClient(url string) *Client
  • func NewClientWithHttpClient(url string, c *http.Client) *Client
  • func (c *Client) ReportJob(ctx context.Context, req *pb.ReportJobRequest) (res *pb.ReportJobResponse, err error)
  • func (c *Client) ReportJobDone(ctx context.Context, req *pb.ReportJobDoneRequest) (res *pb.ReportJobDoneResponse, err error)
  • func (c *Client) SubscribeJob(ctx context.Context, req *pb.SubscribeJobRequest) (res *pb.SubscribeJobResponse, err error)
• type Job
  • func NewJob(ctx context.Context, id string, jobConfig *config.Job) *Job
  • func (job *Job) Done() <-chan struct{}
  • func (job *Job) OnTaskChanged(tr *TaskRunner)
  • func (job *Job) OnTaskDone(tr *TaskRunner)
  • func (job *Job) OnTaskStateChange(handler func(Task))
  • func (job *Job) Run()
• type JobMessage
• type JobTask
  • func NewJobTask(state string) *JobTask
  • func (t *JobTask) State() string
  • func (t *JobTask) TaskName() string
• type Task
• type TaskRunFilter
  • func (f *TaskRunFilter) Filter(task Task, tasks []*c.Task) (matched []*c.Task, notmatched []*c.Task, err error)
  • func (f *TaskRunFilter) Name() string
• type TaskRunner
  • func NewTaskRunner(job *Job, task *config.Task, templateCtx map[string]interface{}) *TaskRunner
  • func (tr *TaskRunner) Cancel()
  • func (tr *TaskRunner) Err() error
  • func (tr *TaskRunner) Ok() bool
  • func (tr *TaskRunner) Output() string
  • func (tr *TaskRunner) Run()
  • func (tr *TaskRunner) StartEndTimes() []*time.Time
  • func (tr *TaskRunner) State() string
  • func (tr *TaskRunner) TaskName() string
• type Tasks
  • func (tasks Tasks) JSON() (result map[string]interface{})
• type Worker
  • func NewWorker(ctx context.Context, name string, serverURL, topic string, maxJobSize int) *Worker
  • func (w *Worker) Init() error
  • func (w *Worker) Run()
  • func (w *Worker) Stop()

Constants

const (
	EQ = iota
	NOTEQ
)

const (
	HTTP_GET  = "GET"
	HTTP_POST = "POST"
)

const (
	TASK_STATE_INIT    = "INIT"
	TASK_STATE_PROCESS = "PROCESS"
	TASK_STATE_DONE    = "DONE"
	TASK_STATE_CANCEL  = "CANCEL"
	TASK_STATE_ERROR   = "ERROR"

	TASK_EVENT_RUN     = "run"
	TASK_EVENT_SUCCESS = "success"
	TASK_EVENT_CANCEL  = "cancel"
	TASK_EVENT_ERROR   = "error"

	TASK_QUIT = "quit"
)

Variables

var (
	HTTPMethodNotSupport = errors.New("Http method is not supported")
)

var OPERATORS = []string{"==", "!="}

Functions

func Main

func Main(serverURL, topic string, maxJobSize int, workerName string, workerPort int) error

Types

type Client

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

func NewClient

func NewClient(url string) *Client

func NewClientWithHttpClient

func NewClientWithHttpClient(url string, c *http.Client) *Client

func (*Client) ReportJob

func (c *Client) ReportJob(ctx context.Context, req *pb.ReportJobRequest) (res *pb.ReportJobResponse, err error)

func (*Client) ReportJobDone

func (c *Client) ReportJobDone(ctx context.Context, req *pb.ReportJobDoneRequest) (res *pb.ReportJobDoneResponse, err error)

func (*Client) SubscribeJob

func (c *Client) SubscribeJob(ctx context.Context, req *pb.SubscribeJobRequest) (res *pb.SubscribeJobResponse, err error)

type Job

type Job struct {
	ID string

	Tasks Tasks
	// contains filtered or unexported fields
}

func NewJob

func NewJob(ctx context.Context, id string, jobConfig *config.Job) *Job

func (*Job) Done

func (job *Job) Done() <-chan struct{}

func (*Job) OnTaskChanged

func (job *Job) OnTaskChanged(tr *TaskRunner)

func (*Job) OnTaskDone

func (job *Job) OnTaskDone(tr *TaskRunner)

func (*Job) OnTaskStateChange

func (job *Job) OnTaskStateChange(handler func(Task))

func (*Job) Run

func (job *Job) Run()

type JobMessage

type JobMessage struct {
	Tasks []*config.Task `json:tasks`
}

type JobTask

type JobTask struct {
	config.Task
	// contains filtered or unexported fields
}

func NewJobTask

func NewJobTask(state string) *JobTask

func (*JobTask) State

func (t *JobTask) State() string

func (*JobTask) TaskName

func (t *JobTask) TaskName() string

type Task

type Task interface {
	TaskName() string
	State() string
	Ok() bool
	Err() error
	Output() string
	StartEndTimes() []*time.Time
}

type TaskRunFilter

type TaskRunFilter struct{}

func (*TaskRunFilter) Filter

func (f *TaskRunFilter) Filter(task Task, tasks []*c.Task) (matched []*c.Task, notmatched []*c.Task, err error)

func (*TaskRunFilter) Name

func (f *TaskRunFilter) Name() string

type TaskRunner

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

func NewTaskRunner

func NewTaskRunner(job *Job, task *config.Task, templateCtx map[string]interface{}) *TaskRunner

func (*TaskRunner) Cancel

func (tr *TaskRunner) Cancel()

func (*TaskRunner) Err

func (tr *TaskRunner) Err() error

func (*TaskRunner) Ok

func (tr *TaskRunner) Ok() bool

func (*TaskRunner) Output

func (tr *TaskRunner) Output() string

func (*TaskRunner) Run

func (tr *TaskRunner) Run()

func (*TaskRunner) StartEndTimes

func (tr *TaskRunner) StartEndTimes() []*time.Time

func (*TaskRunner) State

func (tr *TaskRunner) State() string

func (*TaskRunner) TaskName

func (tr *TaskRunner) TaskName() string

type Tasks

type Tasks map[string]Task

func (Tasks) JSON

func (tasks Tasks) JSON() (result map[string]interface{})

type Worker

type Worker struct {
	Name      string
	Topic     string
	ServerURL string
	// contains filtered or unexported fields
}

func NewWorker

func NewWorker(ctx context.Context, name string, serverURL, topic string, maxJobSize int) *Worker

func (*Worker) Init

func (w *Worker) Init() error

func (*Worker) Run

func (w *Worker) Run()

func (*Worker) Stop

func (w *Worker) Stop()