leave_group_response.go 2.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109
  1. package sarama
  2. import "time"
  3. type MemberResponse struct {
  4. MemberId string
  5. GroupInstanceId *string
  6. Err KError
  7. }
  8. type LeaveGroupResponse struct {
  9. Version int16
  10. ThrottleTime int32
  11. Err KError
  12. Members []MemberResponse
  13. }
  14. func (r *LeaveGroupResponse) encode(pe packetEncoder) error {
  15. if r.Version >= 1 {
  16. pe.putInt32(r.ThrottleTime)
  17. }
  18. pe.putInt16(int16(r.Err))
  19. if r.Version >= 3 {
  20. if err := pe.putArrayLength(len(r.Members)); err != nil {
  21. return err
  22. }
  23. for _, member := range r.Members {
  24. if err := pe.putString(member.MemberId); err != nil {
  25. return err
  26. }
  27. if err := pe.putNullableString(member.GroupInstanceId); err != nil {
  28. return err
  29. }
  30. pe.putInt16(int16(member.Err))
  31. }
  32. }
  33. return nil
  34. }
  35. func (r *LeaveGroupResponse) decode(pd packetDecoder, version int16) (err error) {
  36. r.Version = version
  37. if r.Version >= 1 {
  38. if r.ThrottleTime, err = pd.getInt32(); err != nil {
  39. return err
  40. }
  41. }
  42. kerr, err := pd.getInt16()
  43. if err != nil {
  44. return err
  45. }
  46. r.Err = KError(kerr)
  47. if r.Version >= 3 {
  48. membersLen, err := pd.getArrayLength()
  49. if err != nil {
  50. return err
  51. }
  52. r.Members = make([]MemberResponse, membersLen)
  53. for i := 0; i < len(r.Members); i++ {
  54. if r.Members[i].MemberId, err = pd.getString(); err != nil {
  55. return err
  56. }
  57. if r.Members[i].GroupInstanceId, err = pd.getNullableString(); err != nil {
  58. return err
  59. }
  60. if memberErr, err := pd.getInt16(); err != nil {
  61. return err
  62. } else {
  63. r.Members[i].Err = KError(memberErr)
  64. }
  65. }
  66. }
  67. return nil
  68. }
  69. func (r *LeaveGroupResponse) key() int16 {
  70. return 13
  71. }
  72. func (r *LeaveGroupResponse) version() int16 {
  73. return r.Version
  74. }
  75. func (r *LeaveGroupResponse) headerVersion() int16 {
  76. return 0
  77. }
  78. func (r *LeaveGroupResponse) isValidVersion() bool {
  79. return r.Version >= 0 && r.Version <= 3
  80. }
  81. func (r *LeaveGroupResponse) requiredVersion() KafkaVersion {
  82. switch r.Version {
  83. case 3:
  84. return V2_4_0_0
  85. case 2:
  86. return V2_0_0_0
  87. case 1:
  88. return V0_11_0_0
  89. case 0:
  90. return V0_9_0_0
  91. default:
  92. return V2_4_0_0
  93. }
  94. }
  95. func (r *LeaveGroupResponse) throttleTime() time.Duration {
  96. return time.Duration(r.ThrottleTime) * time.Millisecond
  97. }