producer

Documentation

Index

• Constants
• func MurmurHasher() hash.Hash32
• func NewJVMCompatiblePartitioner(topic string) sarama.Partitioner
• type AvroMessageSerializer
  • func CreateAvroMessageSerializer(topic string, avroSchemaRegistry string) (AvroMessageSerializer, error)
  • func (serializer AvroMessageSerializer) CanSerialize(topic string) (bool, error)
  • func (serializer AvroMessageSerializer) Serialize(key, value []byte, flags ProducerFlags) (*sarama.ProducerMessage, error)
• type DefaultMessageSerializer
  • func (serializer DefaultMessageSerializer) CanSerialize(_ string) (bool, error)
  • func (serializer DefaultMessageSerializer) Serialize(key, value []byte, flags ProducerFlags) (*sarama.ProducerMessage, error)
• type MessageSerializer
• type ProducerFlags
• type ProducerOperation
  • func (operation *ProducerOperation) Produce(topic string, flags ProducerFlags) error

Constants

const (
	HEX    = "hex"
	BASE64 = "base64"
)

Functions

func MurmurHasher

func MurmurHasher() hash.Hash32

MurmurHasher creates murmur2 hasher implementing hash.Hash32 interface. The implementation is not full and does not support streaming. It only implements the interface to comply with sarama.NewCustomHashPartitioner signature. But Sarama only uses Write method once, when writing keys and values of the message, so streaming support is not necessary.

func NewJVMCompatiblePartitioner

func NewJVMCompatiblePartitioner(topic string) sarama.Partitioner

NewJVMCompatiblePartitioner creates a Sarama partitioner that uses the same hashing algorithm as JVM Kafka clients.

Types

type AvroMessageSerializer

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

func CreateAvroMessageSerializer

func CreateAvroMessageSerializer(topic string, avroSchemaRegistry string) (AvroMessageSerializer, error)

func (AvroMessageSerializer) CanSerialize

func (serializer AvroMessageSerializer) CanSerialize(topic string) (bool, error)

func (AvroMessageSerializer) Serialize

func (serializer AvroMessageSerializer) Serialize(key, value []byte, flags ProducerFlags) (*sarama.ProducerMessage, error)

type DefaultMessageSerializer

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

func (DefaultMessageSerializer) CanSerialize

func (serializer DefaultMessageSerializer) CanSerialize(_ string) (bool, error)

func (DefaultMessageSerializer) Serialize

func (serializer DefaultMessageSerializer) Serialize(key, value []byte, flags ProducerFlags) (*sarama.ProducerMessage, error)

type MessageSerializer

type MessageSerializer interface {
	CanSerialize(topic string) (bool, error)
	Serialize(key, value []byte, flags ProducerFlags) (*sarama.ProducerMessage, error)
}

type ProducerFlags

type ProducerFlags struct {
	Partitioner        string
	RequiredAcks       string
	Partition          int32
	Separator          string
	LineSeparator      string
	File               string
	Key                string
	Value              string
	NullValue          bool
	Headers            []string
	KeySchemaVersion   int
	ValueSchemaVersion int
	KeyEncoding        string
	ValueEncoding      string
	Silent             bool
	RateInSeconds      int
}

type ProducerOperation

type ProducerOperation struct {
}

func (*ProducerOperation) Produce

func (operation *ProducerOperation) Produce(topic string, flags ProducerFlags) error