job

Documentation

Index

• func SetLogger(logger *zap.Logger)
• type DefaultStorage
  • func (s *DefaultStorage) LockJob(j *Job) error
  • func (s *DefaultStorage) UpdateJob(j *Job) error
• type Job
  • func ReadJob(rec []interface{}) *Job
  • func (j *Job) Run()
  • func (j *Job) SetTask(f JobFunc)
• type JobFunc
• type JobID
  • func NewJobID() JobID
• type Payload
• type Queue
  • func (q *Queue) AddJob(j *Job)
  • func (queue *Queue) Start()
  • func (queue *Queue) StopQueue()
• type TarantoolStorage
  • func (s *TarantoolStorage) Init(uri string) error
  • func (s *TarantoolStorage) LockJob(j *Job) error
  • func (s *TarantoolStorage) RecentJobs() ([]*Job, error)
  • func (s *TarantoolStorage) UpdateJob(j *Job) error
• type TasksStorage

Functions

func SetLogger

func SetLogger(logger *zap.Logger)

Types

type DefaultStorage

type DefaultStorage struct {
}

func (*DefaultStorage) LockJob

func (s *DefaultStorage) LockJob(j *Job) error

func (*DefaultStorage) UpdateJob

func (s *DefaultStorage) UpdateJob(j *Job) error

type Job

type Job struct {
	ID       JobID
	Name     string // A name of a job supplied when registering it at worker
	WorkerID string // A worker where this job is running

	ArrivedAt  time.Time `msgpack:"-"` // Time when a message is received from a Tasks broker
	LockedAt   time.Time `msgpack:"-"` // Time when this Job is locked for execution by a worker
	StartedAt  time.Time `msgpack:"-"` // Time when this Job is actually run
	FinishedAt time.Time `msgpack:"-"` // Time when this Job is marked as finished

	// queue string
	// payload Payload
	Args []interface{} // `json:"args"`
	// contains filtered or unexported fields
}

func ReadJob

func ReadJob(rec []interface{}) *Job

func (*Job) Run

func (j *Job) Run()

Run executes a user-defined function - the task itself. This function is blocking. This function should never crash no matter what user executes.

func (*Job) SetTask

func (j *Job) SetTask(f JobFunc)

type JobFunc

type JobFunc func(*Job, ...interface{}) error

type JobID

type JobID = string

type JobID uuid.UUID

func NewJobID

func NewJobID() JobID

type Payload

type Payload struct {
	Class string        `json:"class"`
	Args  []interface{} `json:"args"`
}

type Queue

type Queue struct {
	Name        string
	Concurrency int
	Pool        chan *Job
	Stopped     bool
}

func (*Queue) AddJob

func (q *Queue) AddJob(j *Job)

AddJob sends a job into a queue's "Jobs" channel If channel is not full - the job will start immediately

func (*Queue) Start

func (queue *Queue) Start()

Start creates "queue workers". It is a blocking function. It waits for all of the "workers" to finish.

func (*Queue) StopQueue

func (queue *Queue) StopQueue()

type TarantoolStorage

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

func (*TarantoolStorage) Init

func (s *TarantoolStorage) Init(uri string) error

Init prepares (creates) a "table" for tasks.

In Tarantool a table is called Space.

func (*TarantoolStorage) LockJob

func (s *TarantoolStorage) LockJob(j *Job) error

LockJob "requests a permission" for a worker to run the job.

It must insert a record into a consistent storage and if it succeeds - then a worker can start a job.

If it fails - then this worker MUST NOT start the job.

This function is used when a worker receives a new message from TasksBroker and decides wether or not it can start this job.

func (*TarantoolStorage) RecentJobs

func (s *TarantoolStorage) RecentJobs() ([]*Job, error)

func (*TarantoolStorage) UpdateJob

func (s *TarantoolStorage) UpdateJob(j *Job) error

type TasksStorage

type TasksStorage interface {
	// LockJob "requests a permission" for a worker to run the job.
	//
	// It must insert a record into a consistent storage
	// and if it succeeds - then a worker can start a job.
	//
	// If it fails - then this worker MUST NOT start the job.
	//
	// This function is used when a worker receives a new message from
	// TasksBroker and decides wether or not it can start this job.
	LockJob(j *Job) error

	UpdateJob(j *Job) error
}

var Storage TasksStorage