workerpoolpg

Documentation

Overview

Package workerpoolpg - очереди обработки бд запросов

Index

• Variables
• func MakeJob(ctx context.Context, options pgx.TxOptions, worker DatabaseWorker, ...) error
• func MakeJobWithResponse(ctx context.Context, options pgx.TxOptions, worker DatabaseWorkerWithResponse, ...) (proto.Message, error)
• func MakeJobWithResult(ctx context.Context, options pgx.TxOptions, worker DatabaseWorkerWithResult, ...) (structs.Result, error)
• type DatabaseWorker
• type DatabaseWorkerWithResponse
• type DatabaseWorkerWithResult
• type JobResultErr
  • func WrapError(err error) *JobResultErr
• type JobResultProto
  • func WrapMessage(msg proto.Message) *JobResultProto
• type Queue
  • func NewQueue(size int) *Queue
  • func (q *Queue) Close()
  • func (q *Queue) GetQueue() chan *Task
  • func (q *Queue) MakeJob(t *Task) error
• type Task
• type Worker
  • func NewWorker(queue chan *Task) *Worker
  • func (w *Worker) Run()

Variables

var (
	TimeoutErr = errors.New("timeout")
)

Functions

func MakeJob

func MakeJob(ctx context.Context, options pgx.TxOptions, worker DatabaseWorker,
	timeout time.Duration) error

MakeJob - выполнение работы с транзакцией с таймаутом

func MakeJobWithResponse

func MakeJobWithResponse(ctx context.Context, options pgx.TxOptions, worker DatabaseWorkerWithResponse,
	timeout time.Duration) (proto.Message, error)

MakeJobWithResponse - выполнение работы с транзакцией с прото результатом с таймаутом

func MakeJobWithResult

func MakeJobWithResult(ctx context.Context, options pgx.TxOptions, worker DatabaseWorkerWithResult,
	timeout time.Duration) (structs.Result, error)

MakeJobWithResult - выполнение работы с транзакцией с structs.Result результатом(где подписки и т. д.) с таймаутом

Types

type DatabaseWorker

type DatabaseWorker func(ctx context.Context, tx *pg.Transaction) (needCommit bool, res *JobResultErr)

DatabaseWorker - стандартное действие с транзакцией

type DatabaseWorkerWithResponse

type DatabaseWorkerWithResponse func(ctx context.Context, tx *pg.Transaction) (
	needCommit bool, res *JobResultProto)

DatabaseWorkerWithResponse - действие с транзакцией, подразумевающее прото результат

type DatabaseWorkerWithResult

type DatabaseWorkerWithResult func(ctx context.Context, tx *pg.Transaction) (needCommit bool, res structs.Result)

DatabaseWorkerWithResult - действие с транзакцией, подразумевающее результат типа structs.Result

type JobResultErr

type JobResultErr struct {
	Effects []func() // Эффекты, которые надо применить в случае успеха коммита транзакции
	Err     error    // Ошибка
	// contains filtered or unexported fields
}

JobResultErr - результат работы с транзакцией

func WrapError

func WrapError(err error) *JobResultErr

type JobResultProto

type JobResultProto struct {
	Effects []func()      // Эффекты, которые надо применить в случае успеха коммита транзакции
	Msg     proto.Message // Прото результат
	// contains filtered or unexported fields
}

JobResultProto - результат работы с транзакцией

func WrapMessage

func WrapMessage(msg proto.Message) *JobResultProto

type Queue

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

Queue - очередь запросов в БД

func NewQueue

func NewQueue(size int) *Queue

func (*Queue) Close

func (q *Queue) Close()

func (*Queue) GetQueue

func (q *Queue) GetQueue() chan *Task

func (*Queue) MakeJob

func (q *Queue) MakeJob(t *Task) error

type Task

type Task struct {
	Ctx                   context.Context // Контекст
	Options               pgx.TxOptions   // Опции транзакции
	QueueTimeout, Timeout time.Duration   // Таймаут попадания в очередь и таймаут транзакции
	Worker                DatabaseWorker  // Действие с транзакцией
	// contains filtered or unexported fields
}

Task - задача по работы с БД

type Worker

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

Worker - обработчик запроса в БД. Берет запросы из очереди, и запускает выполнение работы

func NewWorker

func NewWorker(queue chan *Task) *Worker

func (*Worker) Run

func (w *Worker) Run()