create_topics_response.go 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112
  1. package sarama
  2. import "time"
  3. type CreateTopicsResponse struct {
  4. Version int16
  5. ThrottleTime time.Duration
  6. TopicErrors map[string]*TopicError
  7. }
  8. func (c *CreateTopicsResponse) encode(pe packetEncoder) error {
  9. if c.Version >= 2 {
  10. pe.putInt32(int32(c.ThrottleTime / time.Millisecond))
  11. }
  12. if err := pe.putArrayLength(len(c.TopicErrors)); err != nil {
  13. return err
  14. }
  15. for topic, topicError := range c.TopicErrors {
  16. if err := pe.putString(topic); err != nil {
  17. return err
  18. }
  19. if err := topicError.encode(pe, c.Version); err != nil {
  20. return err
  21. }
  22. }
  23. return nil
  24. }
  25. func (c *CreateTopicsResponse) decode(pd packetDecoder, version int16) (err error) {
  26. c.Version = version
  27. if version >= 2 {
  28. throttleTime, err := pd.getInt32()
  29. if err != nil {
  30. return err
  31. }
  32. c.ThrottleTime = time.Duration(throttleTime) * time.Millisecond
  33. }
  34. n, err := pd.getArrayLength()
  35. if err != nil {
  36. return err
  37. }
  38. c.TopicErrors = make(map[string]*TopicError, n)
  39. for i := 0; i < n; i++ {
  40. topic, err := pd.getString()
  41. if err != nil {
  42. return err
  43. }
  44. c.TopicErrors[topic] = new(TopicError)
  45. if err := c.TopicErrors[topic].decode(pd, version); err != nil {
  46. return err
  47. }
  48. }
  49. return nil
  50. }
  51. func (c *CreateTopicsResponse) key() int16 {
  52. return 19
  53. }
  54. func (c *CreateTopicsResponse) version() int16 {
  55. return c.Version
  56. }
  57. func (c *CreateTopicsResponse) requiredVersion() KafkaVersion {
  58. switch c.Version {
  59. case 2:
  60. return V1_0_0_0
  61. case 1:
  62. return V0_11_0_0
  63. default:
  64. return V0_10_1_0
  65. }
  66. }
  67. type TopicError struct {
  68. Err KError
  69. ErrMsg *string
  70. }
  71. func (t *TopicError) encode(pe packetEncoder, version int16) error {
  72. pe.putInt16(int16(t.Err))
  73. if version >= 1 {
  74. if err := pe.putNullableString(t.ErrMsg); err != nil {
  75. return err
  76. }
  77. }
  78. return nil
  79. }
  80. func (t *TopicError) decode(pd packetDecoder, version int16) (err error) {
  81. kErr, err := pd.getInt16()
  82. if err != nil {
  83. return err
  84. }
  85. t.Err = KError(kErr)
  86. if version >= 1 {
  87. if t.ErrMsg, err = pd.getNullableString(); err != nil {
  88. return err
  89. }
  90. }
  91. return nil
  92. }