repository

Documentation

Overview

Package repository is a package that handles database related sources

Index

• Constants
• type AdminTXFunc
• type DB
  • func New(opt *Opt) (*DB, error)
  • func (db *DB) GetExecutableTask(ctx context.Context, concurrency int) ([]ExecutableTask, error)
  • func (db *DB) GetRegisterTask(ctx context.Context, taskId string) ([]task, error)
  • func (db *DB) GetTaskProgramName(ctx context.Context, task *ExecutableTask) (string, error)
  • func (db *DB) InsertExecutableTasks(ctx context.Context, taskId string) (bool, error)
  • func (db *DB) ReadWriteTransaction(ctx context.Context, f AdminTXFunc) error
  • func (db *DB) UpdateExecutableTasksFinished(ctx context.Context, e *ExecutableTask) error
  • func (db *DB) UpdateExecutableTasksRunning(ctx context.Context, e *ExecutableTask) (bool, error)
  • func (db *DB) UpdateExecutableTasksSuspended(ctx context.Context, e *ExecutableTask) error
• type ExecutableTask
• type Opt
• type Row

Constants

const (
	// DefaultDBName is the default name of postgres database.
	DefaultDBName = "flower"
	// DefaultDBUserName is the default postgres user name.
	DefaultDBUserName = "flower"
	// DefaultDBHostname is the default postgres host name.
	DefaultDBHostname = "localhost"
	// DefaultDBPort is the default postgres port.
	DefaultDBPort = "5432"
	// ConnectTimeout is the default timeout of the connection to the postgres server.
	ConnectTimeout = 5

	// ExecStatusWait is the status waiting to be executed.
	ExecStatusWait = 0
	// ExecStatusRunning is the running status.
	ExecStatusRunning = 1
	// ExecStatusSuspend is the suspended status.
	ExecStatusSuspend = 2
	// ExecStatusFinish is the completed status.
	ExecStatusFinish = 3
	// ExecStatusIgnore is the status to be ignored.
	ExecStatusIgnore = 9
)

Types

type AdminTXFunc

type AdminTXFunc func(context.Context, *adminTX) error

AdminTXFunc is the signature for functions passed to ReadWriteTransaction.

type DB

type DB struct {
	*sql.DB
}

DB represents a Database handler.

func New

func New(opt *Opt) (*DB, error)

New creates the DB object.

func (*DB) GetExecutableTask

func (db *DB) GetExecutableTask(ctx context.Context, concurrency int) ([]ExecutableTask, error)

GetExecutableTask is the main method. From the series of tasks waiting to be registered for each task flow Id, resolve the dependency of the execution order. Get the executable task and register the task to be executed in the channel to run task.

func (*DB) GetRegisterTask

func (db *DB) GetRegisterTask(ctx context.Context, taskId string) ([]task, error)

GetExecutableTask is a function to get a list of tasks to be registered from taskId.

func (*DB) GetTaskProgramName

func (db *DB) GetTaskProgramName(ctx context.Context, task *ExecutableTask) (string, error)

GetTaskProgramName gets the name of the program to be executed from taskId and taskSeq.

func (*DB) InsertExecutableTasks

func (db *DB) InsertExecutableTasks(ctx context.Context, taskId string) (bool, error)

InsertExecutableTasks registers the task waiting to be executed from the called taskId.

func (*DB) ReadWriteTransaction

func (db *DB) ReadWriteTransaction(ctx context.Context, f AdminTXFunc) error

ReadWriteTransaction creates a transaction, and runs f with it. Some storage implementations may retry aborted transactions, so f MUST be idempotent.

func (*DB) UpdateExecutableTasksFinished

func (db *DB) UpdateExecutableTasksFinished(ctx context.Context, e *ExecutableTask) error

UpdateExecutableTasksFinished updates the status of tasks to finished.

func (*DB) UpdateExecutableTasksRunning

func (db *DB) UpdateExecutableTasksRunning(ctx context.Context, e *ExecutableTask) (bool, error)

UpdateExecutableTasksRunning updates task status to running.

func (*DB) UpdateExecutableTasksSuspended

func (db *DB) UpdateExecutableTasksSuspended(ctx context.Context, e *ExecutableTask) error

UpdateExecutableTasksSuspended updates the status of a task to suspended.

type ExecutableTask

type ExecutableTask struct {
	TaskFlowId  string            `db:"task_flow_id"`
	TaskExecSeq int               `db:"task_exec_seq"`
	TaskId      string            `db:"task_id"`
	TaskSeq     int               `db:"task_seq"`
	Params      map[string]string `db:"parameters"`
}

ExecutableTask is the struct of executable tasks.

type Opt

type Opt struct {
	DBName   string
	User     string
	Password string
	Host     string
	Port     string
	SSLMode  string
}

Opt are options for database connection. https://godoc.org/github.com/lib/pq

type Row

type Row interface {
	Scan(dest ...interface{}) error
}

Row defines a common interface between sql.Row and sql.Rows(!)