|
@@ -1,10 +1,5 @@
|
|
|
package sarama
|
|
|
|
|
|
-import (
|
|
|
- "net"
|
|
|
- "strconv"
|
|
|
-)
|
|
|
-
|
|
|
type ConsumerMetadataResponse struct {
|
|
|
Err KError
|
|
|
Coordinator *Broker
|
|
@@ -14,61 +9,35 @@ type ConsumerMetadataResponse struct {
|
|
|
}
|
|
|
|
|
|
func (r *ConsumerMetadataResponse) decode(pd packetDecoder, version int16) (err error) {
|
|
|
- tmp, err := pd.getInt16()
|
|
|
- if err != nil {
|
|
|
- return err
|
|
|
- }
|
|
|
- r.Err = KError(tmp)
|
|
|
+ tmp := new(FindCoordinatorResponse)
|
|
|
|
|
|
- coordinator := new(Broker)
|
|
|
- if err := coordinator.decode(pd); err != nil {
|
|
|
+ if err := tmp.decode(pd, version); err != nil {
|
|
|
return err
|
|
|
}
|
|
|
- if coordinator.addr == ":0" {
|
|
|
- return nil
|
|
|
- }
|
|
|
- r.Coordinator = coordinator
|
|
|
|
|
|
- // this can all go away in 2.0, but we have to fill in deprecated fields to maintain
|
|
|
- // backwards compatibility
|
|
|
- host, portstr, err := net.SplitHostPort(r.Coordinator.Addr())
|
|
|
- if err != nil {
|
|
|
- return err
|
|
|
- }
|
|
|
- port, err := strconv.ParseInt(portstr, 10, 32)
|
|
|
- if err != nil {
|
|
|
- return err
|
|
|
- }
|
|
|
- r.CoordinatorID = r.Coordinator.ID()
|
|
|
- r.CoordinatorHost = host
|
|
|
- r.CoordinatorPort = int32(port)
|
|
|
+ r.Err = tmp.Err
|
|
|
+ r.Coordinator = tmp.Coordinator
|
|
|
+ r.CoordinatorID = tmp.CoordinatorID
|
|
|
+ r.CoordinatorHost = tmp.CoordinatorHost
|
|
|
+ r.CoordinatorPort = tmp.CoordinatorPort
|
|
|
|
|
|
return nil
|
|
|
}
|
|
|
|
|
|
func (r *ConsumerMetadataResponse) encode(pe packetEncoder) error {
|
|
|
- pe.putInt16(int16(r.Err))
|
|
|
- if r.Coordinator != nil {
|
|
|
- host, portstr, err := net.SplitHostPort(r.Coordinator.Addr())
|
|
|
- if err != nil {
|
|
|
- return err
|
|
|
- }
|
|
|
- port, err := strconv.ParseInt(portstr, 10, 32)
|
|
|
- if err != nil {
|
|
|
- return err
|
|
|
- }
|
|
|
- pe.putInt32(r.Coordinator.ID())
|
|
|
- if err := pe.putString(host); err != nil {
|
|
|
- return err
|
|
|
- }
|
|
|
- pe.putInt32(int32(port))
|
|
|
- return nil
|
|
|
+ tmp := &FindCoordinatorResponse{
|
|
|
+ Version: 0,
|
|
|
+ Err: r.Err,
|
|
|
+ Coordinator: r.Coordinator,
|
|
|
+ CoordinatorID: r.CoordinatorID,
|
|
|
+ CoordinatorHost: r.CoordinatorHost,
|
|
|
+ CoordinatorPort: r.CoordinatorPort,
|
|
|
}
|
|
|
- pe.putInt32(r.CoordinatorID)
|
|
|
- if err := pe.putString(r.CoordinatorHost); err != nil {
|
|
|
+
|
|
|
+ if err := tmp.encode(pe); err != nil {
|
|
|
return err
|
|
|
}
|
|
|
- pe.putInt32(r.CoordinatorPort)
|
|
|
+
|
|
|
return nil
|
|
|
}
|
|
|
|