kafkaavro

Dec 2, 2019

Changes in this version

+ type ProducerConfig struct

+ BackOffConfig backoff.BackOff

+ KeySchema string

+ Producer kafkaProducer

+ SchemaRegistryClient SchemaRegistryClient

+ TopicName string

+ ValueSchema string

Oct 17, 2019

Changes in this version

type ConsumerMessage

+ Textual []byte

Apr 1, 2019

Changes in this version

type Consumer

+ func (ac *Consumer) SubscribeTopics(topics []string, rebalanceCb kafka.RebalanceCb) error

Apr 1, 2019

Changes in this version

type Consumer

+ func (ac *Consumer) EnsureTopics() error

Dec 4, 2018

Changes in this version

+ func DecodeRecordFromNative(src interface{}, dst interface{}) error

+ type NativeDecoder interface

+ FromNative func(interface{}) error

+ type OptionalDay struct

+ Time time.Time

+ Valid bool

+ func NewOptionalDay(t time.Time, valid bool) OptionalDay

+ func (od *OptionalDay) FromNative(data interface{}) error

+ func (od OptionalDay) MarshalJSON() ([]byte, error)

+ type OptionalInt struct

+ func NewOptionalInt(i int64, valid bool) OptionalInt

+ func (i *OptionalInt) FromNative(data interface{}) error

+ type OptionalString struct

+ func NewOptionalString(s string, valid bool) OptionalString

+ func (s *OptionalString) FromNative(data interface{}) error

Nov 10, 2018

Changes in this version

+ type CachedSchemaRegistryClient struct

+ SchemaRegistryClient *schemaregistry.Client

+ func NewCachedSchemaRegistryClient(baseURL string, options ...schemaregistry.Option) (*CachedSchemaRegistryClient, error)

+ func (cached *CachedSchemaRegistryClient) DeleteSubject(subject string) (versions []int, err error)

+ func (cached *CachedSchemaRegistryClient) GetLatestSchema(subject string) (*goavro.Codec, error)

+ func (cached *CachedSchemaRegistryClient) GetSchemaByID(id int) (*goavro.Codec, error)

+ func (cached *CachedSchemaRegistryClient) GetSchemaBySubject(subject string, version int) (*goavro.Codec, error)

+ func (cached *CachedSchemaRegistryClient) IsSchemaRegistered(subject string, codec *goavro.Codec) (bool, schemaregistry.Schema, error)

+ func (cached *CachedSchemaRegistryClient) RegisterNewSchema(subject string, codec *goavro.Codec) (int, error)

+ func (cached *CachedSchemaRegistryClient) Subjects() ([]string, error)

+ func (cached *CachedSchemaRegistryClient) Versions(subject string) ([]int, error)

+ type Consumer struct

+ func NewConsumer(topics []string, consumer *kafka.Consumer, ...) (*Consumer, error)

+ func (ac *Consumer) Close()

+ func (ac *Consumer) CommitMessage(msg ConsumerMessage) ([]kafka.TopicPartition, error)

+ func (ac *Consumer) Messages(stopChan chan struct{}) (chan ConsumerMessage, chan kafka.Event)

+ type ConsumerMessage struct

+ Error error

+ Parsed interface{}

+ type Producer struct

+ func NewProducer(topic string, keySchema string, valueSchema string, producer *kafka.Producer, ...) (*Producer, error)

+ func (ac *Producer) Close()

+ func (ap *Producer) Produce(key interface{}, value interface{}, deliveryChan chan kafka.Event) error

+ type SchemaRegistryClient interface

+ GetSchemaByID func(id int) (*goavro.Codec, error)

+ RegisterNewSchema func(subject string, codec *goavro.Codec) (int, error)