offset_fetch_response.go 1.5 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788
  1. package sarama
  2. type OffsetFetchResponseBlock struct {
  3. Offset int64
  4. Metadata string
  5. Err KError
  6. }
  7. func (r *OffsetFetchResponseBlock) decode(pd packetDecoder) (err error) {
  8. r.Offset, err = pd.getInt64()
  9. if err != nil {
  10. return err
  11. }
  12. r.Metadata, err = pd.getString()
  13. if err != nil {
  14. return err
  15. }
  16. tmp, err := pd.getInt16()
  17. if err != nil {
  18. return err
  19. }
  20. r.Err = KError(tmp)
  21. return nil
  22. }
  23. func (r *OffsetFetchResponseBlock) encode(pe packetEncoder) (err error) {
  24. pe.putInt64(r.Offset)
  25. err = pe.putString(r.Metadata)
  26. if err != nil {
  27. return err
  28. }
  29. pe.putInt16(int16(r.Err))
  30. return nil
  31. }
  32. type OffsetFetchResponse struct {
  33. ClientID string
  34. Blocks map[string]map[int32]*OffsetFetchResponseBlock
  35. }
  36. func (r *OffsetFetchResponse) decode(pd packetDecoder) (err error) {
  37. r.ClientID, err = pd.getString()
  38. if err != nil {
  39. return err
  40. }
  41. numTopics, err := pd.getArrayLength()
  42. if err != nil {
  43. return err
  44. }
  45. r.Blocks = make(map[string]map[int32]*OffsetFetchResponseBlock, numTopics)
  46. for i := 0; i < numTopics; i++ {
  47. name, err := pd.getString()
  48. if err != nil {
  49. return err
  50. }
  51. numBlocks, err := pd.getArrayLength()
  52. if err != nil {
  53. return err
  54. }
  55. r.Blocks[name] = make(map[int32]*OffsetFetchResponseBlock, numBlocks)
  56. for j := 0; j < numBlocks; j++ {
  57. id, err := pd.getInt32()
  58. if err != nil {
  59. return err
  60. }
  61. block := new(OffsetFetchResponseBlock)
  62. err = block.decode(pd)
  63. if err != nil {
  64. return err
  65. }
  66. r.Blocks[name][id] = block
  67. }
  68. }
  69. return nil
  70. }