fixtures

Documentation

Index

• Constants
• func NewRecord(ts time.Time) (*models.Record, int32, int32)
• type AsyncProducer
  • 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
  • func NewFixtureRecord() *FixtureRecord
  • func (r *FixtureRecord) Schema() string
  • func (r *FixtureRecord) ToAvroSerialization() ([]byte, error)
  • func (r *FixtureRecord) Topic() string
• type FixtureSchemaRegistry
  • func (this *FixtureSchemaRegistry) RegisterOrGetSchemaId(event KafkaEvent) (int, error)
• type KafkaEvent
• type Producer
  • func NewProducer(kafkaAddr string, config *sarama.Config, ...) (Producer, error)

Constants

const DefaultTopic = "my-topic"

Functions

func NewRecord

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

Types

type AsyncProducer

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

func (*AsyncProducer) Close

func (this *AsyncProducer) Close() error

func (*AsyncProducer) GetHeader

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

func (*AsyncProducer) GetSuccesses

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

func (*AsyncProducer) Publish

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

func (*AsyncProducer) Start

func (this *AsyncProducer) Start()

type FixtureRecord

type FixtureRecord struct {
	Id        int32
	Value     int32
	Timestamp int64 `json:"@timestamp"`
}

func NewFixtureRecord

func NewFixtureRecord() *FixtureRecord

func (*FixtureRecord) Schema

func (r *FixtureRecord) Schema() string

func (*FixtureRecord) ToAvroSerialization

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

func (*FixtureRecord) Topic

func (r *FixtureRecord) Topic() string

type FixtureSchemaRegistry

type FixtureSchemaRegistry struct {
	*schema_registry.SchemaRegistry
}

func (*FixtureSchemaRegistry) RegisterOrGetSchemaId

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

type KafkaEvent

type KafkaEvent interface {
	Topic() string
	Schema() string
	ToAvroSerialization() ([]byte, error) // should return the avro serialized object
}

type Producer

type Producer interface {
	Publish(KafkaEvent) error
	Start()
	Close() error
	GetSuccesses() <-chan *sarama.ProducerMessage
}

func NewProducer

func NewProducer(kafkaAddr string, config *sarama.Config, schemaRegistry *schema_registry.SchemaRegistry) (Producer, error)

Config example :

config.Producer.Return.Successes = true
config.Producer.MaxMessageBytes = 20 * 1024 * 1024 // 20mb
config.Producer.Flush.Frequency = 500 * time.Millisecond
config.Version = sarama.V0_10_0_0                  // This version is the same as in production