block

Documentation

Index

• Variables
• type PlanConsumer
  • func NewPlanConsumer[V fmt.Stringer](name string, cg *sync.WaitGroup, wg *sync.WaitGroup, ctx context.Context, ...) (PlanConsumer, error)
• type PlanEntry
  • func NewPlanEntry[V fmt.Stringer](name string, start int64, end int64, step int64, data *V) PlanEntry[V]
• type PlanExecutor
• type PlanExecutorCreator
• type PlanGenerator
• type PlanLogger
• type PlanProducer
  • func NewPlanProducer[V fmt.Stringer](cfg PlannerConfig, wg *sync.WaitGroup, ctx context.Context, ...) (PlanProducer, error)
• type PlannedBlockWriter
  • func NewPlannedBlockWriter[V fmt.Stringer](config PlannerConfig, generator PlanGenerator[V], ...) PlannedBlockWriter
• type Planner
  • func NewPlanner[V fmt.Stringer](config PlannerConfig) Planner[V]
• type PlannerConfig
  • func NewPlannerConfig(outputDirectory string, startTime time.Time, endTime time.Time, ...) PlannerConfig

Variables

var (
	MaxParallelism = uint8(runtime.NumCPU())
)

Types

type PlanConsumer

type PlanConsumer interface {
	Run()
}

func NewPlanConsumer

func NewPlanConsumer[V fmt.Stringer](name string, cg *sync.WaitGroup, wg *sync.WaitGroup, ctx context.Context, errorChan chan<- error, input <-chan PlanEntry[V], s common.Canceller, e PlanExecutor[V]) (PlanConsumer, error)

type PlanEntry

type PlanEntry[V fmt.Stringer] interface {
	Start() int64
	End() int64
	Step() int64
	Data() *V
}

func NewPlanEntry

func NewPlanEntry[V fmt.Stringer](name string, start int64, end int64, step int64, data *V) PlanEntry[V]

type PlanExecutor

type PlanExecutor[V fmt.Stringer] interface {
	Execute(ctx context.Context, logger PlanLogger[V], plan PlanEntry[V]) error
}

type PlanExecutorCreator

type PlanExecutorCreator[V fmt.Stringer] interface {
	Create(name string, appender database.Appender) (PlanExecutor[V], error)
}

type PlanGenerator

type PlanGenerator[V fmt.Stringer] interface {
	Generate(chunkStart int64, chunkEnd int64, stepDuration int64) []PlanEntry[V]
}

type PlanLogger

type PlanLogger[V fmt.Stringer] interface {
	PrintMessage(format string, args ...interface{})
	PrintExecutePlanError(plan PlanEntry[V], msg string, err error)
}

type PlanProducer

type PlanProducer interface {
	Run()
}

func NewPlanProducer

func NewPlanProducer[V fmt.Stringer](cfg PlannerConfig, wg *sync.WaitGroup, ctx context.Context, generator PlanGenerator[V], output chan<- PlanEntry[V]) (PlanProducer, error)

type PlannedBlockWriter

type PlannedBlockWriter interface {
	Run() error
}

func NewPlannedBlockWriter

func NewPlannedBlockWriter[V fmt.Stringer](config PlannerConfig, generator PlanGenerator[V], executorCreator PlanExecutorCreator[V]) PlannedBlockWriter

type Planner

type Planner[V fmt.Stringer] interface {
	Plan(transform func(int64, int64, int64) []PlanEntry[V]) [][]PlanEntry[V]
}

func NewPlanner

func NewPlanner[V fmt.Stringer](config PlannerConfig) Planner[V]

type PlannerConfig

type PlannerConfig interface {
	OutputDirectory() string
	StartTime() time.Time
	EndTime() time.Time
	BlockDuration() int64
	SampleInterval() time.Duration
	Parallelism() uint8
}

func NewPlannerConfig

func NewPlannerConfig(outputDirectory string, startTime time.Time, endTime time.Time, sampleInterval time.Duration, parallelism int) PlannerConfig