xjm

Documentation

Index

• Constants
• Variables
• func Decode(p string, v any) error
• func Encode(v any) string
• type Job
  • func (j *Job) IsAborted() bool
  • func (j *Job) IsCompleted() bool
  • func (j *Job) IsPending() bool
  • func (j *Job) IsRunning() bool
  • func (j *Job) Params() (m map[string]any)
  • func (j *Job) Results() (m map[string]any)
  • func (j *Job) String() string
• type JobLog
  • func (jl *JobLog) String() string
• type JobLogWriter
  • func NewJobLogWriter(jmr JobManager, jid int64) *JobLogWriter
  • func (jlw *JobLogWriter) Close()
  • func (jlw *JobLogWriter) Flush()
  • func (jlw *JobLogWriter) Write(le *log.Event) (err error)
• type JobManager
• type JobRunner
  • func NewJobRunner(jmr JobManager, jid, rid int64, logger ...log.Logger) *JobRunner
  • func (jr *JobRunner) Abort(reason string) error
  • func (jr *JobRunner) Checkout() error
  • func (jr *JobRunner) Complete(result string) error
  • func (jr *JobRunner) GetJob() (*Job, error)
  • func (jr *JobRunner) JobID() int64
  • func (jr *JobRunner) Ping() error
  • func (jr *JobRunner) PingAborted() bool
  • func (jr *JobRunner) RunnerID() int64
  • func (jr *JobRunner) Running(state string) error

Constants

const (
	JobStatusAborted   = "A"
	JobStatusCompleted = "C"
	JobStatusPending   = "P"
	JobStatusRunning   = "R"
)

Variables

var (
	JobPendingRunning   = []string{JobStatusPending, JobStatusRunning}
	JobAbortedCompleted = []string{JobStatusAborted, JobStatusCompleted}
)

var (
	ErrJobAborted   = errors.New("job aborted")
	ErrJobCompleted = errors.New("job completed")
	ErrJobCheckout  = errors.New("job checkout failed: job running or missing")
	ErrJobExisting  = errors.New("job existing")
	ErrJobMissing   = errors.New("job missing")
	ErrJobPing      = errors.New("job ping failed")
)

Functions

func Decode

func Decode(p string, v any) error

func Encode

func Encode(v any) string

Types

type Job

type Job struct {
	ID        int64     `gorm:"not null;primaryKey;autoIncrement" uri:"id" form:"id" json:"id,omitempty"`
	RID       int64     `gorm:"column:rid;not null" form:"rid" json:"rid,omitempty"` // job runner id
	Name      string    `gorm:"size:250;not null;index" json:"name,omitempty"`
	Status    string    `gorm:"size:1;not null" form:"status" json:"status,omitempty"`
	File      string    `gorm:"not null" json:"file,omitempty"`
	Param     string    `gorm:"not null" json:"param,omitempty"`
	State     string    `gorm:"not null" form:"state" json:"state,omitempty"`
	Result    string    `gorm:"not null" json:"result,omitempty"`
	Error     string    `gorm:"not null" json:"error,omitempty"`
	CreatedAt time.Time `gorm:"<-:create;not null" json:"created_at,omitempty"`
	UpdatedAt time.Time `gorm:"not null" json:"updated_at,omitempty"`
}

func (*Job) IsAborted

func (j *Job) IsAborted() bool

func (*Job) IsCompleted

func (j *Job) IsCompleted() bool

func (*Job) IsPending

func (j *Job) IsPending() bool

func (*Job) IsRunning

func (j *Job) IsRunning() bool

func (*Job) Params

func (j *Job) Params() (m map[string]any)

func (*Job) Results

func (j *Job) Results() (m map[string]any)

func (*Job) String

func (j *Job) String() string

type JobLog

