chain

Apr 8, 2019

Changes in this version

+ var ErrConflict = errors.New("chain already exists in repo")

+ var ErrMultipleDeleted = errors.New("multiple chains deleted")

+ var ErrNotFound = errors.New("chain not found in repo")

+ var ErrShuttingDown = fmt.Errorf("chain not stopped because traverser is shutting down")

+ func NewMemoryRepo() *memoryRepo

+ func NewTraverser(cfg TraverserConfig) *traverser

+ func Validate(jobChain proto.JobChain, new bool) error

+ type Chain struct

+ func NewChain(jc *proto.JobChain, sequenceTries map[string]uint, ...) *Chain

+ func (c *Chain) CanRetrySequence(jobId string) bool

+ func (c *Chain) FailedJobs() uint

+ func (c *Chain) FinishedJobs() uint

+ func (c *Chain) IncrementFinishedJobs(delta int)

+ func (c *Chain) IncrementJobTries(jobId string, delta int)

+ func (c *Chain) IncrementSequenceTries(jobId string, delta int)

+ func (c *Chain) IsDoneRunning() (done bool, complete bool)

+ func (c *Chain) IsRunnable(jobId string) bool

+ func (c *Chain) IsSequenceStartJob(jobId string) bool

+ func (c *Chain) JobState(jobId string) byte

+ func (c *Chain) JobTries(jobId string) (cur uint, total uint)

+ func (c *Chain) NextJobs(jobId string) proto.Jobs

+ func (c *Chain) RequestId() string

+ func (c *Chain) RunnableJobs() proto.Jobs

+ func (c *Chain) SequenceStartJob(jobId string) proto.Job

+ func (c *Chain) SequenceTries(jobId string) uint

+ func (c *Chain) SetJobState(jobId string, state byte)

+ func (c *Chain) SetState(state byte)

+ func (c *Chain) State() byte

+ func (c *Chain) ToSuspended() proto.SuspendedJobChain

+ type ChainReaperFactory struct

+ Chain *Chain

+ ChainRepo Repo

+ DoneJobChan chan proto.Job

+ Logger *log.Entry

+ RMCRetryWait time.Duration

+ RMCTries int

+ RMClient rm.Client

+ RunJobChan chan proto.Job

+ RunnerRepo runner.Repo

+ func (f *ChainReaperFactory) MakeRunning() JobReaper

+ func (f *ChainReaperFactory) MakeStopped() JobReaper

+ func (f *ChainReaperFactory) MakeSuspended() JobReaper

+ type ErrInvalidChain struct

+ Message string

+ func (e ErrInvalidChain) Error() string

+ type JobReaper interface

+ Run func()

+ Stop func()

+ type ReaperFactory interface

+ MakeRunning func() JobReaper

+ MakeStopped func() JobReaper

+ MakeSuspended func() JobReaper

+ type Repo interface

+ Add func(*Chain) error

+ Get func(string) (*Chain, error)

+ GetAll func() ([]*Chain, error)

+ Remove func(string) error

+ Set func(*Chain) error

+ type RunningChainReaper struct

+ func (r *RunningChainReaper) Finalize(complete bool)

+ func (r *RunningChainReaper) Reap(job proto.Job)

+ func (r *RunningChainReaper) Run()

+ func (r *RunningChainReaper) Stop()

+ type StoppedChainReaper struct

+ func (r *StoppedChainReaper) Finalize()

+ func (r *StoppedChainReaper) Reap(job proto.Job)

+ func (r *StoppedChainReaper) Run()

+ func (r *StoppedChainReaper) Stop()

+ type SuspendedChainReaper struct

+ func (r *SuspendedChainReaper) Finalize()

+ func (r *SuspendedChainReaper) Reap(job proto.Job)

+ func (r *SuspendedChainReaper) Run()

+ func (r *SuspendedChainReaper) Stop()

+ type Traverser interface

+ Run func()

+ Running func() []proto.JobStatus

+ Stop func() error

+ type TraverserConfig struct

+ Chain *Chain

+ ChainRepo Repo

+ RMClient rm.Client

+ RunnerFactory runner.Factory

+ SendTimeout time.Duration

+ ShutdownChan chan struct{}

+ StopTimeout time.Duration

+ type TraverserFactory interface

+ Make func(*proto.JobChain) (Traverser, error)

+ MakeFromSJC func(*proto.SuspendedJobChain) (Traverser, error)

+ func NewTraverserFactory(chainRepo Repo, rf runner.Factory, rmc rm.Client, shutdownChan chan struct{}) TraverserFactory