sarama/protocol/produce_request.go

package protocol

type ProduceRequest struct {
	RequiredAcks int16
	Timeout      int32
	msgSets      map[string]map[int32]*MessageSet
}

func (p *ProduceRequest) encode(pe packetEncoder) {
	pe.putInt16(p.RequiredAcks)
	pe.putInt32(p.Timeout)
	pe.putArrayCount(len(p.msgSets))
	for topic, partitions := range p.msgSets {
		pe.putString(topic)
		pe.putArrayCount(len(partitions))
		for id, msgSet := range partitions {
			pe.putInt32(id)
			pe.pushLength32()
			msgSet.encode(pe)
			pe.pop()
		}
	}
}

func (p *ProduceRequest) key() int16 {
	return 0
}

func (p *ProduceRequest) version() int16 {
	return 0
}

func (p *ProduceRequest) AddMessage(topic string, partition int32, msg *Message) {
	if p.msgSets == nil {
		p.msgSets = make(map[string]map[int32]*MessageSet)
	}

	if p.msgSets[topic] == nil {
		p.msgSets[topic] = make(map[int32]*MessageSet)
	}

	set := p.msgSets[topic][partition]

	if set == nil {
		set = new(MessageSet)
		p.msgSets[topic][partition] = set
	}

	set.addMessage(msg)
}