metadata_request.go 1.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081
  1. package sarama
  2. type MetadataRequest struct {
  3. Version int16
  4. Topics []string
  5. AllowAutoTopicCreation bool
  6. }
  7. func (r *MetadataRequest) encode(pe packetEncoder) error {
  8. if r.Version < 0 || r.Version > 5 {
  9. return PacketEncodingError{"invalid or unsupported MetadataRequest version field"}
  10. }
  11. if r.Version == 0 || len(r.Topics) > 0 {
  12. err := pe.putArrayLength(len(r.Topics))
  13. if err != nil {
  14. return err
  15. }
  16. for i := range r.Topics {
  17. err = pe.putString(r.Topics[i])
  18. if err != nil {
  19. return err
  20. }
  21. }
  22. } else {
  23. pe.putInt32(-1)
  24. }
  25. if r.Version > 3 {
  26. pe.putBool(r.AllowAutoTopicCreation)
  27. }
  28. return nil
  29. }
  30. func (r *MetadataRequest) decode(pd packetDecoder, version int16) error {
  31. r.Version = version
  32. size, err := pd.getInt32()
  33. if err != nil {
  34. return err
  35. }
  36. if size > 0 {
  37. r.Topics = make([]string, size)
  38. for i := range r.Topics {
  39. topic, err := pd.getString()
  40. if err != nil {
  41. return err
  42. }
  43. r.Topics[i] = topic
  44. }
  45. }
  46. if r.Version > 3 {
  47. autoCreation, err := pd.getBool()
  48. if err != nil {
  49. return err
  50. }
  51. r.AllowAutoTopicCreation = autoCreation
  52. }
  53. return nil
  54. }
  55. func (r *MetadataRequest) key() int16 {
  56. return 3
  57. }
  58. func (r *MetadataRequest) version() int16 {
  59. return r.Version
  60. }
  61. func (r *MetadataRequest) requiredVersion() KafkaVersion {
  62. switch r.Version {
  63. case 1:
  64. return V0_10_0_0
  65. case 2:
  66. return V0_10_1_0
  67. case 3, 4:
  68. return V0_11_0_0
  69. case 5:
  70. return V1_0_0_0
  71. default:
  72. return MinVersion
  73. }
  74. }