plugins

Documentation

Overview

Package plugins defines APIs for source and destination plugins

Index

• func SetDestinationManagedCqColumns(tables []*schema.Table)
• func TestSourcePluginSync(t *testing.T, plugin *SourcePlugin, logger zerolog.Logger, spec specs.Source)
• type DestinationClient
• type DestinationPlugin
  • func NewDestinationPlugin(name string, version string, newDestinationClient NewDestinationClientFunc) *DestinationPlugin
  • func (p *DestinationPlugin) Close(ctx context.Context) error
  • func (p *DestinationPlugin) DeleteStale(ctx context.Context, tables []string, sourceName string, syncTime time.Time) uint64
  • func (p *DestinationPlugin) Init(ctx context.Context, logger zerolog.Logger, spec specs.Destination) error
  • func (p *DestinationPlugin) Migrate(ctx context.Context, tables schema.Tables) error
  • func (p *DestinationPlugin) Name() string
  • func (p *DestinationPlugin) Version() string
  • func (p *DestinationPlugin) Write(ctx context.Context, sourceName string, syncTime time.Time, ...) *WriteSummary
• type NewDestinationClientFunc
• type ResourceTestCase
• type SourceNewExecutionClientFunc
• type SourcePlugin
  • func NewSourcePlugin(name string, version string, tables []*schema.Table, ...) *SourcePlugin
  • func (p *SourcePlugin) GenerateSourcePluginDocs(dir string) error
  • func (p *SourcePlugin) Name() string
  • func (p *SourcePlugin) Sync(ctx context.Context, logger zerolog.Logger, spec specs.Source, ...) (*schema.SyncSummary, error)
  • func (p *SourcePlugin) Tables() schema.Tables
  • func (p *SourcePlugin) Version() string
• type WriteSummary

Functions

func SetDestinationManagedCqColumns

func SetDestinationManagedCqColumns(tables []*schema.Table)

Overwrites or adds the CQ columns that are managed by the destination plugins (_cq_sync_time, _cq_source_name).

func TestSourcePluginSync

func TestSourcePluginSync(t *testing.T, plugin *SourcePlugin, logger zerolog.Logger, spec specs.Source)

Types

type DestinationClient

type DestinationClient interface {
	Migrate(ctx context.Context, tables schema.Tables) error
	Write(ctx context.Context, table string, data map[string]interface{}) error
	DeleteStale(ctx context.Context, tables string, sourceName string, syncTime time.Time) error
	Close(ctx context.Context) error
}

type DestinationPlugin

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

func NewDestinationPlugin

func NewDestinationPlugin(name string, version string, newDestinationClient NewDestinationClientFunc) *DestinationPlugin

func (*DestinationPlugin) Close

func (p *DestinationPlugin) Close(ctx context.Context) error

func (*DestinationPlugin) DeleteStale

func (p *DestinationPlugin) DeleteStale(ctx context.Context, tables []string, sourceName string, syncTime time.Time) uint64

func (*DestinationPlugin) Init

func (p *DestinationPlugin) Init(ctx context.Context, logger zerolog.Logger, spec specs.Destination) error

we need lazy loading because we want to be able to initialize after

func (*DestinationPlugin) Migrate

func (p *DestinationPlugin) Migrate(ctx context.Context, tables schema.Tables) error

we implement all DestinationClient functions so we can hook into pre-post behavior

func (*DestinationPlugin) Name

func (p *DestinationPlugin) Name() string

func (*DestinationPlugin) Version

func (p *DestinationPlugin) Version() string

func (*DestinationPlugin) Write

func (p *DestinationPlugin) Write(ctx context.Context, sourceName string, syncTime time.Time, res <-chan *schema.Resource) *WriteSummary

type NewDestinationClientFunc

type NewDestinationClientFunc func(context.Context, zerolog.Logger, specs.Destination) (DestinationClient, error)

type ResourceTestCase

type ResourceTestCase struct {
	Plugin *SourcePlugin
	Spec   specs.Source
	// ParallelFetchingLimit limits parallel resources fetch at a time
	ParallelFetchingLimit uint64
	// SkipIgnoreInTest flag which detects if schema.Table or schema.Column should be ignored
	SkipIgnoreInTest bool
}

type SourceNewExecutionClientFunc

type SourceNewExecutionClientFunc func(context.Context, zerolog.Logger, specs.Source) (schema.ClientMeta, error)

type SourcePlugin

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

SourcePlugin is the base structure required to pass to sdk.serve We take a declarative approach to API here similar to Cobra

func NewSourcePlugin

func NewSourcePlugin(name string, version string, tables []*schema.Table, newExecutionClient SourceNewExecutionClientFunc) *SourcePlugin

NewSourcePlugin returns a new plugin with a given name, version, tables, newExecutionClient and additional options.

func (*SourcePlugin) GenerateSourcePluginDocs

func (p *SourcePlugin) GenerateSourcePluginDocs(dir string) error

GenerateSourcePluginDocs creates table documentation for the source plugin based on its list of tables

func (*SourcePlugin) Name

func (p *SourcePlugin) Name() string

Name return the name of this plugin

func (*SourcePlugin) Sync

func (p *SourcePlugin) Sync(ctx context.Context, logger zerolog.Logger, spec specs.Source, res chan<- *schema.Resource) (*schema.SyncSummary, error)

Sync is syncing data from the requested tables in spec to the given channel

func (*SourcePlugin) Tables

func (p *SourcePlugin) Tables() schema.Tables

Tables returns all supported tables by this source plugin

func (*SourcePlugin) Version

func (p *SourcePlugin) Version() string

Version returns the version of this plugin

type WriteSummary

type WriteSummary struct {
	SuccessWrites uint64
	FailedWrites  uint64
	FailedDeletes uint64
}