fixtures

Jun 11, 2020

Changes in this version

type FixtureRecord

+ func NewEmptyFixtureRecord() *FixtureRecord

Apr 11, 2018

Changes in this version

+ func NewElasticRecord() (*models.ElasticRecord, int32)

Mar 19, 2018

Changes in this version

+ const DefaultTopic

+ func NewRecord(ts time.Time) (*models.Record, int32, int32)

+ type AsyncProducer struct

+ func (this *AsyncProducer) Close() error

+ func (this *AsyncProducer) GetHeader(event KafkaEvent) ([]byte, error)

+ func (this *AsyncProducer) GetSuccesses() <-chan *sarama.ProducerMessage

+ func (this *AsyncProducer) Publish(event KafkaEvent) error

+ func (this *AsyncProducer) Start()

+ type FixtureRecord struct

+ Id int32

+ Timestamp int64

+ Value int32

+ func NewFixtureRecord() *FixtureRecord

+ func (r *FixtureRecord) Schema() string

+ func (r *FixtureRecord) ToAvroSerialization() ([]byte, error)

+ func (r *FixtureRecord) Topic() string

+ type FixtureSchemaRegistry struct

+ func (this *FixtureSchemaRegistry) RegisterOrGetSchemaId(event KafkaEvent) (int, error)

+ type KafkaEvent interface

+ Schema func() string

+ ToAvroSerialization func() ([]byte, error)

+ Topic func() string

+ type Producer interface

+ Close func() error

+ GetSuccesses func() <-chan *sarama.ProducerMessage

+ Publish func(KafkaEvent) error

+ Start func()

+ func NewProducer(kafkaAddr string, config *sarama.Config, ...) (Producer, error)