kschema

Documentation

Index

• Constants
• Variables
• type Codec
• type Field
• type FieldSchema
  • func (s FieldSchema) Validate(row Row) error
  • func (s FieldSchema) ValidateMessage(msg api.Message) error
  • func (s FieldSchema) ValidateRows(rows ...Row) error
• type FieldType
• type Message
  • func CopyMessage(msg api.Message) Message
  • func NewMessage(topic string, key, value []byte) *Message
  • func RawMessage(topic string, offset uint64, key, value []byte) Message
  • func (m *Message) Decode(data []byte) error
  • func (m *Message) Encode() ([]byte, error)
  • func (m *Message) Key() []byte
  • func (m *Message) MustEncode() []byte
  • func (m *Message) Offset() uint64
  • func (m *Message) String() string
  • func (m *Message) Topic() string
  • func (m *Message) Value() []byte
• type Row
  • func (r *Row) Decode(data []byte) error
  • func (r *Row) Decoded(data []byte) (*Row, error)
  • func (r *Row) Encode() ([]byte, error)
  • func (r *Row) GetKey() []byte
  • func (r *Row) GetValues() []any
• type Schema
  • func NewTableSchema(name string, fields ...Field) (*Schema, error)
  • func (t *Schema) GetTable() string
  • func (t *Schema) GetTopic() string
  • func (t *Schema) Validate() error
• type TableTopic

Constants

const (
	FieldTypeString  = "string"
	FieldTypeInt64   = "int64"
	FieldTypeFloat64 = "float64"
	FieldTypeBool    = "bool"
	FieldTypeRecord  = "record"
)

Variables

var (
	ErrorTooManyValues        = errors.New("row has more values than schema fields")
	ErrorInvalidFieldType     = errors.New("row value has invalid field type")
	ErrorUnsupportedFieldType = errors.New("unsupported field type")
	ErrorEmptyTableName       = errors.New("Table.Name must not be empty")
	ErrorEmptyTopicName       = errors.New("Table.Topic must not be empty")
)

Types

type Codec

type Codec interface {
	GetKey() []byte           // GetKey gets the key bytes
	Encode() ([]byte, error)  // Encode encodes the entire record as bytes (incl. the key)
	Decode(data []byte) error // Decode decodes the data bytes into the record (incl. the key)

}

type Field

type Field struct {
	Name string    `json:"name,omitempty"`
	Type FieldType `json:"type,omitempty"`
}

type FieldSchema

type FieldSchema []Field

func (FieldSchema) Validate

func (s FieldSchema) Validate(row Row) error

func (FieldSchema) ValidateMessage

func (s FieldSchema) ValidateMessage(msg api.Message) error

func (FieldSchema) ValidateRows

func (s FieldSchema) ValidateRows(rows ...Row) error

type FieldType

type FieldType string

type Message

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

Message is the defaulr message type used to create new concrete payloads and implements the api.Message interface.

func CopyMessage

func CopyMessage(msg api.Message) Message

func NewMessage

func NewMessage(topic string, key, value []byte) *Message

func RawMessage

func RawMessage(topic string, offset uint64, key, value []byte) Message

func (*Message) Decode

func (m *Message) Decode(data []byte) error

func (*Message) Encode

func (m *Message) Encode() ([]byte, error)

func (*Message) Key

func (m *Message) Key() []byte

func (*Message) MustEncode

func (m *Message) MustEncode() []byte

func (*Message) Offset

func (m *Message) Offset() uint64

func (*Message) String

func (m *Message) String() string

func (*Message) Topic

func (m *Message) Topic() string

func (*Message) Value

func (m *Message) Value() []byte

type Row

type Row struct {
	Key    []byte `json:"key,omitempty"`
	Values []any  `json:"values,omitempty"`
}

func (*Row) Decode

func (r *Row) Decode(data []byte) error

func (*Row) Decoded

func (r *Row) Decoded(data []byte) (*Row, error)

func (*Row) Encode

func (r *Row) Encode() ([]byte, error)

func (*Row) GetKey

func (r *Row) GetKey() []byte

func (*Row) GetValues

func (r *Row) GetValues() []any

type Schema

type Schema struct {
	Name   string      `json:"name,omitempty"`
	Topic  string      `json:"topic,omitempty"`
	Schema FieldSchema `json:"schema,omitempty"`
	// contains filtered or unexported fields
}

Schema idenfies and fully defines a kstore table.

func NewTableSchema

func NewTableSchema(name string, fields ...Field) (*Schema, error)

func (*Schema) GetTable

func (t *Schema) GetTable() string

func (*Schema) GetTopic

func (t *Schema) GetTopic() string

func (*Schema) Validate

func (t *Schema) Validate() error

type TableTopic

type TableTopic interface {
	Topic() string
	Table() string
}