create_partitions_response.go 2.1 KB

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