type JobLog struct {
	ID      int64     `gorm:"not null;primaryKey;autoIncrement" uri:"id" form:"id" json:"id,omitempty"`
	JID     int64     `gorm:"column:jid;not null;index:idx_job_logs_jid" uri:"jid" form:"jid" json:"jid,omitempty"`
	Time    time.Time `gorm:"not null" json:"time,omitempty"`
	Level   string    `gorm:"size:1;not null" json:"level,omitempty"`
	Message string    `gorm:"not null" json:"message,omitempty"`
}

func (*JobLog) String

func (jl *JobLog) String() string

type JobLogWriter

type JobLogWriter struct {
	log.LogFilter
	log.BatchWriter
	// contains filtered or unexported fields
}

JobLogWriter implements log Writer Interface and writes messages to terminal.

func NewJobLogWriter

func NewJobLogWriter(jmr JobManager, jid int64) *JobLogWriter

func (*JobLogWriter) Close

func (jlw *JobLogWriter) Close()

Close implementing method. empty.

func (*JobLogWriter) Flush

func (jlw *JobLogWriter) Flush()

Flush implementing method. empty.

func (*JobLogWriter) Write

func (jlw *JobLogWriter) Write(le *log.Event) (err error)

Write write message in console.

type JobManager

type JobManager interface {
	// CountJobLogs count job logs
	CountJobLogs(jid int64, levels ...string) (int64, error)

	// GetJobLogs get job logs
	// set levels to ("I", "W", "E", "F") to filter DEBUG/TRACE logs
	GetJobLogs(jid int64, min, max int64, asc bool, limit int, levels ...string) ([]*JobLog, error)

	// AddJobLogs append job logs
	AddJobLogs([]*JobLog) error

	// GetJob get job detail
	GetJob(jid int64) (*Job, error)

	// FindJob find the latest job by name.
	// status: status to filter.
	FindJob(name string, asc bool, status ...string) (*Job, error)

	// FindJobs find jobs by name.
	// status: status to filter.
	FindJobs(name string, start, limit int, asc bool, status ...string) ([]*Job, error)

	// IterJobs find jobs by name and iterate.
	// status: status to filter.
	IterJobs(it func(job *Job) error, name string, start, limit int, asc bool, status ...string) error

	// AppendJob append a pendding job
	AppendJob(name, file, param string) (int64, error)

	// AbortJob abort the job
	AbortJob(jid int64, reason string) error

	// CompleteJob complete the job
	CompleteJob(jid int64, result string) error

	// CheckoutJob checkout the job to the running status
	CheckoutJob(jid, rid int64) error

	// PingJob update the job updated_at to now
	PingJob(jid, rid int64) error

	// RunningJob update the running job state
	RunningJob(jid, rid int64, state string) error

	// ReappendJobs reappend the interrupted runnings job to the pennding status
	ReappendJobs(before time.Time) (int64, error)

	// StartJobs start to run jobs
	StartJobs(limit int, run func(*Job)) error

	// CleanOutdatedJobs delete outdated jobs
	CleanOutdatedJobs(before time.Time) (int64, int64, error)
}

type JobRunner

type JobRunner struct {
	Log *log.Log

	PingAfter time.Duration // Ping after duration
	// contains filtered or unexported fields
}

func NewJobRunner

func NewJobRunner(jmr JobManager, jid, rid int64, logger ...log.Logger) *JobRunner

NewJobRunner create a JobRunner

func (*JobRunner) Abort

func (jr *JobRunner) Abort(reason string) error

func (*JobRunner) Checkout

func (jr *JobRunner) Checkout() error

func (*JobRunner) Complete

func (jr *JobRunner) Complete(result string) error

func (*JobRunner) GetJob

func (jr *JobRunner) GetJob() (*Job, error)

func (*JobRunner) JobID

func (jr *JobRunner) JobID() int64

func (*JobRunner) Ping

func (jr *JobRunner) Ping() error

func (*JobRunner) PingAborted

func (jr *JobRunner) PingAborted() bool

func (*JobRunner) RunnerID

func (jr *JobRunner) RunnerID() int64

func (*JobRunner) Running

func (jr *JobRunner) Running(state string) error