offset_fetch_request.go 854 B

12345678910111213141516171819202122232425262728293031323334353637383940
  1. package sarama
  2. type OffsetFetchRequest struct {
  3. ConsumerGroup string
  4. partitions map[string][]int32
  5. }
  6. func (r *OffsetFetchRequest) encode(pe packetEncoder) (err error) {
  7. if err = pe.putString(r.ConsumerGroup); err != nil {
  8. return err
  9. }
  10. if err = pe.putArrayLength(len(r.partitions)); err != nil {
  11. return err
  12. }
  13. for topic, partitions := range r.partitions {
  14. if err = pe.putString(topic); err != nil {
  15. return err
  16. }
  17. if err = pe.putInt32Array(partitions); err != nil {
  18. return err
  19. }
  20. }
  21. return nil
  22. }
  23. func (r *OffsetFetchRequest) key() int16 {
  24. return 9
  25. }
  26. func (r *OffsetFetchRequest) version() int16 {
  27. return 0
  28. }
  29. func (r *OffsetFetchRequest) AddPartition(topic string, partitionID int32) {
  30. if r.partitions == nil {
  31. r.partitions = make(map[string][]int32)
  32. }
  33. r.partitions[topic] = append(r.partitions[topic], partitionID)
  34. }