compute

Documentation

Index

• Variables
• type DefaultPool
  • func (p *DefaultPool) Close() (err error)
  • func (p *DefaultPool) GetWorker(ctx context.Context) (Worker, error)
  • func (p *DefaultPool) ReturnWorker(worker Worker)
• type DefaultWorkQueue
  • func (q *DefaultWorkQueue) Add(info WorkInfo)
  • func (q *DefaultWorkQueue) GetMaxSize() int
  • func (q *DefaultWorkQueue) SetMaxSize(size int)
  • func (q *DefaultWorkQueue) Wait()
• type GenericWorkInfo
  • func NewWorkInfo[T, U any](ctx context.Context, request T, runFunc WorkRunFunc[T, U]) *GenericWorkInfo[T, U]
• type Pool
  • func NewPool(logger *zap.Logger, factory WorkerFactory) Pool
• type ReadyOptions
• type ReadyOptionsFunc
  • func WithConnTimeout(timeout time.Duration) ReadyOptionsFunc
  • func WithTickerInterval(interval time.Duration) ReadyOptionsFunc
• type WorkInfo
• type WorkQueue
  • func NewQueue(logger *zap.Logger, pool Pool, maxSize int) WorkQueue
• type WorkRunFunc
• type Worker
• type WorkerFactory

Variables

var (
	ErrClosed = errors.New("worker closed")
)

Types

type DefaultPool

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

func (*DefaultPool) Close

func (p *DefaultPool) Close() (err error)

func (*DefaultPool) GetWorker

func (p *DefaultPool) GetWorker(ctx context.Context) (Worker, error)

func (*DefaultPool) ReturnWorker

func (p *DefaultPool) ReturnWorker(worker Worker)

type DefaultWorkQueue

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

func (*DefaultWorkQueue) Add

func (q *DefaultWorkQueue) Add(info WorkInfo)

func (*DefaultWorkQueue) GetMaxSize

func (q *DefaultWorkQueue) GetMaxSize() int

func (*DefaultWorkQueue) SetMaxSize

func (q *DefaultWorkQueue) SetMaxSize(size int)

func (*DefaultWorkQueue) Wait

func (q *DefaultWorkQueue) Wait()

type GenericWorkInfo

type GenericWorkInfo[T, U any] struct {
	Ctx     context.Context
	Request T
	Result  chan U
	Err     chan error
	Run     WorkRunFunc[T, U]
}

func NewWorkInfo

func NewWorkInfo[T, U any](ctx context.Context, request T, runFunc WorkRunFunc[T, U]) *GenericWorkInfo[T, U]

type Pool

type Pool interface {
	io.Closer

	GetWorker(ctx context.Context) (Worker, error)
	ReturnWorker(worker Worker)
}

func NewPool

func NewPool(logger *zap.Logger, factory WorkerFactory) Pool

type ReadyOptions

type ReadyOptions struct {
	TickerInterval time.Duration
	ConnTimeout    time.Duration
}

type ReadyOptionsFunc

type ReadyOptionsFunc func(options *ReadyOptions)

func WithConnTimeout

func WithConnTimeout(timeout time.Duration) ReadyOptionsFunc

func WithTickerInterval

func WithTickerInterval(interval time.Duration) ReadyOptionsFunc

type WorkInfo

type WorkInfo interface {
	// contains filtered or unexported methods
}

type WorkQueue

type WorkQueue interface {
	// Add new job to the WorkQueue
	Add(info WorkInfo)
	// Wait for all jobs to complete
	Wait()
	// GetMaxSize returns maximum number of allowed concurrent workers
	GetMaxSize() int
	// SetMaxSize sets maximum number of allowed concurrent workers
	SetMaxSize(size int)
}

func NewQueue

func NewQueue(logger *zap.Logger, pool Pool, maxSize int) WorkQueue

type WorkRunFunc

type WorkRunFunc[T, U any] func(ctx context.Context, logger *zap.Logger, req T, instance Worker) (U, error)

type Worker

type Worker interface {
	io.Closer

	Equals(other Worker) bool

	Connect(ctx context.Context) error

	Worker() proto.WorkerServiceClient
	Job() proto.JobServiceClient

	IsReady(ctx context.Context, opts ...ReadyOptionsFunc) (bool, error)
	IsReadyChan(ctx context.Context, opts ...ReadyOptionsFunc) <-chan error
}

type WorkerFactory

type WorkerFactory interface {
	Create(ctx context.Context) (Worker, error)
}