pulsavro

README

go-pulsar-avro

A library provides consumer/producer to work with pulsar, avro and schema registry

References

• go-kafka-avro
• pulsar-client-go

Documentation

Index

• func DecodeAvroFromBytes(schema string, payload []byte) (interface{}, error)
• func EncodeAvroToBytes(schema string, datum interface{}) ([]byte, error)
• type AvroConsumer
  • func NewAvroConsumer(client pulsar.Client, consumerOptions pulsar.ConsumerOptions, ...) (*AvroConsumer, error)
  • func (ac *AvroConsumer) Close()
  • func (ac *AvroConsumer) DecodeAvroMessage(msg pulsar.Message) (interface{}, error)
  • func (ac *AvroConsumer) GetSchemaByTopic(topic string) (*goavro.Codec, error)
• type AvroProducer
  • func NewAvroProducer(client pulsar.Client, producerOptions pulsar.ProducerOptions, ...) (*AvroProducer, error)
  • func (ap *AvroProducer) Close()
  • func (ap *AvroProducer) EncodeAvroMessage(value interface{}) ([]byte, error)
  • func (ap *AvroProducer) Send(message *pulsar.ProducerMessage) (pulsar.MessageID, error)
• type CachedSchemaRegistryClient
  • func NewCachedSchemaRegistryClient(connect []string) *CachedSchemaRegistryClient
  • func NewCachedSchemaRegistryClientWithRetries(connect []string, retries int) *CachedSchemaRegistryClient
  • func (client *CachedSchemaRegistryClient) CreateSchemaByTopic(topic string, schema string) error
  • func (client *CachedSchemaRegistryClient) GetSchemaCodecByTopic(topic string) (*goavro.Codec, error)
• type SchemaRegistryClient
  • func NewSchemaRegistryClient(connect []string) *SchemaRegistryClient
  • func NewSchemaRegistryClientWithRetries(connect []string, retries int) *SchemaRegistryClient
  • func (client *SchemaRegistryClient) CreateSchemaByTopic(topic string, schema string) error
  • func (client *SchemaRegistryClient) GetSchemaCodecByTopic(topic string) (*goavro.Codec, error)
• type SchemaRegistryClientInterface

Functions

func DecodeAvroFromBytes

func DecodeAvroFromBytes(schema string, payload []byte) (interface{}, error)

DecodeAvroFromBytes convert bytes to interface{} datum

func EncodeAvroToBytes

func EncodeAvroToBytes(schema string, datum interface{}) ([]byte, error)

EncodeAvroToBytes convert interface{} datum to bytes

Types

type AvroConsumer

type AvroConsumer struct {
	Consumer pulsar.Consumer
	// contains filtered or unexported fields
}

AvroConsumer is a basic consumer to interact with schema registry, avro and pulsar

func NewAvroConsumer

func NewAvroConsumer(client pulsar.Client, consumerOptions pulsar.ConsumerOptions, schemaRegistryUrls []string) (*AvroConsumer, error)

func (*AvroConsumer) Close

func (ac *AvroConsumer) Close()

func (*AvroConsumer) DecodeAvroMessage

func (ac *AvroConsumer) DecodeAvroMessage(msg pulsar.Message) (interface{}, error)

func (*AvroConsumer) GetSchemaByTopic

func (ac *AvroConsumer) GetSchemaByTopic(topic string) (*goavro.Codec, error)

type AvroProducer

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

func NewAvroProducer

func NewAvroProducer(client pulsar.Client, producerOptions pulsar.ProducerOptions, schemaRegistryUrls []string, schema string) (*AvroProducer, error)

NewAvroProducer is a basic producer to interact with schema registry, avro and pulsar

func (*AvroProducer) Close

func (ap *AvroProducer) Close()

func (*AvroProducer) EncodeAvroMessage

func (ap *AvroProducer) EncodeAvroMessage(value interface{}) ([]byte, error)

func (*AvroProducer) Send

func (ap *AvroProducer) Send(message *pulsar.ProducerMessage) (pulsar.MessageID, error)

type CachedSchemaRegistryClient

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

CachedSchemaRegistryClient is a schema registry client that will cache some data to improve performance

func NewCachedSchemaRegistryClient

func NewCachedSchemaRegistryClient(connect []string) *CachedSchemaRegistryClient

func NewCachedSchemaRegistryClientWithRetries

func NewCachedSchemaRegistryClientWithRetries(connect []string, retries int) *CachedSchemaRegistryClient

func (*CachedSchemaRegistryClient) CreateSchemaByTopic

func (client *CachedSchemaRegistryClient) CreateSchemaByTopic(topic string, schema string) error

CreateSchemaByTopic will create a schema for the specified topic

func (*CachedSchemaRegistryClient) GetSchemaCodecByTopic

func (client *CachedSchemaRegistryClient) GetSchemaCodecByTopic(topic string) (*goavro.Codec, error)

GetSchemaCodecByTopic will return and cache the codec with the given topic information

type SchemaRegistryClient

type SchemaRegistryClient struct {
	SchemaRegistryConnect []string
	// contains filtered or unexported fields
}

SchemaRegistryClient is a basic http client to interact with schema registry

func NewSchemaRegistryClient

func NewSchemaRegistryClient(connect []string) *SchemaRegistryClient

NewSchemaRegistryClient creates a client to talk with the schema registry at the connect string By default it will retry failed requests (5XX responses and http errors) len(connect) number of times

func NewSchemaRegistryClientWithRetries

func NewSchemaRegistryClientWithRetries(connect []string, retries int) *SchemaRegistryClient

NewSchemaRegistryClientWithRetries creates an http client with a configurable amount of retries on 5XX responses

func (*SchemaRegistryClient) CreateSchemaByTopic

func (client *SchemaRegistryClient) CreateSchemaByTopic(topic string, schema string) error

CreateSchemaByTopic creates a schema for the specified topic

func (*SchemaRegistryClient) GetSchemaCodecByTopic

func (client *SchemaRegistryClient) GetSchemaCodecByTopic(topic string) (*goavro.Codec, error)

GetSchemaCodecByTopic returns a goavro.Codec by topic

type SchemaRegistryClientInterface

type SchemaRegistryClientInterface interface {
	GetSchemaCodecByTopic(string) (*goavro.Codec, error)
	CreateSchemaByTopic(string, string) error
}

SchemaRegistryClientInterface defines the api for all clients interfacing with schema registry