offset_fetch_request.go 673 B

12345678910111213141516171819202122232425262728293031
  1. package protocol
  2. type OffsetFetchRequest struct {
  3. ConsumerGroup string
  4. partitions map[string][]int32
  5. }
  6. func (r *OffsetFetchRequest) encode(pe packetEncoder) {
  7. pe.putString(r.ConsumerGroup)
  8. pe.putArrayCount(len(r.partitions))
  9. for topic, partitions := range r.partitions {
  10. pe.putString(topic)
  11. pe.putInt32Array(partitions)
  12. }
  13. }
  14. func (r *OffsetFetchRequest) key() int16 {
  15. return 7
  16. }
  17. func (r *OffsetFetchRequest) version() int16 {
  18. return 0
  19. }
  20. func (r *OffsetFetchRequest) AddPartition(topic string, partition_id int32) {
  21. if r.partitions == nil {
  22. r.partitions = make(map[string][]int32)
  23. }
  24. r.partitions[topic] = append(r.partitions[topic], partition_id)
  25. }