factory

Documentation

Index

• Constants
• type Category
• type SinkFactory
  • func New(ctx context.Context, changefeedID model.ChangeFeedID, sinkURIStr string, ...) (*SinkFactory, error)
  • func (s *SinkFactory) Category() Category
  • func (s *SinkFactory) Close()
  • func (s *SinkFactory) CreateTableSink(changefeedID model.ChangeFeedID, span tablepb.Span, startTs model.Ts, ...) tablesink.TableSink
  • func (s *SinkFactory) CreateTableSinkForConsumer(changefeedID model.ChangeFeedID, span tablepb.Span, startTs model.Ts) tablesink.TableSink

Constants

const (
	// CategoryTxn is for Txn sink.
	CategoryTxn Category = 1
	// CategoryMQ is for MQ sink.
	CategoryMQ = 2
	// CategoryCloudStorage is for CloudStorage sink.
	CategoryCloudStorage = 3
	// CategoryBlackhole is for Blackhole sink.
	CategoryBlackhole = 4
)

Types

type Category

type Category = int

Category is for different DML sink categories.

type SinkFactory

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

SinkFactory is the factory of sink. It is responsible for creating sink and closing it. Because there is no way to convert the eventsink.EventSink[*model.RowChangedEvent] to eventsink.EventSink[eventsink.TableEvent]. So we have to use this factory to create and store the sink.

func New

func New(
	ctx context.Context,
	changefeedID model.ChangeFeedID,
	sinkURIStr string,
	cfg *config.ReplicaConfig,
	errCh chan error,
	pdClock pdutil.Clock,
) (*SinkFactory, error)

New creates a new SinkFactory by scheme.

func (*SinkFactory) Category

func (s *SinkFactory) Category() Category

Category returns category of s.

func (*SinkFactory) Close

func (s *SinkFactory) Close()

Close closes the sink.

func (*SinkFactory) CreateTableSink

func (s *SinkFactory) CreateTableSink(
	changefeedID model.ChangeFeedID,
	span tablepb.Span,
	startTs model.Ts,
	PDClock pdutil.Clock,
	totalRowsCounter prometheus.Counter,
	flushLagDuration prometheus.Observer,
) tablesink.TableSink

CreateTableSink creates a TableSink by schema.

func (*SinkFactory) CreateTableSinkForConsumer

func (s *SinkFactory) CreateTableSinkForConsumer(
	changefeedID model.ChangeFeedID,
	span tablepb.Span, startTs model.Ts,
) tablesink.TableSink

CreateTableSinkForConsumer creates a TableSink by schema for consumer. The difference between CreateTableSink and CreateTableSinkForConsumer is that CreateTableSinkForConsumer will not create a new sink for each table. NOTICE: This only used for the consumer. Please do not use it in the processor.