scheduler

Documentation

Index

• Constants
• Variables
• type BatchOption
  • func WithBatchMaxRows(rows int) BatchOption
  • func WithBatchTimeout(timeout time.Duration) BatchOption
• type BatchSettings
• type Client
• type Option
  • func WithBatchOptions(options ...BatchOption) Option
  • func WithConcurrency(concurrency int) Option
  • func WithInvocationID(invocationID string) Option
  • func WithLogger(logger zerolog.Logger) Option
  • func WithMaxDepth(maxDepth uint64) Option
  • func WithSingleNestedTableMaxConcurrency(concurrency int64) Option
  • func WithSingleResourceMaxConcurrency(concurrency int64) Option
  • func WithStrategy(strategy Strategy) Option
  • func WithoutBatching() Option
• type Scheduler
  • func NewScheduler(opts ...Option) *Scheduler
  • func (s *Scheduler) Sync(ctx context.Context, client schema.ClientMeta, tables schema.Tables, ...) error
  • func (s *Scheduler) SyncAll(ctx context.Context, client schema.ClientMeta, tables schema.Tables) (message.SyncMessages, error)
• type Strategies
  • func (s Strategies) String() string
• type Strategy
  • func StrategyForName(s string) (Strategy, error)
  • func (Strategy) JSONSchema() *jsonschema.Schema
  • func (s *Strategy) MarshalJSON() ([]byte, error)
  • func (s *Strategy) String() string
  • func (s *Strategy) UnmarshalJSON(b []byte) error
  • func (s *Strategy) Validate() error
• type SyncOption
  • func WithShard(num int32, total int32) SyncOption
  • func WithSyncDeterministicCQID(deterministicCQID bool) SyncOption

Constants

const (
	DefaultBatchMaxRows = 50
	DefaultBatchTimeout = 5 * time.Second
)

const (
	DefaultSingleResourceMaxConcurrency    = 5
	DefaultSingleNestedTableMaxConcurrency = 5
	DefaultConcurrency                     = 50000
	DefaultMaxDepth                        = 4
)

Variables

var AllStrategies = Strategies{StrategyDFS, StrategyRoundRobin, StrategyShuffle, StrategyShuffleQueue}

var AllStrategyNames = [...]string{
	StrategyDFS:          "dfs",
	StrategyRoundRobin:   "round-robin",
	StrategyShuffle:      "shuffle",
	StrategyShuffleQueue: "shuffle-queue",
}

var ErrNoTables = errors.New("no tables specified for syncing, review `tables` and `skip_tables` in your config and specify at least one table to sync")

Types

type BatchOption

type BatchOption func(settings *BatchSettings)

func WithBatchMaxRows

func WithBatchMaxRows(rows int) BatchOption

func WithBatchTimeout

func WithBatchTimeout(timeout time.Duration) BatchOption

type BatchSettings

type BatchSettings struct {
	MaxRows int
	Timeout time.Duration
}

type Client

type Client interface {
	ID() string
}

type Option

type Option func(*Scheduler)

func WithBatchOptions

func WithBatchOptions(options ...BatchOption) Option

func WithConcurrency

func WithConcurrency(concurrency int) Option

func WithInvocationID

func WithInvocationID(invocationID string) Option

func WithLogger

func WithLogger(logger zerolog.Logger) Option

func WithMaxDepth

func WithMaxDepth(maxDepth uint64) Option

func WithSingleNestedTableMaxConcurrency

func WithSingleNestedTableMaxConcurrency(concurrency int64) Option

func WithSingleResourceMaxConcurrency

func WithSingleResourceMaxConcurrency(concurrency int64) Option

func WithStrategy

func WithStrategy(strategy Strategy) Option

func WithoutBatching

func WithoutBatching() Option

type Scheduler

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

func NewScheduler

func NewScheduler(opts ...Option) *Scheduler

func (*Scheduler) Sync

func (s *Scheduler) Sync(ctx context.Context, client schema.ClientMeta, tables schema.Tables, res chan<- message.SyncMessage, opts ...SyncOption) error

func (*Scheduler) SyncAll

func (s *Scheduler) SyncAll(ctx context.Context, client schema.ClientMeta, tables schema.Tables) (message.SyncMessages, error)

SyncAll is mostly used for testing as it will sync all tables and can run out of memory in the real world. Should use Sync for production.

type Strategies

type Strategies []Strategy

func (Strategies) String

func (s Strategies) String() string

type Strategy

type Strategy int

const (
	StrategyDFS Strategy = iota
	StrategyRoundRobin
	StrategyShuffle
	StrategyShuffleQueue
)

func StrategyForName

func StrategyForName(s string) (Strategy, error)

func (Strategy) JSONSchema

func (Strategy) JSONSchema() *jsonschema.Schema

func (*Strategy) MarshalJSON

func (s *Strategy) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler.

func (*Strategy) String

func (s *Strategy) String() string

func (*Strategy) UnmarshalJSON

func (s *Strategy) UnmarshalJSON(b []byte) error

UnmarshalJSON implements json.Unmarshaler.

func (*Strategy) Validate

func (s *Strategy) Validate() error

type SyncOption

type SyncOption func(*syncClient)

func WithShard

func WithShard(num int32, total int32) SyncOption

func WithSyncDeterministicCQID

func WithSyncDeterministicCQID(deterministicCQID bool) SyncOption