offset_fetch_response.go 1.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071
  1. package protocol
  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. r.Err, err = pd.getError()
  17. return err
  18. }
  19. type OffsetFetchResponse struct {
  20. ClientID string
  21. Blocks map[string]map[int32]*OffsetFetchResponseBlock
  22. }
  23. func (r *OffsetFetchResponse) decode(pd packetDecoder) (err error) {
  24. r.ClientID, err = pd.getString()
  25. if err != nil {
  26. return err
  27. }
  28. numTopics, err := pd.getArrayCount()
  29. if err != nil {
  30. return err
  31. }
  32. r.Blocks = make(map[string]map[int32]*OffsetFetchResponseBlock, numTopics)
  33. for i := 0; i < numTopics; i++ {
  34. name, err := pd.getString()
  35. if err != nil {
  36. return err
  37. }
  38. numBlocks, err := pd.getArrayCount()
  39. if err != nil {
  40. return err
  41. }
  42. r.Blocks[name] = make(map[int32]*OffsetFetchResponseBlock, numBlocks)
  43. for j := 0; j < numBlocks; j++ {
  44. id, err := pd.getInt32()
  45. if err != nil {
  46. return err
  47. }
  48. block := new(OffsetFetchResponseBlock)
  49. err = block.decode(pd)
  50. if err != nil {
  51. return err
  52. }
  53. r.Blocks[name][id] = block
  54. }
  55. }
  56. return nil
  57. }