offset_fetch_response.go 1.4 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182
  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. Blocks map[string]map[int32]*OffsetFetchResponseBlock
  34. }
  35. func (r *OffsetFetchResponse) decode(pd packetDecoder) (err error) {
  36. numTopics, err := pd.getArrayLength()
  37. if err != nil {
  38. return err
  39. }
  40. r.Blocks = make(map[string]map[int32]*OffsetFetchResponseBlock, numTopics)
  41. for i := 0; i < numTopics; i++ {
  42. name, err := pd.getString()
  43. if err != nil {
  44. return err
  45. }
  46. numBlocks, err := pd.getArrayLength()
  47. if err != nil {
  48. return err
  49. }
  50. r.Blocks[name] = make(map[int32]*OffsetFetchResponseBlock, numBlocks)
  51. for j := 0; j < numBlocks; j++ {
  52. id, err := pd.getInt32()
  53. if err != nil {
  54. return err
  55. }
  56. block := new(OffsetFetchResponseBlock)
  57. err = block.decode(pd)
  58. if err != nil {
  59. return err
  60. }
  61. r.Blocks[name][id] = block
  62. }
  63. }
  64. return nil
  65. }