producer

Documentation

Index

• Constants
• func MurmurHasher() hash.Hash32
• func NewJVMCompatiblePartitioner(topic string) sarama.Partitioner
• type AvroMessageSerializer
  • func (serializer AvroMessageSerializer) CanSerialize(topic string) (bool, error)
  • func (serializer AvroMessageSerializer) Serialize(key, value []byte, flags Flags) (*sarama.ProducerMessage, error)
• type DefaultMessageSerializer
  • func (serializer DefaultMessageSerializer) CanSerialize(_ string) (bool, error)
  • func (serializer DefaultMessageSerializer) Serialize(key, value []byte, flags Flags) (*sarama.ProducerMessage, error)
• type Flags
• type MessageSerializer
• type MessageSerializerChain
  • func (serializer MessageSerializerChain) CanSerialize(topic string) (bool, error)
  • func (serializer MessageSerializerChain) Serialize(key, value []byte, flags Flags) (*sarama.ProducerMessage, error)
• type Operation
  • func (operation *Operation) Produce(topic string, flags Flags) error
• type ProtobufMessageSerializer
  • func CreateProtobufMessageSerializer(topic string, context protobuf.SearchContext, keyType, valueType string) (*ProtobufMessageSerializer, error)
  • func (serializer ProtobufMessageSerializer) CanSerialize(string) (bool, error)
  • func (serializer ProtobufMessageSerializer) Serialize(key, value []byte, flags Flags) (*sarama.ProducerMessage, error)

Constants

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

const DefaultMaxMessagesBytes = 1000000

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 (AvroMessageSerializer) CanSerialize

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

func (AvroMessageSerializer) Serialize

func (serializer AvroMessageSerializer) Serialize(key, value []byte, flags Flags) (*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 Flags) (*sarama.ProducerMessage, error)

type Flags

type Flags struct {
	Partitioner        string
	RequiredAcks       string
	MaxMessageBytes    int
	Partition          int32
	Separator          string
	LineSeparator      string
	File               string
	InputFormat        string
	Key                string
	Value              string
	NullValue          bool
	Headers            []string
	KeySchemaVersion   int
	ValueSchemaVersion int
	KeyEncoding        string
	ValueEncoding      string
	Silent             bool
	RateInSeconds      int
	ProtoFiles         []string
	ProtoImportPaths   []string
	ProtosetFiles      []string
	KeyProtoType       string
	ValueProtoType     string
}

type MessageSerializer

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

type MessageSerializerChain

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

func (MessageSerializerChain) CanSerialize

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

func (MessageSerializerChain) Serialize

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

type Operation

type Operation struct {
}

func (*Operation) Produce

func (operation *Operation) Produce(topic string, flags Flags) error

type ProtobufMessageSerializer

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

func CreateProtobufMessageSerializer

func CreateProtobufMessageSerializer(topic string, context protobuf.SearchContext, keyType, valueType string) (*ProtobufMessageSerializer, error)

func (ProtobufMessageSerializer) CanSerialize

func (serializer ProtobufMessageSerializer) CanSerialize(string) (bool, error)

func (ProtobufMessageSerializer) Serialize

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