offset_fetch_response.go 1.2 KB

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