request.go 647 B

1234567891011121314151617181920212223242526272829303132
  1. package kafka
  2. type API struct {
  3. key int16
  4. version int16
  5. }
  6. var (
  7. REQUEST_PRODUCE = API{0, 0}
  8. REQUEST_FETCH = API{1, 0}
  9. REQUEST_OFFSET = API{2, 0}
  10. REQUEST_METADATA = API{3, 0}
  11. REQUEST_LEADER_AND_ISR = API{4, 0}
  12. REQUEST_STOP_REPLICA = API{5, 0}
  13. REQUEST_OFFSET_COMMIT = API{6, 0}
  14. REQUEST_OFFSET_FETCH = API{7, 0}
  15. )
  16. type request struct {
  17. api API
  18. correlation_id int32
  19. id *string
  20. body encoder
  21. }
  22. func (r *request) encode(pe packetEncoder) {
  23. pe.putInt16(r.api.key)
  24. pe.putInt16(r.api.version)
  25. pe.putInt32(r.correlation_id)
  26. pe.putString(r.id)
  27. r.body.encode(pe)
  28. }