lib

Documentation

Index

• func Marshal(header []byte, msg proto.Message) ([]byte, error)
• func Unmarshal(bytes []byte, msg proto.Message) error
• type DeserializationFunc
• type DeserializationType
• type InitializationFunc
• type JsonDeserializer
  • func NewJsonDeserializer(unmarshal func(topic *string, data []byte) (interface{}, error)) *JsonDeserializer
  • func (s *JsonDeserializer) Deserialize(topic *string, bytes []byte) (interface{}, error)
• type JsonSerializer
  • func (s *JsonSerializer) Serialize(topic string, thing interface{}) ([]byte, error)
• type ProtobufDeserializer
  • func NewProtobufDeserializer(protobufResolver ProtobufResolver, deserializationFunc ...DeserializationFunc) *ProtobufDeserializer
  • func (ps *ProtobufDeserializer) Deserialize(topic *string, bytes []byte) (interface{}, error)
• type ProtobufRegistry
• type ProtobufResolver
• type ProtobufSerializer
  • func NewProtobufSerializer(schemaLookupStrategy SchemaResolver, initialize InitializationFunc, ...) *ProtobufSerializer
  • func (ps *ProtobufSerializer) Serialize(topic string, thing interface{}) ([]byte, error)
• type ProtoregistryTopicNameProtobufResolver
  • func NewProtoregistryTopicNameProtobufResolver(registry ProtobufRegistry, deserializationType DeserializationType) *ProtoregistryTopicNameProtobufResolver
  • func (reg *ProtoregistryTopicNameProtobufResolver) ResolveProtobuf(topic *string, schemaId uint32, msgIndexes []int64) (proto.Message, error)
• type SchemaRegistryClient
• type SchemaRegistryProtobufResolver
  • func NewSchemaRegistryProtobufResolver(schemaRegistry srclient.SchemaRegistryClient, ...) *SchemaRegistryProtobufResolver
  • func (reg *SchemaRegistryProtobufResolver) ResolveProtobuf(topic *string, schemaId uint32, msgIndexes []int64) (proto.Message, error)
• type SchemaResolver
• type SerializationFunc
• type SerializationType
• type StringDeserializer
  • func (s *StringDeserializer) Deserialize(topic *string, bytes []byte) (interface{}, error)
• type StringSerializer
  • func (s *StringSerializer) Serialize(topic string, thing interface{}) ([]byte, error)
• type TopicNameSchemaResolver
  • func NewTopicNameSchemaResolver(client SchemaRegistryClient, serializationType SerializationType) *TopicNameSchemaResolver
  • func (ls *TopicNameSchemaResolver) ResolveSchema(topic string, msg proto.Message) (*srclient.Schema, error)
• type VTMarshal
• type VTUnmarshal

Functions

func Marshal

func Marshal(header []byte, msg proto.Message) ([]byte, error)

Marshal is a wrapper around proto which will use MarshalToVT if that method is available in the proto, which serializes much more rapidly than the reflection-based proto.Marshal

func Unmarshal

func Unmarshal(bytes []byte, msg proto.Message) error

Unmarshal is a wrapper around proto.Unmarshal which will use UnmarshalVT when deserializing any proto that has been modified by protoc-gen-go-vtproto with the unmarshal option

Types

type DeserializationFunc

type DeserializationFunc = func([]byte, proto.Message) error

DeserializationFunc is a type that describes the function that is ultimately used to deserialize a protobuf.

type DeserializationType

type DeserializationType int

DeserializationType is a type alias for representing Key and Value deserialization types

const (
	KeyDeserialization DeserializationType = iota
	ValueDeserialization
)

type InitializationFunc

type InitializationFunc = func(proto.Message)

InitializationFunc is a type that describes a function to be used to initialize a messsage prior to serialization.

type JsonDeserializer

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

func NewJsonDeserializer

func NewJsonDeserializer(unmarshal func(topic *string, data []byte) (interface{}, error)) *JsonDeserializer

func (*JsonDeserializer) Deserialize

func (s *JsonDeserializer) Deserialize(topic *string, bytes []byte) (interface{}, error)

type JsonSerializer

type JsonSerializer struct {
}

func (*JsonSerializer) Serialize

func (s *JsonSerializer) Serialize(topic string, thing interface{}) ([]byte, error)

type ProtobufDeserializer

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

ProtobufDeserializer hydrates a []byte into a Protobuf which is resolved via a ProtobufResolver

func NewProtobufDeserializer

func NewProtobufDeserializer(
	protobufResolver ProtobufResolver,
	deserializationFunc ...DeserializationFunc,
) *ProtobufDeserializer

NewProtobufDeserializer is a constructor that takes a SchemaRegistryClient and a ProtobufResolver, which are used to determine schema and resolve an empty protobuf that data can be unmarshalled into.

func (*ProtobufDeserializer) Deserialize

func (ps *ProtobufDeserializer) Deserialize(
	topic *string,
	bytes []byte,
) (interface{}, error)

Deserialize hydrates an []byte into a protobuf instance which is resolved from the topic name and schemaId by the ProtobufResolver

type ProtobufRegistry

type ProtobufRegistry interface {
	RangeMessages(f func(protoreflect.MessageType) bool)
	FindMessageByName(message protoreflect.FullName) (protoreflect.MessageType, error)
}

ProtobufRegistry is the minimum interface of protoregistry.Types registry needed to resolve MessageType from topic name (plus a registration function, for convenience)

type ProtobufResolver

type ProtobufResolver interface {
	ResolveProtobuf(topic *string, schemaId uint32, msgIndexes []int64) (proto.Message, error)
}

