kafka_connector

Documentation

Index

• func GetTableDescribeQuery(meter *models.Meter) (string, error)
• func GetTableQuery(data meterTableQueryData) (string, error)
• func GetTableValuesQuery(meter *models.Meter, params *GetValuesParams) (string, error)
• func MeterQueryAssert(query string, data meterTableQueryData) error
• func NewKafkaConnector(config *KafkaConnectorConfig) (Connector, error)
• func NewMeterValues(header ksqldb.Header, payload ksqldb.Payload) ([]*models.MeterValue, error)
• type KafkaConnector
  • func (c *KafkaConnector) GetValues(meter *models.Meter, params *GetValuesParams) ([]*models.MeterValue, error)
  • func (c *KafkaConnector) Init(meter *models.Meter) error
  • func (c *KafkaConnector) MeterAssert(data meterTableQueryData) error
• type KafkaConnectorConfig

Functions

func GetTableDescribeQuery

func GetTableDescribeQuery(meter *models.Meter) (string, error)

func GetTableQuery

func GetTableQuery(data meterTableQueryData) (string, error)

func GetTableValuesQuery

func GetTableValuesQuery(meter *models.Meter, params *GetValuesParams) (string, error)

func MeterQueryAssert

func MeterQueryAssert(query string, data meterTableQueryData) error

func NewKafkaConnector

func NewKafkaConnector(config *KafkaConnectorConfig) (Connector, error)

func NewMeterValues

func NewMeterValues(header ksqldb.Header, payload ksqldb.Payload) ([]*models.MeterValue, error)

Types

type KafkaConnector

type KafkaConnector struct {
	KsqlDBClient *ksqldb.KsqldbClient
	// contains filtered or unexported fields
}

func (*KafkaConnector) GetValues

func (c *KafkaConnector) GetValues(meter *models.Meter, params *GetValuesParams) ([]*models.MeterValue, error)

func (*KafkaConnector) Init

func (c *KafkaConnector) Init(meter *models.Meter) error

func (*KafkaConnector) MeterAssert

func (c *KafkaConnector) MeterAssert(data meterTableQueryData) error

MeterAssert ensures meter table immutability by checking that existing meter table is the same as new

type KafkaConnectorConfig

type KafkaConnectorConfig struct {
	KsqlDBClient ksqldb.KsqldbClient
	EventsTopic  string
	Partitions   int

	KeySchemaID   int
	ValueSchemaID int
}