cron

Documentation

Index

• Variables
• func RegisterRunner(params RunnerParams) error
• type DLQProcessorTaskParams
• type Job
  • func NewJob(ctx context.Context, cfg *config.Config, logger *zap.Logger, ...) (*Job, error)
  • func (j *Job) Run()
• type NodeCanaryTaskParams
• type PollingCanaryTaskParams
• type RunnerParams
• type StreamingCanaryTaskParams
• type Task
  • func NewDLQProcessor(params DLQProcessorTaskParams) Task
  • func NewNodeCanary(params NodeCanaryTaskParams) (Task, error)
  • func NewPollingCanary(params PollingCanaryTaskParams) (Task, error)
  • func NewStreamingCanary(params StreamingCanaryTaskParams) (Task, error)
  • func NewWorkflowStatus(params WorkflowStatusTaskParams) (Task, error)
• type WorkflowStatusTaskParams

Variables

var (
	ErrSkipped = xerrors.New("skipped")
)

var Module = fx.Options(
	fx.Provide(fx.Annotated{
		Group:  "task",
		Target: NewDLQProcessor,
	}),
	fx.Provide(fx.Annotated{
		Group:  "task",
		Target: NewPollingCanary,
	}),
	fx.Provide(fx.Annotated{
		Group:  "task",
		Target: NewStreamingCanary,
	}),
	fx.Provide(fx.Annotated{
		Group:  "task",
		Target: NewNodeCanary,
	}),
	fx.Provide(fx.Annotated{
		Group:  "task",
		Target: NewWorkflowStatus,
	}),
)

Functions

func RegisterRunner

func RegisterRunner(params RunnerParams) error

Types

type DLQProcessorTaskParams

type DLQProcessorTaskParams struct {
	fx.In
	fxparams.Params
	DLQ              dlq.DLQ
	BlockchainClient client.Client `name:"slave"`
	BlobStorage      blobstorage.BlobStorage
	MetaStorage      metastorage.MetaStorage
}

type Job

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

func NewJob

func NewJob(ctx context.Context, cfg *config.Config, logger *zap.Logger, metrics tally.Scope, task Task) (*Job, error)

func (*Job) Run

func (j *Job) Run()

type NodeCanaryTaskParams

type NodeCanaryTaskParams struct {
	fx.In
	fxparams.Params
	Config          *config.Config
	Clients         client.ClientParams
	BlockStorage    storage.BlockStorage
	EventStorage    storage.EventStorage
	FailoverManager endpoints.FailoverManager
}

type PollingCanaryTaskParams

type PollingCanaryTaskParams struct {
	fx.In
	fxparams.Params
	Client gateway.Client
	Config *config.Config
}

type RunnerParams

type RunnerParams struct {
	fx.In
	fxparams.Params
	Lifecycle fx.Lifecycle
	Manager   services.SystemManager
	Tasks     []Task `group:"task"`
}

type StreamingCanaryTaskParams

type StreamingCanaryTaskParams struct {
	fx.In
	fxparams.Params
	Client       sdk.Client
	EventStorage storage.EventStorage
	Config       *config.Config
}

type Task

type Task interface {
	Name() string
	Spec() string
	Parallelism() int64
	DelayStartDuration() time.Duration
	Run(ctx context.Context) error
	Enabled() bool
}

func NewDLQProcessor

func NewDLQProcessor(params DLQProcessorTaskParams) Task

func NewNodeCanary

func NewNodeCanary(params NodeCanaryTaskParams) (Task, error)

func NewPollingCanary

func NewPollingCanary(params PollingCanaryTaskParams) (Task, error)

func NewStreamingCanary

func NewStreamingCanary(params StreamingCanaryTaskParams) (Task, error)

func NewWorkflowStatus

func NewWorkflowStatus(params WorkflowStatusTaskParams) (Task, error)

type WorkflowStatusTaskParams

type WorkflowStatusTaskParams struct {
	fx.In
	fxparams.Params
	Config  *config.Config
	Runtime cadence.Runtime
}