create_partitions_request.go 2.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121
  1. package sarama
  2. import "time"
  3. type CreatePartitionsRequest struct {
  4. TopicPartitions map[string]*TopicPartition
  5. Timeout time.Duration
  6. ValidateOnly bool
  7. }
  8. func (c *CreatePartitionsRequest) encode(pe packetEncoder) error {
  9. if err := pe.putArrayLength(len(c.TopicPartitions)); err != nil {
  10. return err
  11. }
  12. for topic, partition := range c.TopicPartitions {
  13. if err := pe.putString(topic); err != nil {
  14. return err
  15. }
  16. if err := partition.encode(pe); err != nil {
  17. return err
  18. }
  19. }
  20. pe.putInt32(int32(c.Timeout / time.Millisecond))
  21. pe.putBool(c.ValidateOnly)
  22. return nil
  23. }
  24. func (c *CreatePartitionsRequest) decode(pd packetDecoder, version int16) (err error) {
  25. n, err := pd.getArrayLength()
  26. if err != nil {
  27. return err
  28. }
  29. c.TopicPartitions = make(map[string]*TopicPartition, n)
  30. for i := 0; i < n; i++ {
  31. topic, err := pd.getString()
  32. if err != nil {
  33. return err
  34. }
  35. c.TopicPartitions[topic] = new(TopicPartition)
  36. if err := c.TopicPartitions[topic].decode(pd, version); err != nil {
  37. return err
  38. }
  39. }
  40. timeout, err := pd.getInt32()
  41. if err != nil {
  42. return err
  43. }
  44. c.Timeout = time.Duration(timeout) * time.Millisecond
  45. if c.ValidateOnly, err = pd.getBool(); err != nil {
  46. return err
  47. }
  48. return nil
  49. }
  50. func (r *CreatePartitionsRequest) key() int16 {
  51. return 37
  52. }
  53. func (r *CreatePartitionsRequest) version() int16 {
  54. return 0
  55. }
  56. func (r *CreatePartitionsRequest) requiredVersion() KafkaVersion {
  57. return V1_0_0_0
  58. }
  59. type TopicPartition struct {
  60. Count int32
  61. Assignment [][]int32
  62. }
  63. func (t *TopicPartition) encode(pe packetEncoder) error {
  64. pe.putInt32(t.Count)
  65. if len(t.Assignment) == 0 {
  66. pe.putInt32(-1)
  67. return nil
  68. }
  69. if err := pe.putArrayLength(len(t.Assignment)); err != nil {
  70. return err
  71. }
  72. for _, assign := range t.Assignment {
  73. if err := pe.putInt32Array(assign); err != nil {
  74. return err
  75. }
  76. }
  77. return nil
  78. }
  79. func (t *TopicPartition) decode(pd packetDecoder, version int16) (err error) {
  80. if t.Count, err = pd.getInt32(); err != nil {
  81. return err
  82. }
  83. n, err := pd.getInt32()
  84. if err != nil {
  85. return err
  86. }
  87. if n <= 0 {
  88. return nil
  89. }
  90. t.Assignment = make([][]int32, n)
  91. for i := 0; i < int(n); i++ {
  92. if t.Assignment[i], err = pd.getInt32Array(); err != nil {
  93. return err
  94. }
  95. }
  96. return nil
  97. }