middlewares

README

Middlewares

Middlewares are common code components that are intended to be reused by multiple sources, sinks, or in general transfer pipeline.

Each middleware is a function with the following properties:

1. Its arguments are its configuration properties — the ones which configure the behaviour of this middleware;
2. Its result is a function which can be used between source and sink — that is, it should fulfill at least one of two properties:
   • It accepts an interface from abstract, making no assumptions about possible middlewares which wrap the actual object;
   • It returns an interface from abstract, for which the callers must not make assumptions about possible middlewares which wrap it.

The second requirement defines the main property to call a function "middleware" — it operates on abstract interfaces.

Documentation

Index

• func AtReplicationStage(config *Config)
• func ErrorTracker(mtrcs metrics.Registry) func(abstract.Sinker) abstract.Sinker
• func ExcludeSystemTables(item *abstract.ChangeItem) bool
• func Filter(registry metrics.Registry, predicates ...ExcludePredicate) func(abstract.Sinker) abstract.Sinker
• func InputDataMetering() func(abstract.Sinker) abstract.Sinker
• func IntervalThrottler(logger log.Logger, interval time.Duration) func(abstract.Sinker) abstract.Sinker
• func NonRowSeparator() func(abstract.Sinker) abstract.Sinker
• func OutputDataBatchMetering() func(middlewares.Asynchronizer) middlewares.Asynchronizer
• func OutputDataMetering() func(abstract.Sinker) abstract.Sinker
• func PlugTransformer(pt PluggableTransformer)
• func PluggableTransformersChain(t *server.Transfer, r metrics.Registry, cp coordinator.Coordinator) func(abstract.Sinker) abstract.Sinker
• func Retrier(logger log.Logger, ctx context.Context) func(abstract.Sinker) abstract.Sinker
• func SourceFallbacks(version int, typ abstract.ProviderType, lgr log.Logger, ...) abstract.SinkOption
• func Statistician(logger log.Logger, stats *stats.WrapperStats) func(abstract.Sinker) abstract.Sinker
• func TableTemporator(logger log.Logger, transferID string, config server.TmpPolicyConfig) func(abstract.Movable) abstract.Sinker
• func TargetFallbacks(version int, typ abstract.ProviderType, lgr log.Logger, ...) abstract.SinkOption
• func Transformation(transfer *server.Transfer, logger log.Logger, metrics metrics.Registry) (func(abstract.Sinker) abstract.Sinker, error)
• func TypeStrictnessTracker(logger log.Logger, stats *stats.TypeStrictnessStats) func(abstract.Sinker) abstract.Sinker
• func WithEnableRetries(config *Config)
• func WithNoData(config *Config)
• type Config
  • func MakeConfig(options ...ConfigOption) Config
• type ConfigOption
• type ExcludePredicate
• type PluggableTransformer
• type TransferStage

Functions

func AtReplicationStage

func AtReplicationStage(config *Config)

func ErrorTracker

func ErrorTracker(mtrcs metrics.Registry) func(abstract.Sinker) abstract.Sinker

ErrorTracker do nothing except tracking error / success pushes into metrics

func ExcludeSystemTables

func ExcludeSystemTables(item *abstract.ChangeItem) bool

func Filter

func Filter(registry metrics.Registry, predicates ...ExcludePredicate) func(abstract.Sinker) abstract.Sinker

func InputDataMetering

func InputDataMetering() func(abstract.Sinker) abstract.Sinker

func IntervalThrottler

func IntervalThrottler(logger log.Logger, interval time.Duration) func(abstract.Sinker) abstract.Sinker

IntervalThrottler blocks Push until a given interval since the finish of the previous Push passes

func NonRowSeparator

func NonRowSeparator() func(abstract.Sinker) abstract.Sinker

NonRowSeparator separates non-row items and pushes each of them in a distinct call to Push (which will contain just one item)

func OutputDataBatchMetering

func OutputDataBatchMetering() func(middlewares.Asynchronizer) middlewares.Asynchronizer

func OutputDataMetering

func OutputDataMetering() func(abstract.Sinker) abstract.Sinker

func PlugTransformer

func PlugTransformer(pt PluggableTransformer)

PlugTransformer adds a new pluggable transformer to a chain of such transformers. This method should be called from `init()` function.

func PluggableTransformersChain

func PluggableTransformersChain(t *server.Transfer, r metrics.Registry, cp coordinator.Coordinator) func(abstract.Sinker) abstract.Sinker

func Retrier

func Retrier(logger log.Logger, ctx context.Context) func(abstract.Sinker) abstract.Sinker

Retrier retries Push operations automatically using the hardcoded delay and interval. Retries can be interrupted using the given context. Push operations containing non-row items are NOT retried

func SourceFallbacks

func SourceFallbacks(version int, typ abstract.ProviderType, lgr log.Logger, stat *stats.FallbackStats) abstract.SinkOption

func Statistician

func Statistician(logger log.Logger, stats *stats.WrapperStats) func(abstract.Sinker) abstract.Sinker

Statistician tracks the traffic of sinker

func TableTemporator

func TableTemporator(logger log.Logger, transferID string, config server.TmpPolicyConfig) func(abstract.Movable) abstract.Sinker

TableTemporator provides support for temporary policy

func TargetFallbacks

func TargetFallbacks(version int, typ abstract.ProviderType, lgr log.Logger, stat *stats.FallbackStats) abstract.SinkOption

func Transformation

func Transformation(transfer *server.Transfer, logger log.Logger, metrics metrics.Registry) (func(abstract.Sinker) abstract.Sinker, error)

func TypeStrictnessTracker

func TypeStrictnessTracker(logger log.Logger, stats *stats.TypeStrictnessStats) func(abstract.Sinker) abstract.Sinker

TypeStrictnessTracker tracks the "strictness" of types passing through it.

"Strictness" is the compliance of the column value in an item with the *strict* representation defined by the Data Transfer type system.

func WithEnableRetries

func WithEnableRetries(config *Config)

func WithNoData

func WithNoData(config *Config)

Types

type Config

type Config struct {
	NoData           bool
	EnableRetries    bool
	ReplicationStage bool
}

func MakeConfig

func MakeConfig(options ...ConfigOption) Config

type ConfigOption

type ConfigOption func(*Config)

type ExcludePredicate

type ExcludePredicate func(*abstract.ChangeItem) bool

type PluggableTransformer

type PluggableTransformer func(*server.Transfer, metrics.Registry, coordinator.Coordinator) func(abstract.Sinker) abstract.Sinker

PluggableTransformer is a transformer with a middleware interface which packages outside of `middlewares` can provide.

type TransferStage

type TransferStage string