sync_group_request.go 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100
  1. package sarama
  2. type SyncGroupRequest struct {
  3. GroupId string
  4. GenerationId int32
  5. MemberId string
  6. GroupAssignments map[string][]byte
  7. }
  8. func (r *SyncGroupRequest) encode(pe packetEncoder) error {
  9. if err := pe.putString(r.GroupId); err != nil {
  10. return err
  11. }
  12. pe.putInt32(r.GenerationId)
  13. if err := pe.putString(r.MemberId); err != nil {
  14. return err
  15. }
  16. if err := pe.putArrayLength(len(r.GroupAssignments)); err != nil {
  17. return err
  18. }
  19. for memberId, memberAssignment := range r.GroupAssignments {
  20. if err := pe.putString(memberId); err != nil {
  21. return err
  22. }
  23. if err := pe.putBytes(memberAssignment); err != nil {
  24. return err
  25. }
  26. }
  27. return nil
  28. }
  29. func (r *SyncGroupRequest) decode(pd packetDecoder, version int16) (err error) {
  30. if r.GroupId, err = pd.getString(); err != nil {
  31. return
  32. }
  33. if r.GenerationId, err = pd.getInt32(); err != nil {
  34. return
  35. }
  36. if r.MemberId, err = pd.getString(); err != nil {
  37. return
  38. }
  39. n, err := pd.getArrayLength()
  40. if err != nil {
  41. return err
  42. }
  43. if n == 0 {
  44. return nil
  45. }
  46. r.GroupAssignments = make(map[string][]byte)
  47. for i := 0; i < n; i++ {
  48. memberId, err := pd.getString()
  49. if err != nil {
  50. return err
  51. }
  52. memberAssignment, err := pd.getBytes()
  53. if err != nil {
  54. return err
  55. }
  56. r.GroupAssignments[memberId] = memberAssignment
  57. }
  58. return nil
  59. }
  60. func (r *SyncGroupRequest) key() int16 {
  61. return 14
  62. }
  63. func (r *SyncGroupRequest) version() int16 {
  64. return 0
  65. }
  66. func (r *SyncGroupRequest) requiredVersion() KafkaVersion {
  67. return V0_9_0_0
  68. }
  69. func (r *SyncGroupRequest) AddGroupAssignment(memberId string, memberAssignment []byte) {
  70. if r.GroupAssignments == nil {
  71. r.GroupAssignments = make(map[string][]byte)
  72. }
  73. r.GroupAssignments[memberId] = memberAssignment
  74. }
  75. func (r *SyncGroupRequest) AddGroupAssignmentMember(memberId string, memberAssignment *ConsumerGroupMemberAssignment) error {
  76. bin, err := encode(memberAssignment, nil)
  77. if err != nil {
  78. return err
  79. }
  80. r.AddGroupAssignment(memberId, bin)
  81. return nil
  82. }