codec

Documentation

Index

• Constants
• func InitMetrics(registry *prometheus.Registry)
• func IsColumnValueEqual(preValue, updatedValue interface{}) bool
• func NewEncoderGroup(cfg *config.SinkConfig, builder RowEventEncoderBuilder, ...) *encoderGroup
• type DDLEventBatchEncoder
• type EncoderGroup
• type MessageBuilder
• type MockRowEventEncoder
  • func (m *MockRowEventEncoder) AppendRowChangedEvent(ctx context.Context, tableID string, event *model.RowChangedEvent, ...) error
  • func (m *MockRowEventEncoder) Build() []*common.Message
  • func (m *MockRowEventEncoder) EncodeCheckpointEvent(ts uint64) (*common.Message, error)
  • func (m *MockRowEventEncoder) EncodeDDLEvent(e *model.DDLEvent) (*common.Message, error)
• type MockRowEventEncoderBuilder
  • func (m *MockRowEventEncoderBuilder) Build() RowEventEncoder
  • func (m *MockRowEventEncoderBuilder) CleanMetrics()
• type RowEventDecoder
• type RowEventEncoder
• type RowEventEncoderBuilder
• type TxnEventEncoder
• type TxnEventEncoderBuilder

Constants

const (
	// BatchVersion1 represents the version of batch format
	BatchVersion1 uint64 = 1

	// MemBufShrinkThreshold represents the threshold of shrinking the buffer.
	MemBufShrinkThreshold = 1024 * 1024
)

Functions

func InitMetrics

func InitMetrics(registry *prometheus.Registry)

InitMetrics registers all metrics in this file

func IsColumnValueEqual

func IsColumnValueEqual(preValue, updatedValue interface{}) bool

IsColumnValueEqual checks whether the preValue and updatedValue are equal.

func NewEncoderGroup

func NewEncoderGroup(
	cfg *config.SinkConfig,
	builder RowEventEncoderBuilder,
	changefeedID model.ChangeFeedID,
) *encoderGroup

NewEncoderGroup creates a new EncoderGroup instance

Types

type DDLEventBatchEncoder

type DDLEventBatchEncoder interface {
	// EncodeCheckpointEvent appends a checkpoint event into the batch.
	// This event will be broadcast to all partitions to signal a global checkpoint.
	EncodeCheckpointEvent(ts uint64) (*common.Message, error)
	// EncodeDDLEvent appends a DDL event into the batch
	EncodeDDLEvent(e *model.DDLEvent) (*common.Message, error)
}

DDLEventBatchEncoder is an abstraction for DDL event encoder.

type EncoderGroup

type EncoderGroup interface {
	// Run start the group
	Run(ctx context.Context) error
	// AddEvents add events into the group and encode them by one of the encoders in the group.
	// Note: The caller should make sure all events should belong to the same topic and partition.
	AddEvents(ctx context.Context, key model.TopicPartitionKey, events ...*dmlsink.RowChangeCallbackableEvent) error
	// Output returns a channel produce futures
	Output() <-chan *future
}

EncoderGroup manages a group of encoders

type MessageBuilder

type MessageBuilder interface {
	// Build builds the batch and returns the bytes of key and value.
	// Should be called after `AppendRowChangedEvent`
	Build() []*common.Message
}

MessageBuilder is an abstraction to build message.

type MockRowEventEncoder

type MockRowEventEncoder struct{}

MockRowEventEncoder is a mock implementation of RowEventEncoder

func (*MockRowEventEncoder) AppendRowChangedEvent

func (m *MockRowEventEncoder) AppendRowChangedEvent(
	ctx context.Context, tableID string, event *model.RowChangedEvent, callback func(),
) error

AppendRowChangedEvent implement the RowEventEncoder interface

func (*MockRowEventEncoder) Build

func (m *MockRowEventEncoder) Build() []*common.Message

Build implement the RowEventEncoder interface

func (*MockRowEventEncoder) EncodeCheckpointEvent

func (m *MockRowEventEncoder) EncodeCheckpointEvent(ts uint64) (*common.Message, error)

EncodeCheckpointEvent implement the DDLEventBatchEncoder interface

func (*MockRowEventEncoder) EncodeDDLEvent

func (m *MockRowEventEncoder) EncodeDDLEvent(e *model.DDLEvent) (*common.Message, error)

EncodeDDLEvent implement the DDLEventBatchEncoder interface

type MockRowEventEncoderBuilder

type MockRowEventEncoderBuilder struct{}

MockRowEventEncoderBuilder is a mock implementation of RowEventEncoderBuilder

func (*MockRowEventEncoderBuilder) Build

func (m *MockRowEventEncoderBuilder) Build() RowEventEncoder

Build implement the RowEventEncoderBuilder interface

func (*MockRowEventEncoderBuilder) CleanMetrics

func (m *MockRowEventEncoderBuilder) CleanMetrics()

CleanMetrics implement the RowEventEncoderBuilder interface

type RowEventDecoder

type RowEventDecoder interface {
	// AddKeyValue add the received key and values to the decoder,
	// should be called before `HasNext`
	// decoder decode the key and value into the event format.
	AddKeyValue(key, value []byte) error

	// HasNext returns
	//     1. the type of the next event
	//     2. a bool if the next event is exist
	//     3. error
	HasNext() (model.MessageType, bool, error)
	// NextResolvedEvent returns the next resolved event if exists
	NextResolvedEvent() (uint64, error)
	// NextRowChangedEvent returns the next row changed event if exists
	NextRowChangedEvent() (*model.RowChangedEvent, error)
	// NextDDLEvent returns the next DDL event if exists
	NextDDLEvent() (*model.DDLEvent, error)
}

RowEventDecoder is an abstraction for events decoder this interface is only for testing now

type RowEventEncoder

type RowEventEncoder interface {
	DDLEventBatchEncoder
	// AppendRowChangedEvent appends a row changed event into the batch or buffer.
	AppendRowChangedEvent(context.Context, string, *model.RowChangedEvent, func()) error
	MessageBuilder
}

RowEventEncoder is an abstraction for events encoder

type RowEventEncoderBuilder

type RowEventEncoderBuilder interface {
	Build() RowEventEncoder
	CleanMetrics()
}

RowEventEncoderBuilder builds row encoder with context.

type TxnEventEncoder

type TxnEventEncoder interface {
	// AppendTxnEvent append a txn event into the buffer.
	AppendTxnEvent(*model.SingleTableTxn, func()) error
	MessageBuilder
}

TxnEventEncoder is an abstraction for txn events encoder.

type TxnEventEncoderBuilder

type TxnEventEncoderBuilder interface {
	Build() TxnEventEncoder
}

TxnEventEncoderBuilder builds txn encoder with context.