ProtobufResolver is an interface which can resolve a protobuf MessageDescriptor from topic name and the info contained in the message header and instantiate an instance of the message described by the MessageDescriptor

type ProtobufSerializer

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

ProtobufSerializer is an instance of Serializer which serializes protobufs according to the confluent schema registry line protocol

func NewProtobufSerializer

func NewProtobufSerializer(
	schemaLookupStrategy SchemaResolver,
	initialize InitializationFunc,
	serializationFunc ...SerializationFunc,
) *ProtobufSerializer

NewProtobufSerializer is a constructor function for ProtobufSerializer. Receives a SchemaResolver as parameter.

func (*ProtobufSerializer) Serialize

func (ps *ProtobufSerializer) Serialize(
	topic string,
	thing interface{},
) ([]byte, error)

Serialize encodes a protobuf for the specified topic.

type ProtoregistryTopicNameProtobufResolver

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

ProtoregistryTopicNameProtobufResolver is a concrete implementation of ProtobufResolver which uses topic name in combination with protoregistry to resolve a protoreflect.MessageType that can then be used to instantiate a new instance of that type

func NewProtoregistryTopicNameProtobufResolver

func NewProtoregistryTopicNameProtobufResolver(
	registry ProtobufRegistry,
	deserializationType DeserializationType,
) *ProtoregistryTopicNameProtobufResolver

NewProtoregistryTopicNameProtobufResolver is a constructor

func (*ProtoregistryTopicNameProtobufResolver) ResolveProtobuf

func (reg *ProtoregistryTopicNameProtobufResolver) ResolveProtobuf(
	topic *string,
	schemaId uint32,
	msgIndexes []int64,
) (proto.Message, error)

ResolveProtobuf uses topic name in combination with protorregistry to find protoreflect.MessageType that matches. It then instantiates a new instance ot that type and returns it.

type SchemaRegistryClient

type SchemaRegistryClient interface {
	GetSchema(schemaID int) (*srclient.Schema, error)
	GetLatestSchema(subject string) (*srclient.Schema, error)
	SetCredentials(username string, password string)
}

Simplified interface for srclient.SchemaRegistryClient

type SchemaRegistryProtobufResolver

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

SchemaRegistryProtobufResolver

func NewSchemaRegistryProtobufResolver

func NewSchemaRegistryProtobufResolver(
	schemaRegistry srclient.SchemaRegistryClient,
	protobufRegistry ProtobufRegistry,
	deserializationType DeserializationType,
) *SchemaRegistryProtobufResolver

NewSchemaRegistryProtobufResolver

func (*SchemaRegistryProtobufResolver) ResolveProtobuf

func (reg *SchemaRegistryProtobufResolver) ResolveProtobuf(
	topic *string,
	schemaId uint32,
	msgIndexes []int64,
) (proto.Message, error)

ResolveProtobuf

type SchemaResolver

type SchemaResolver interface {
	ResolveSchema(topic string, msg proto.Message) (*srclient.Schema, error)
}

SchemaResolver is an interface that can resolve a schema registry schema from destination topic and the entity being serialized. It is analogous to the SubjectNameStrategy in confluent clients, but also performs the registry schema lookup.

type SerializationFunc

type SerializationFunc = func([]byte, proto.Message) ([]byte, error)

SerializationFunc is a type that describes the function that is ultimately used to serialize a protobuf.

type SerializationType

type SerializationType int

SerializationType is a type alias for representing Key and Value serialization types

const (
	KeySerialization SerializationType = iota
	ValueSerialization
)

type StringDeserializer

type StringDeserializer struct {
}

func (*StringDeserializer) Deserialize

func (s *StringDeserializer) Deserialize(topic *string, bytes []byte) (interface{}, error)

type StringSerializer

type StringSerializer struct {
}

func (*StringSerializer) Serialize

func (s *StringSerializer) Serialize(topic string, thing interface{}) ([]byte, error)

type TopicNameSchemaResolver

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

TopicNameSchemaResolver is an instance of SchemaResolver which uses the topic name as the subject when looking up schema via schema registry

func NewTopicNameSchemaResolver

func NewTopicNameSchemaResolver(
	client SchemaRegistryClient,
	serializationType SerializationType,
) *TopicNameSchemaResolver

NewTopicNameSchemaResolver is a constructor for TopicNameSchemaResolver. Receives a SchemaRegistryClient, which should have caching enabled as schema is resolved for every serialization performed by a serializer, as well as a SerializationType, which specifies whether to resolve a key or value schema for the topic

func (*TopicNameSchemaResolver) ResolveSchema

func (ls *TopicNameSchemaResolver) ResolveSchema(
	topic string,
	msg proto.Message,
) (*srclient.Schema, error)

ResolveSchema using the TopicNameStrategy, which uses the topic name as the subject. Ensure the schema registry client that was pass to the constructor has caching enabled or this will be slow to execute

type VTMarshal

type VTMarshal interface {
	SizeVT() int
	MarshalToVT(data []byte) (int, error)
}

VTMarshal is an interface that will be satisfied by any protobuf that has had the protoc-gen-go-vtproto plugin applied to it with the marshal and size options. If a proto satisfies this interface, the Marshal function will apply the much more efficient MarshalToVT serialization

type VTUnmarshal

type VTUnmarshal interface {
	UnmarshalVT(data []byte) error
}

VTUnmarshal is an inerface satisfied by any protobuf that has been built with the protoc-gen-go-vtproto tool to generate an efficient unmarshal method