process

Documentation

Index

• Variables
• func NewPoolWorker(name string, concurrency uint64, scheduler PoolWorkerScheduler) *poolWorker
• type Interface
• type PeriodicTask
  • func NewPeriodicTask(name string, ticker utils.Ticker, taskFn func(ctx context.Context)) *PeriodicTask
  • func (task *PeriodicTask) AbortIfRunning()
  • func (task *PeriodicTask) Close() error
  • func (task *PeriodicTask) Enqueue()
  • func (task *PeriodicTask) ForceRerun()
  • func (task *PeriodicTask) Start() error
• type Pool
  • func NewPool(name string, concurrency uint64, retryInterval time.Duration) *Pool
  • func (p *Pool) Add(item PoolUniqueIDer)
  • func (p *Pool) Complete(id PoolUniqueID)
  • func (p *Pool) ForceRetry(id PoolUniqueID)
  • func (p *Pool) Get(ctx context.Context) (item interface{}, err error)
  • func (p *Pool) NumItemsPending() int
  • func (p *Pool) RetryLater(id PoolUniqueID, when time.Time)
  • func (p *Pool) Start() error
• type PoolUniqueID
• type PoolUniqueIDer
• type PoolWorker
• type PoolWorkerItem
• type PoolWorkerScheduler
• type Process
  • func New(name string) *Process
  • func (p *Process) Autoclose()
  • func (p *Process) AutocloseWithCleanup(closeFn func())
  • func (p *Process) Close() error
  • func (p *Process) Ctx() context.Context
  • func (p *Process) Done() <-chan struct{}
  • func (p *Process) Go(ctx context.Context, name string, fn func(ctx context.Context)) <-chan struct{}
  • func (p *Process) Name() string
  • func (p *Process) NewChild(ctx context.Context, name string) *Process
  • func (p *Process) ProcessTree() map[string]interface{}
  • func (p *Process) SpawnChild(ctx context.Context, child Spawnable) error
  • func (p *Process) Start() error
  • func (p *Process) State() State
• type ProcessTreer
• type Spawnable
• type State
  • func (s State) String() string
• type StaticScheduler
  • func NewStaticScheduler(checkForRetriesInterval time.Duration, retryAfter time.Duration) StaticScheduler
  • func (s StaticScheduler) CheckForRetriesInterval() time.Duration
  • func (s StaticScheduler) RetryWhen(item PoolWorkerItem) time.Time

Variables

var (
	ErrUnstarted = errors.New("unstarted")
	ErrClosed    = errors.New("closed")
)

Functions

func NewPoolWorker

func NewPoolWorker(name string, concurrency uint64, scheduler PoolWorkerScheduler) *poolWorker

Types

type Interface

type Interface interface {
	Spawnable
	ProcessTreer
	Autoclose()
	AutocloseWithCleanup(closeFn func())
	Ctx() context.Context
	NewChild(ctx context.Context, name string) *Process
	SpawnChild(ctx context.Context, child Spawnable) error
	Go(ctx context.Context, name string, fn func(ctx context.Context)) <-chan struct{}
}

type PeriodicTask

type PeriodicTask struct {
	Process
	// contains filtered or unexported fields
}

func NewPeriodicTask

func NewPeriodicTask(name string, ticker utils.Ticker, taskFn func(ctx context.Context)) *PeriodicTask

func (*PeriodicTask) AbortIfRunning

func (task *PeriodicTask) AbortIfRunning()

func (*PeriodicTask) Close

func (task *PeriodicTask) Close() error

func (*PeriodicTask) Enqueue

func (task *PeriodicTask) Enqueue()

func (*PeriodicTask) ForceRerun

func (task *PeriodicTask) ForceRerun()

func (*PeriodicTask) Start

func (task *PeriodicTask) Start() error

type Pool

type Pool struct {
	Process
	// contains filtered or unexported fields
}

func NewPool

func NewPool(name string, concurrency uint64, retryInterval time.Duration) *Pool

func (*Pool) Add

func (p *Pool) Add(item PoolUniqueIDer)

func (*Pool) Complete

func (p *Pool) Complete(id PoolUniqueID)

func (*Pool) ForceRetry

func (p *Pool) ForceRetry(id PoolUniqueID)

func (*Pool) Get

func (p *Pool) Get(ctx context.Context) (item interface{}, err error)

func (*Pool) NumItemsPending

func (p *Pool) NumItemsPending() int

func (*Pool) RetryLater

func (p *Pool) RetryLater(id PoolUniqueID, when time.Time)

func (*Pool) Start

func (p *Pool) Start() error

type PoolUniqueID

type PoolUniqueID interface{}

type PoolUniqueIDer

type PoolUniqueIDer interface {
	ID() PoolUniqueID
}

type PoolWorker

type PoolWorker interface {
	Interface
	Add(item PoolWorkerItem)
	ForceRetry(uniqueID PoolUniqueID)
}

type PoolWorkerItem

type PoolWorkerItem interface {
	PoolUniqueIDer
	Work(ctx context.Context) (retry bool)
}

type PoolWorkerScheduler

type PoolWorkerScheduler interface {
	CheckForRetriesInterval() time.Duration
	RetryWhen(item PoolWorkerItem) time.Time
}

type Process

type Process struct {
	log.Logger
	// contains filtered or unexported fields
}

func New

func New(name string) *Process

func (*Process) Autoclose

func (p *Process) Autoclose()

func (*Process) AutocloseWithCleanup

func (p *Process) AutocloseWithCleanup(closeFn func())

func (*Process) Close

func (p *Process) Close() error

func (*Process) Ctx

func (p *Process) Ctx() context.Context

func (*Process) Done

func (p *Process) Done() <-chan struct{}

func (*Process) Go

func (p *Process) Go(ctx context.Context, name string, fn func(ctx context.Context)) <-chan struct{}

func (*Process) Name

func (p *Process) Name() string

func (*Process) NewChild

func (p *Process) NewChild(ctx context.Context, name string) *Process

func (*Process) ProcessTree

func (p *Process) ProcessTree() map[string]interface{}

func (*Process) SpawnChild

func (p *Process) SpawnChild(ctx context.Context, child Spawnable) error

func (*Process) Start

func (p *Process) Start() error

func (*Process) State

func (p *Process) State() State

type ProcessTreer

type ProcessTreer interface {
	Spawnable
	ProcessTree() map[string]interface{}
}

type Spawnable

type Spawnable interface {
	Name() string
	Start() error
	Close() error
	Done() <-chan struct{}
	State() State
}

type State

type State int

const (
	Unstarted State = iota
	Started
	Closed
)

func (State) String

func (s State) String() string

type StaticScheduler

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

func NewStaticScheduler

func NewStaticScheduler(checkForRetriesInterval time.Duration, retryAfter time.Duration) StaticScheduler

func (StaticScheduler) CheckForRetriesInterval

func (s StaticScheduler) CheckForRetriesInterval() time.Duration

func (StaticScheduler) RetryWhen

func (s StaticScheduler) RetryWhen(item PoolWorkerItem) time.Time