Skip to content

Commit

Permalink
remove deprecated fields
Browse files Browse the repository at this point in the history
  • Loading branch information
Robin committed Jan 25, 2018
1 parent a5a9b83 commit d034048
Show file tree
Hide file tree
Showing 4 changed files with 55 additions and 62 deletions.
41 changes: 32 additions & 9 deletions consumer_metadata_response.go
Original file line number Diff line number Diff line change
@@ -1,5 +1,10 @@
package sarama

import (
"net"
"strconv"
)

type ConsumerMetadataResponse struct {
Err KError
Coordinator *Broker
Expand All @@ -16,22 +21,40 @@ func (r *ConsumerMetadataResponse) decode(pd packetDecoder, version int16) (err
}

r.Err = tmp.Err

r.Coordinator = tmp.Coordinator
r.CoordinatorID = tmp.CoordinatorID
r.CoordinatorHost = tmp.CoordinatorHost
r.CoordinatorPort = tmp.CoordinatorPort
if tmp.Coordinator == nil {
return nil
}

// 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)

return nil
}

func (r *ConsumerMetadataResponse) encode(pe packetEncoder) error {
if r.Coordinator == nil {
r.Coordinator = new(Broker)
r.Coordinator.id = r.CoordinatorID
r.Coordinator.addr = net.JoinHostPort(r.CoordinatorHost, strconv.Itoa(int(r.CoordinatorPort)))
}

tmp := &FindCoordinatorResponse{
Version: 0,
Err: r.Err,
Coordinator: r.Coordinator,
CoordinatorID: r.CoordinatorID,
CoordinatorHost: r.CoordinatorHost,
CoordinatorPort: r.CoordinatorPort,
Version: 0,
Err: r.Err,
Coordinator: r.Coordinator,
}

if err := tmp.encode(pe); err != nil {
Expand Down
13 changes: 11 additions & 2 deletions consumer_metadata_response_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,17 @@ var (
)

func TestConsumerMetadataResponseError(t *testing.T) {
response := ConsumerMetadataResponse{Err: ErrOffsetsLoadInProgress}
testResponse(t, "error", &response, consumerMetadataResponseError)
response := &ConsumerMetadataResponse{Err: ErrOffsetsLoadInProgress}
testEncodable(t, "", response, consumerMetadataResponseError)

decodedResp := &ConsumerMetadataResponse{}
if err := versionedDecode(consumerMetadataResponseError, decodedResp, 0); err != nil {
t.Error("could not decode: ", err)
}

if decodedResp.Err != ErrOffsetsLoadInProgress {
t.Errorf("got %s, want %s", decodedResp.Err, ErrOffsetsLoadInProgress)
}
}

func TestConsumerMetadataResponseSuccess(t *testing.T) {
Expand Down
50 changes: 7 additions & 43 deletions find_coordinator_response.go
Original file line number Diff line number Diff line change
@@ -1,20 +1,15 @@
package sarama

import (
"net"
"strconv"
"time"
)

type FindCoordinatorResponse struct {
Version int16
ThrottleTime time.Duration
Err KError
ErrMsg *string
Coordinator *Broker
CoordinatorID int32 // deprecated: use Coordinator.ID()
CoordinatorHost string // deprecated: use Coordinator.Addr()
CoordinatorPort int32 // deprecated: use Coordinator.Addr()
Version int16
ThrottleTime time.Duration
Err KError
ErrMsg *string
Coordinator *Broker
}

func (f *FindCoordinatorResponse) decode(pd packetDecoder, version int16) (err error) {
Expand Down Expand Up @@ -49,20 +44,6 @@ func (f *FindCoordinatorResponse) decode(pd packetDecoder, version int16) (err e
}
f.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(f.Coordinator.Addr())
if err != nil {
return err
}
port, err := strconv.ParseInt(portstr, 10, 32)
if err != nil {
return err
}
f.CoordinatorID = f.Coordinator.ID()
f.CoordinatorHost = host
f.CoordinatorPort = int32(port)

return nil
}

Expand All @@ -79,27 +60,10 @@ func (f *FindCoordinatorResponse) encode(pe packetEncoder) error {
}
}

if f.Coordinator != nil {
host, portstr, err := net.SplitHostPort(f.Coordinator.Addr())
if err != nil {
return err
}
port, err := strconv.ParseInt(portstr, 10, 32)
if err != nil {
return err
}
pe.putInt32(f.Coordinator.ID())
if err := pe.putString(host); err != nil {
return err
}
pe.putInt32(int32(port))
return nil
}
pe.putInt32(f.CoordinatorID)
if err := pe.putString(f.CoordinatorHost); err != nil {
if err := f.Coordinator.encode(pe); err != nil {
return err
}
pe.putInt32(f.CoordinatorPort)

return nil
}

Expand Down
13 changes: 5 additions & 8 deletions find_coordinator_response_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,14 +29,11 @@ func TestFindCoordinatorResponse(t *testing.T) {
broker := NewBroker("host:9092")
broker.id = 1
resp := &FindCoordinatorResponse{
Version: 1,
ThrottleTime: 100 * time.Millisecond,
Err: ErrNoError,
ErrMsg: nil,
CoordinatorID: 1,
CoordinatorHost: "host",
CoordinatorPort: 9092,
Coordinator: broker,
Version: 1,
ThrottleTime: 100 * time.Millisecond,
Err: ErrNoError,
ErrMsg: nil,
Coordinator: broker,
}

testResponse(t, "version 1 - no error", resp, findCoordinatorResponse)
Expand Down

0 comments on commit d034048

Please sign in to comment.