sr

Documentation

Overview

Package sr is a schema registry client for Redpanda for usage within inline Data Transforms.

Index

• Variables
• func ExtractID(b []byte) (int, error)
• type ClientOpt
  • func MaxCacheEntries(size int) ClientOpt
• type Reference
• type Schema
• type SchemaRegistryClient
  • func NewClient(opts ...ClientOpt) (c SchemaRegistryClient)
• type SchemaType
• type Serde
  • func (s *Serde[T]) AppendEncode(b []byte, v T) ([]byte, error)
  • func (s *Serde[T]) Decode(b []byte, v T) error
  • func (s *Serde[T]) Encode(v T) ([]byte, error)
  • func (s *Serde[T]) MustAppendEncode(b []byte, v T) []byte
  • func (s *Serde[T]) MustEncode(v T) []byte
  • func (s *Serde[T]) Register(id int, opts ...SerdeOpt[T])
  • func (s *Serde[T]) SetDefaults(opts ...SerdeOpt[T])
• type SerdeOpt
  • func AppendEncodeFn[T any](fn func([]byte, T) ([]byte, error)) SerdeOpt[T]
  • func DecodeFn[T any](fn func([]byte, T) error) SerdeOpt[T]
  • func EncodeFn[T any](fn func(T) ([]byte, error)) SerdeOpt[T]
• type SubjectSchema

Variables

var (
	// ErrNotRegistered is returned from Serde when attempting to encode a
	// value or decode an ID that has not been registered, or when using
	// Decode with a missing new value function.
	ErrNotRegistered = errors.New("registration is missing for encode/decode")

	// ErrBadHeader is returned from Decode when the input slice is shorter
	// than five bytes, or if the first byte is not the magic 0 byte.
	ErrBadHeader = errors.New("5 byte header for value is missing or does not have the 0 magic byte")
)

Functions

func ExtractID

func ExtractID(b []byte) (int, error)

Extract the ID from the header of a schema registry encoded value.

Returns ErrBadHeader if the array is missing the leading magic byte or is too small.

Types

type ClientOpt

type ClientOpt interface {
	// contains filtered or unexported methods
}

ClientOpt is an option to configure a SchemaRegistryClient

func MaxCacheEntries

func MaxCacheEntries(size int) ClientOpt

MaxCacheEntries configures how many entries to cache within the client.

By default the cache is unbounded, use 0 to disable the cache.

type Reference

type Reference struct {
	Name    string
	Subject string
	Version int
}

SchemaReference is a way for a one schema to reference another. The details for how referencing is done are type specific; for example, JSON objects that use the key "$ref" can refer to another schema via URL. For more details on references, see the following link:

https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html#schema-references
https://docs.confluent.io/platform/current/schema-registry/develop/api.html

type Schema

type Schema struct {
	Schema     string
	Type       SchemaType
	References []Reference
}

Schema is a schema that can be registered within schema registry

type SchemaRegistryClient

type SchemaRegistryClient interface {
	// LookupSchemaById looks up a schema via it's global ID.
	LookupSchemaById(id int) (s *Schema, err error)
	// LookupSchemaByVersion looks up a schema via a subject for a specific version.
	//
	// Use version -1 to get the latest version.
	LookupSchemaByVersion(subject string, version int) (s *SubjectSchema, err error)
	// CreateSchema creates a schema under the given subject, returning the version and ID.
	//
	// If an equivalent schema already exists globally, that schema ID will be reused.
	// If an equivalent schema already exists within that subject, this will be a noop and the
	// existing schema will be returned.
	CreateSchema(subject string, schema Schema) (s *SubjectSchema, err error)
}

SchemaRegistryClient is a client for interacting with the schema registry within Redpanda.

The client provides caching out of the box, which can be configured with options.

func NewClient

func NewClient(opts ...ClientOpt) (c SchemaRegistryClient)

NewClient creates a new SchemaRegistryClient with the specified options applied.

type SchemaType

type SchemaType int

SchemaType is an enum for the different types of schemas that can be stored in schema registry.

const (
	TypeAvro SchemaType = iota
	TypeProtobuf
	TypeJSON
)

type Serde

type Serde[T any] struct {
	// contains filtered or unexported fields
}

Serde encodes and decodes values according to the schema registry wire format. A Serde itself does not perform schema auto-discovery and type auto-decoding. To aid in strong typing and validated encoding/decoding, you must register IDs and values to how to encode or decode them.

To use a Serde for encoding, you must pre-register schema ids and values you will encode, and then you can use the encode functions. The latest registered ID that supports encoding will be used to encode.

To use a Serde for decoding, you can either pre-register schema ids and values you will consume, or you can discover the schema every time you receive an ErrNotRegistered error from decode.

func (*Serde[T]) AppendEncode

func (s *Serde[T]) AppendEncode(b []byte, v T) ([]byte, error)

AppendEncode appends an encoded value to b according to the schema registry wire format and returns it. If EncodeFn was not used, this returns ErrNotRegistered.

func (*Serde[T]) Decode

func (s *Serde[T]) Decode(b []byte, v T) error

Decode decodes b into v. If DecodeFn option was not used, this returns ErrNotRegistered.

Serde does not handle references in schemas; it is up to you to register the full decode function for any top-level ID, regardless of how many other schemas are referenced in top-level ID.

func (*Serde[T]) Encode

func (s *Serde[T]) Encode(v T) ([]byte, error)

Encode encodes a value according to the schema registry wire format and returns it. If EncodeFn was not used, this returns ErrNotRegistered.

func (*Serde[T]) MustAppendEncode

func (s *Serde[T]) MustAppendEncode(b []byte, v T) []byte

MustAppendEncode returns the value of AppendEncode, panicking on error. This is a shortcut for if your encode function cannot error.

func (*Serde[T]) MustEncode

func (s *Serde[T]) MustEncode(v T) []byte

MustEncode returns the value of Encode, panicking on error. This is a shortcut for if your encode function cannot error.

func (*Serde[T]) Register

func (s *Serde[T]) Register(id int, opts ...SerdeOpt[T])

Register registers a schema ID and the value it corresponds to, as well as the encoding or decoding functions. You need to register functions depending on whether you are only encoding, only decoding, or both.

func (*Serde[T]) SetDefaults

func (s *Serde[T]) SetDefaults(opts ...SerdeOpt[T])

SetDefaults sets default options to apply to every registered type. These options are always applied first, so you can override them as necessary when registering.

This can be useful if you always want to use the same encoding or decoding functions.

type SerdeOpt

type SerdeOpt[T any] interface {
	// contains filtered or unexported methods
}

SerdeOpt is an option to configure a Serde.

func AppendEncodeFn

func AppendEncodeFn[T any](fn func([]byte, T) ([]byte, error)) SerdeOpt[T]

AppendEncodeFn allows Serde to encode a value to an existing slice. This can be more efficient than EncodeFn; this function is used if it exists.

func DecodeFn

func DecodeFn[T any](fn func([]byte, T) error) SerdeOpt[T]

DecodeFn allows Serde to decode into a value.

func EncodeFn

func EncodeFn[T any](fn func(T) ([]byte, error)) SerdeOpt[T]

EncodeFn allows Serde to encode a value.

type SubjectSchema

type SubjectSchema struct {
	Schema

	Subject string
	Version int
	ID      int
}

SchemaSubject is a schema along with the subject, version and ID of the schema