source

Aug 16, 2022

Changes in this version

type Source

+ func (s *Source) SetMaxIngestRate(rate int)

Jul 6, 2022

Changes in this version

+ var MaxVarCharOverride = -1

Jun 23, 2022

Changes in this version

+ func CoerceDecimal(val interface{}) (*common.Decimal, error)

+ func CoerceFloat64(val interface{}) (float64, error)

+ func CoerceInt64(val interface{}) (int64, error)

+ func CoerceString(val interface{}) (string, error)

+ func CoerceTimestamp(val interface{}) (common.Timestamp, error)

+ func GenerateGroupID(clusterID uint64, sourceInfo *common.SourceInfo) string

+ type Decoder interface

+ Decode func(bytes []byte) (interface{}, error)

+ type IngestLimiter interface

+ Limit func()

+ type JSONDecoder struct

+ func (j *JSONDecoder) Decode(bytes []byte) (interface{}, error)

+ type KafkaDecoder struct

+ func (k *KafkaDecoder) Decode(bytes []byte) (interface{}, error)

+ type MessageConsumer struct

+ func NewMessageConsumer(msgProvider kafka.MessageProvider, pollTimeout time.Duration, maxMessages int, ...) (*MessageConsumer, error)

+ func (m *MessageConsumer) Close() error

+ func (m *MessageConsumer) Stop() error

+ type MessageParser struct

+ func NewMessageParser(sourceInfo *common.SourceInfo, registry protolib.Resolver) (*MessageParser, error)

+ func (m *MessageParser) ParseMessages(messages []*kafka.Message) (*common.Rows, error)

+ type ProtobufDecoder struct

+ func (p *ProtobufDecoder) Decode(bytes []byte) (interface{}, error)

+ type RowProcessor interface

+ type Source struct

+ func NewSource(sourceInfo *common.SourceInfo, tableExec *exec.TableExecutor, ...) (*Source, error)

+ func (s *Source) AddConsumingExecutor(mvName string, executor exec.PushExecutor)

+ func (s *Source) Drop() error

+ func (s *Source) GetCommittedCount() int64

+ func (s *Source) GetConsumingMVs() []string

+ func (s *Source) IsRunning() bool

+ func (s *Source) RemoveConsumingExecutor(mvName string)

+ func (s *Source) SetCommitOffsets(enable bool)

+ func (s *Source) Start() error

+ func (s *Source) Stop() error

+ func (s *Source) TableExecutor() *exec.TableExecutor