pipeline

Documentation

Index

• type CDCKafkaSink
  • func NewCDCKafkaSink(conf *config.KafkaSink) *CDCKafkaSink
  • func (k *CDCKafkaSink) Close()
  • func (k *CDCKafkaSink) Write(msgs []cdc_event.CDCEvent, ctx context.Context) error
• type CDCPipeline
  • func NewCDCPipeline(name string, conf *config.CDCConfig, wg *sync.WaitGroup) CDCPipeline
  • func (cdc *CDCPipeline) Close()
  • func (cdc *CDCPipeline) Init() error
  • func (cdc *CDCPipeline) Query(query string) (*mysql.Result, error)
  • func (cdc *CDCPipeline) Run(ctx context.Context) error
• type PipelineManager
  • func NewPipelineManager(confPath string) (*PipelineManager, error)
  • func (pm *PipelineManager) Start()
• type PipelineState
• type StateTracker
  • func NewStateTracker(pname string, conf *config.CDCConfig) (*StateTracker, error)

Types

type CDCKafkaSink

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

func NewCDCKafkaSink

func NewCDCKafkaSink(conf *config.KafkaSink) *CDCKafkaSink

func (*CDCKafkaSink) Close

func (k *CDCKafkaSink) Close()

func (*CDCKafkaSink) Write

func (k *CDCKafkaSink) Write(msgs []cdc_event.CDCEvent, ctx context.Context) error

type CDCPipeline

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

func NewCDCPipeline

func NewCDCPipeline(name string, conf *config.CDCConfig, wg *sync.WaitGroup) CDCPipeline

func (*CDCPipeline) Close

func (cdc *CDCPipeline) Close()

func (*CDCPipeline) Init

func (cdc *CDCPipeline) Init() error

func (*CDCPipeline) Query

func (cdc *CDCPipeline) Query(query string) (*mysql.Result, error)

func (*CDCPipeline) Run

func (cdc *CDCPipeline) Run(ctx context.Context) error

type PipelineManager

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

func NewPipelineManager

func NewPipelineManager(confPath string) (*PipelineManager, error)

func (*PipelineManager) Start

func (pm *PipelineManager) Start()

type PipelineState

type PipelineState struct {
	Gtid      string `json:"gtid"`
	Name      string `json:"name"`
	Timestamp int64  `json:"timestamp"`
}

type StateTracker

type StateTracker struct {
	State  *PipelineState
	Config *config.StateTracking
	// contains filtered or unexported fields
}

func NewStateTracker

func NewStateTracker(pname string, conf *config.CDCConfig) (*StateTracker, error)