ksqldb_connector

Documentation

Index

• func DeleteTableQuery(data deleteMeterTableQueryData) (string, error)
• func GetTableDescribeQuery(meter *models.Meter, namespace string) (string, error)
• func GetTableQuery(data meterTableQueryData) (string, error)
• func GetTableValuesQuery(meter *models.Meter, params *streaming.GetValuesParams, namespace string) (string, error)
• func MeterQueryAssert(query string, data meterTableQueryData) error
• func NewMeterValues(header ksqldb.Header, payload ksqldb.Payload) ([]*models.MeterValue, error)
• type KsqlDBConnector
  • func NewKsqlDBConnector(ksqldbClient *ksqldb.KsqldbClient, partitions int, format string, ...) (*KsqlDBConnector, error)
  • func (c *KsqlDBConnector) CreateMeter(ctx context.Context, namespace string, meter *models.Meter) error
  • func (c *KsqlDBConnector) DeleteMeter(ctx context.Context, namespace string, meterSlug string) error
  • func (c *KsqlDBConnector) MeterAssert(ctx context.Context, data meterTableQueryData) error
  • func (c *KsqlDBConnector) QueryMeter(ctx context.Context, namespace string, meter *models.Meter, ...) ([]*models.MeterValue, error)
• type NamespaceHandler
  • func (h NamespaceHandler) CreateNamespace(ctx context.Context, namespace string) error

Functions

func DeleteTableQuery

func DeleteTableQuery(data deleteMeterTableQueryData) (string, error)

func GetTableDescribeQuery

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

func GetTableQuery

func GetTableQuery(data meterTableQueryData) (string, error)

func GetTableValuesQuery

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

func MeterQueryAssert

func MeterQueryAssert(query string, data meterTableQueryData) error

func NewMeterValues

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

Types

type KsqlDBConnector

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

func NewKsqlDBConnector

func NewKsqlDBConnector(ksqldbClient *ksqldb.KsqldbClient, partitions int, format string, logger *slog.Logger) (*KsqlDBConnector, error)

func (*KsqlDBConnector) CreateMeter

func (c *KsqlDBConnector) CreateMeter(ctx context.Context, namespace string, meter *models.Meter) error

func (*KsqlDBConnector) DeleteMeter

func (c *KsqlDBConnector) DeleteMeter(ctx context.Context, namespace string, meterSlug string) error

func (*KsqlDBConnector) MeterAssert

func (c *KsqlDBConnector) MeterAssert(ctx context.Context, data meterTableQueryData) error

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

func (*KsqlDBConnector) QueryMeter

func (c *KsqlDBConnector) QueryMeter(ctx context.Context, namespace string, meter *models.Meter, params *streaming.GetValuesParams) ([]*models.MeterValue, error)

type NamespaceHandler

type NamespaceHandler struct {
	KsqlDBClient *ksqldb.KsqldbClient

	// NamespacedTopicTemplate needs to contain at least one string parameter passed to fmt.Sprintf.
	// For example: "om_%s_events"
	NamespacedEventsTopicTemplate string

	// NamespacedDetectedEventsTopicTemplate needs to contain at least one string parameter passed to fmt.Sprintf.
	// For example: "om_%s_detected_events"
	NamespacedDetectedEventsTopicTemplate string

	Format        string
	KeySchemaID   int
	ValueSchemaID int
	Partitions    int
}

NamespaceHandler is a namespace handler for Kafka ingest topics.

func (NamespaceHandler) CreateNamespace

func (h NamespaceHandler) CreateNamespace(ctx context.Context, namespace string) error

CreateNamespace implements the namespace handler interface.