avro

README

go-kafka-avro

go-kafka-avro implements Avro encoding and decoding compatible with Confluent Schema Registry.

Installation

Install go 1.4 (or higher) http://golang.org/doc/install

$ go get github.com/dejan/go-kafka-avro

Documentation

Index

• Constants
• Variables
• type CachedSchemaRegistryClient
  • func NewCachedSchemaRegistryClient(registryURL string) *CachedSchemaRegistryClient
  • func NewCachedSchemaRegistryClientAuth(registryURL string, auth *KafkaAvroAuth) *CachedSchemaRegistryClient
  • func (this *CachedSchemaRegistryClient) GetByID(id int32) (avro.Schema, error)
  • func (this *CachedSchemaRegistryClient) GetLatestSchemaMetadata(subject string) (*SchemaMetadata, error)
  • func (this *CachedSchemaRegistryClient) GetVersion(subject string, schema avro.Schema) (int32, error)
  • func (this *CachedSchemaRegistryClient) Register(subject string, schema avro.Schema) (int32, error)
• type CompatibilityLevel
• type ErrorMessage
  • func (this *ErrorMessage) Error() string
• type GetSchemaResponse
• type GetSubjectVersionResponse
• type KafkaAvroAuth
  • func NewAuth(user string, key string) *KafkaAvroAuth
• type KafkaAvroDecoder
  • func NewKafkaAvroDecoder(url string) *KafkaAvroDecoder
  • func NewKafkaAvroDecoderAuth(url string, auth *KafkaAvroAuth) *KafkaAvroDecoder
  • func (this *KafkaAvroDecoder) Decode(bytes []byte) (interface{}, error)
  • func (this *KafkaAvroDecoder) DecodeSpecific(bytes []byte, value interface{}) error
• type KafkaAvroEncoder
  • func NewKafkaAvroEncoder(url string) *KafkaAvroEncoder
  • func NewKafkaAvroEncoderAuth(url string, auth *KafkaAvroAuth) *KafkaAvroEncoder
  • func (this *KafkaAvroEncoder) Encode(obj interface{}) ([]byte, error)
• type RegisterSchemaResponse
• type SchemaMetadata
• type SchemaRegistryClient

Constants

const (
	GET_SCHEMA_BY_ID             = "/schemas/ids/%d"
	GET_SUBJECTS                 = "/subjects"
	GET_SUBJECT_VERSIONS         = "/subjects/%s/versions"
	GET_SPECIFIC_SUBJECT_VERSION = "/subjects/%s/versions/%s"
	REGISTER_NEW_SCHEMA          = "/subjects/%s/versions"
	CHECK_IS_REGISTERED          = "/subjects/%s"
	TEST_COMPATIBILITY           = "/compatibility/subjects/%s/versions/%s"
	CONFIG                       = "/config"
)

const (
	SCHEMA_REGISTRY_V1_JSON               = "application/vnd.schemaregistry.v1+json"
	SCHEMA_REGISTRY_V1_JSON_WEIGHTED      = "application/vnd.schemaregistry.v1+json"
	SCHEMA_REGISTRY_MOST_SPECIFIC_DEFAULT = "application/vnd.schemaregistry.v1+json"
	SCHEMA_REGISTRY_DEFAULT_JSON          = "application/vnd.schemaregistry+json"
	SCHEMA_REGISTRY_DEFAULT_JSON_WEIGHTED = "application/vnd.schemaregistry+json qs=0.9"
	JSON                                  = "application/json"
	JSON_WEIGHTED                         = "application/json qs=0.5"
	GENERIC_REQUEST                       = "application/octet-stream"
)

Variables

var PREFERRED_RESPONSE_TYPES = []string{SCHEMA_REGISTRY_V1_JSON, SCHEMA_REGISTRY_DEFAULT_JSON, JSON}

Types

type CachedSchemaRegistryClient

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

func NewCachedSchemaRegistryClient

func NewCachedSchemaRegistryClient(registryURL string) *CachedSchemaRegistryClient

func NewCachedSchemaRegistryClientAuth

func NewCachedSchemaRegistryClientAuth(registryURL string, auth *KafkaAvroAuth) *CachedSchemaRegistryClient

func (*CachedSchemaRegistryClient) GetByID

func (this *CachedSchemaRegistryClient) GetByID(id int32) (avro.Schema, error)

func (*CachedSchemaRegistryClient) GetLatestSchemaMetadata

func (this *CachedSchemaRegistryClient) GetLatestSchemaMetadata(subject string) (*SchemaMetadata, error)

func (*CachedSchemaRegistryClient) GetVersion

func (this *CachedSchemaRegistryClient) GetVersion(subject string, schema avro.Schema) (int32, error)

func (*CachedSchemaRegistryClient) Register

func (this *CachedSchemaRegistryClient) Register(subject string, schema avro.Schema) (int32, error)

type CompatibilityLevel

type CompatibilityLevel string

const (
	BackwardCompatibilityLevel CompatibilityLevel = "BACKWARD"
	ForwardCompatibilityLevel  CompatibilityLevel = "FORWARD"
	FullCompatibilityLevel     CompatibilityLevel = "FULL"
	NoneCompatibilityLevel     CompatibilityLevel = "NONE"
)

type ErrorMessage

type ErrorMessage struct {
	Error_code int32
	Message    string
}

func (*ErrorMessage) Error

func (this *ErrorMessage) Error() string

type GetSchemaResponse

type GetSchemaResponse struct {
	Schema string
}

type GetSubjectVersionResponse

type GetSubjectVersionResponse struct {
	Subject string
	Version int32
	Id      int32
	Schema  string
}

type KafkaAvroAuth

type KafkaAvroAuth struct {
	User string
	Key  string
}

func NewAuth

func NewAuth(user string, key string) *KafkaAvroAuth

type KafkaAvroDecoder

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

func NewKafkaAvroDecoder

func NewKafkaAvroDecoder(url string) *KafkaAvroDecoder

func NewKafkaAvroDecoderAuth

func NewKafkaAvroDecoderAuth(url string, auth *KafkaAvroAuth) *KafkaAvroDecoder

func (*KafkaAvroDecoder) Decode

func (this *KafkaAvroDecoder) Decode(bytes []byte) (interface{}, error)

func (*KafkaAvroDecoder) DecodeSpecific

func (this *KafkaAvroDecoder) DecodeSpecific(bytes []byte, value interface{}) error

type KafkaAvroEncoder

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

func NewKafkaAvroEncoder

func NewKafkaAvroEncoder(url string) *KafkaAvroEncoder

func NewKafkaAvroEncoderAuth

func NewKafkaAvroEncoderAuth(url string, auth *KafkaAvroAuth) *KafkaAvroEncoder

func (*KafkaAvroEncoder) Encode

func (this *KafkaAvroEncoder) Encode(obj interface{}) ([]byte, error)

type RegisterSchemaResponse

type RegisterSchemaResponse struct {
	Id int32
}

type SchemaMetadata

type SchemaMetadata struct {
	Id      int32
	Version int32
	Schema  string
}

type SchemaRegistryClient

type SchemaRegistryClient interface {
	Register(subject string, schema avro.Schema) (int32, error)
	GetByID(id int32) (avro.Schema, error)
	GetLatestSchemaMetadata(subject string) (*SchemaMetadata, error)
	GetVersion(subject string, schema avro.Schema) (int32, error)
}