moconnector

Documentation

Index

• Constants
• Variables
• func KafkaSinkConnectorExecutor(logger *zap.Logger, ts taskservice.TaskService, ...) func(context.Context, task.Task) error
• type Converter
• type Decoder
• type KafkaMoConnector
  • func NewKafkaMoConnector(logger *zap.Logger, options map[string]string, ie ie.InternalExecutor, ...) (*KafkaMoConnector, error)
  • func (k *KafkaMoConnector) Cancel() error
  • func (k *KafkaMoConnector) Close() error
  • func (k *KafkaMoConnector) Pause() error
  • func (k *KafkaMoConnector) Resume() error
  • func (k *KafkaMoConnector) Start(ctx context.Context) error
• type OptConstraint
• type OptType
• type OptValidator
• type RawObject
• type SQLConverter
  • func (c *SQLConverter) Convert(ctx context.Context, obj ie.InternalExecResult) (string, error)
• type StmtOpts
  • func MakeStmtOpts(ctx context.Context, opts map[string]string) (StmtOpts, error)

Constants

const (
	SourceKafka string = "kafka"
	FormatJson  string = "json"
)

const (
	OptConnectorType    = "type"
	OptConnectorServers = "bootstrap.servers"
	OptConnectorTopic   = "topic"
	OptConnectorValue   = "value"

	OptConnectorSql = "sql"

	OptConnectorRel       = "relkind"
	OptConnectorPartition = "partition"

	OptConnectorBufferLimit = "buffer_limit"
	OptConnectorTimeWindow  = "time_window"
)

Variables

var ConnectorEssentialOpts = map[string]struct{}{
	OptConnectorType: {},
}

var ConnectorEssentialTypeOpts = map[string]map[string]struct{}{
	"kafka": {
		OptConnectorServers: {},
		OptConnectorTopic:   {},
		OptConnectorValue:   {},
	},
}

var ConnectorOptConstraint = map[string]OptConstraint{
	OptConnectorType:        enumOpt(SourceKafka),
	OptConnectorServers:     addressOpt,
	OptConnectorTopic:       stringOpt,
	OptConnectorValue:       enumOpt(FormatJson),
	OptConnectorSql:         stringOpt,
	OptConnectorRel:         stringOpt,
	OptConnectorPartition:   integerOpt,
	OptConnectorBufferLimit: integerOpt,
	OptConnectorTimeWindow:  integerOpt,
}

Functions

func KafkaSinkConnectorExecutor

func KafkaSinkConnectorExecutor(
	logger *zap.Logger,
	ts taskservice.TaskService,
	ieFactory func() ie.InternalExecutor,
	attachToTask func(context.Context, uint64, taskservice.ActiveRoutine) error,
) func(context.Context, task.Task) error

Types

type Converter

type Converter interface {
	Convert(context.Context, ie.InternalExecResult) (string, error)
}

type Decoder

type Decoder interface {
	Decode([]byte) (RawObject, error)
}

type KafkaMoConnector

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

func NewKafkaMoConnector

func NewKafkaMoConnector(logger *zap.Logger, options map[string]string, ie ie.InternalExecutor, buffer_limit int) (*KafkaMoConnector, error)

func (*KafkaMoConnector) Cancel

func (k *KafkaMoConnector) Cancel() error

Cancel implements the taskservice.ActiveRoutine interface.

func (*KafkaMoConnector) Close

func (k *KafkaMoConnector) Close() error

func (*KafkaMoConnector) Pause

func (k *KafkaMoConnector) Pause() error

Pause implements the taskservice.ActiveRoutine interface.

func (*KafkaMoConnector) Resume

func (k *KafkaMoConnector) Resume() error

Resume implements the taskservice.ActiveRoutine interface.

func (*KafkaMoConnector) Start

func (k *KafkaMoConnector) Start(ctx context.Context) error

Start begins consuming messages from Kafka and writing them to the MO Table.

type OptConstraint

type OptConstraint struct {
	Type      OptType
	Validator OptValidator
}

type OptType

type OptType uint16

const (
	OptTypeString OptType = iota
	OptTypeEnum
	OptTypeInteger
	OptTypeAddress
)

type OptValidator

type OptValidator func(string) bool

type RawObject

type RawObject map[string]any

type SQLConverter

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

func (*SQLConverter) Convert

func (c *SQLConverter) Convert(ctx context.Context, obj ie.InternalExecResult) (string, error)

type StmtOpts

type StmtOpts map[string]string

func MakeStmtOpts

func MakeStmtOpts(ctx context.Context, opts map[string]string) (StmtOpts, error)