grpc
/
grpc-go
mirror of https://github.com/grpc/grpc-go.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

internal: clean up deprecated Invoke() usage (#1966)

This commit is contained in:
Jean de Klerk 2018-04-05 14:09:56 -07:00 committed by Menghan Li
parent 76b07ed73d
commit 4b17696981
4 changed files with 72 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
balancer_switching_test.go
View File

@ -40,7 +40,7 @@ func checkPickFirst(cc *ClientConn, servers []*server) error {
) )
connected := false connected := false
for i := 0; i < 5000; i++ { for i := 0; i < 5000; i++ {
if err = Invoke(context.Background(), "/foo/bar", &req, &reply, cc); errorDesc(err) == servers[0].port { if err = cc.Invoke(context.Background(), "/foo/bar", &req, &reply); errorDesc(err) == servers[0].port {
if connected { if connected {
// connected is set to false if peer is not server[0]. So if // connected is set to false if peer is not server[0]. So if
// connected is true here, this is the second time we saw // connected is true here, this is the second time we saw
@ -58,7 +58,7 @@ func checkPickFirst(cc *ClientConn, servers []*server) error {
} }
// The following RPCs should all succeed with the first server. // The following RPCs should all succeed with the first server.
for i := 0; i < 3; i++ { for i := 0; i < 3; i++ {
err = Invoke(context.Background(), "/foo/bar", &req, &reply, cc) err = cc.Invoke(context.Background(), "/foo/bar", &req, &reply)
if errorDesc(err) != servers[0].port { if errorDesc(err) != servers[0].port {
return fmt.Errorf("Index %d: want peer %v, got peer %v", i, servers[0].port, err) return fmt.Errorf("Index %d: want peer %v, got peer %v", i, servers[0].port, err)
} }
@ -80,7 +80,7 @@ func checkRoundRobin(cc *ClientConn, servers []*server) error {
for _, s := range servers { for _, s := range servers {
var up bool var up bool
for i := 0; i < 5000; i++ { for i := 0; i < 5000; i++ {
if err = Invoke(context.Background(), "/foo/bar", &req, &reply, cc); errorDesc(err) == s.port { if err = cc.Invoke(context.Background(), "/foo/bar", &req, &reply); errorDesc(err) == s.port {
up = true up = true
break break
} }
@ -94,7 +94,7 @@ func checkRoundRobin(cc *ClientConn, servers []*server) error {
serverCount := len(servers) serverCount := len(servers)
for i := 0; i < 3*serverCount; i++ { for i := 0; i < 3*serverCount; i++ {
err = Invoke(context.Background(), "/foo/bar", &req, &reply, cc) err = cc.Invoke(context.Background(), "/foo/bar", &req, &reply)
if errorDesc(err) != servers[i%serverCount].port { if errorDesc(err) != servers[i%serverCount].port {
return fmt.Errorf("Index %d: want peer %v, got peer %v", i, servers[i%serverCount].port, err) return fmt.Errorf("Index %d: want peer %v, got peer %v", i, servers[i%serverCount].port, err)
} }

78
balancer_test.go
View File

@ -130,7 +130,7 @@ func TestNameDiscovery(t *testing.T) {
defer cc.Close() defer cc.Close()
req := "port" req := "port"
var reply string var reply string
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[0].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[0].port {
t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, want %s", err, servers[0].port) t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, want %s", err, servers[0].port)
} }
// Inject the name resolution change to remove servers[0] and add servers[1]. // Inject the name resolution change to remove servers[0] and add servers[1].
@ -146,7 +146,7 @@ func TestNameDiscovery(t *testing.T) {
r.w.inject(updates) r.w.inject(updates)
// Loop until the rpcs in flight talks to servers[1]. // Loop until the rpcs in flight talks to servers[1].
for { for {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[1].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[1].port {
break break
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -163,7 +163,7 @@ func TestEmptyAddrs(t *testing.T) {
} }
defer cc.Close() defer cc.Close()
var reply string var reply string
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc); err != nil || reply != expectedResponse { if err := cc.Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply); err != nil || reply != expectedResponse {
t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, reply = %q, want %q, <nil>", err, reply, expectedResponse) t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, reply = %q, want %q, <nil>", err, reply, expectedResponse)
} }
// Inject name resolution change to remove the server so that there is no address // Inject name resolution change to remove the server so that there is no address
@ -177,7 +177,7 @@ func TestEmptyAddrs(t *testing.T) {
for { for {
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
if err := Invoke(ctx, "/foo/bar", &expectedRequest, &reply, cc); err != nil { if err := cc.Invoke(ctx, "/foo/bar", &expectedRequest, &reply); err != nil {
cancel() cancel()
break break
} }
@ -206,7 +206,7 @@ func TestRoundRobin(t *testing.T) {
var reply string var reply string
// Loop until servers[1] is up // Loop until servers[1] is up
for { for {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[1].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[1].port {
break break
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -219,14 +219,14 @@ func TestRoundRobin(t *testing.T) {
r.w.inject([]*naming.Update{u}) r.w.inject([]*naming.Update{u})
// Loop until both servers[2] are up. // Loop until both servers[2] are up.
for { for {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[2].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[2].port {
break break
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
} }
// Check the incoming RPCs served in a round-robin manner. // Check the incoming RPCs served in a round-robin manner.
for i := 0; i < 10; i++ { for i := 0; i < 10; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[i%numServers].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[i%numServers].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", i, err, servers[i%numServers].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", i, err, servers[i%numServers].port)
} }
} }
@ -242,7 +242,7 @@ func TestCloseWithPendingRPC(t *testing.T) {
} }
defer cc.Close() defer cc.Close()
var reply string var reply string
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc, FailFast(false)); err != nil { if err := cc.Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, FailFast(false)); err != nil {
t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, want %s", err, servers[0].port) t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, want %s", err, servers[0].port)
} }
// Remove the server. // Remove the server.
@ -254,7 +254,7 @@ func TestCloseWithPendingRPC(t *testing.T) {
// Loop until the above update applies. // Loop until the above update applies.
for { for {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
if err := Invoke(ctx, "/foo/bar", &expectedRequest, &reply, cc, FailFast(false)); status.Code(err) == codes.DeadlineExceeded { if err := cc.Invoke(ctx, "/foo/bar", &expectedRequest, &reply, FailFast(false)); status.Code(err) == codes.DeadlineExceeded {
cancel() cancel()
break break
} }
@ -267,7 +267,7 @@ func TestCloseWithPendingRPC(t *testing.T) {
go func() { go func() {
defer wg.Done() defer wg.Done()
var reply string var reply string
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc, FailFast(false)); err == nil { if err := cc.Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, FailFast(false)); err == nil {
t.Errorf("grpc.Invoke(_, _, _, _, _) = %v, want not nil", err) t.Errorf("grpc.Invoke(_, _, _, _, _) = %v, want not nil", err)
} }
}() }()
@ -275,7 +275,7 @@ func TestCloseWithPendingRPC(t *testing.T) {
defer wg.Done() defer wg.Done()
var reply string var reply string
time.Sleep(5 * time.Millisecond) time.Sleep(5 * time.Millisecond)
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc, FailFast(false)); err == nil { if err := cc.Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, FailFast(false)); err == nil {
t.Errorf("grpc.Invoke(_, _, _, _, _) = %v, want not nil", err) t.Errorf("grpc.Invoke(_, _, _, _, _) = %v, want not nil", err)
} }
}() }()
@ -302,7 +302,7 @@ func TestGetOnWaitChannel(t *testing.T) {
for { for {
var reply string var reply string
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
if err := Invoke(ctx, "/foo/bar", &expectedRequest, &reply, cc, FailFast(false)); status.Code(err) == codes.DeadlineExceeded { if err := cc.Invoke(ctx, "/foo/bar", &expectedRequest, &reply, FailFast(false)); status.Code(err) == codes.DeadlineExceeded {
cancel() cancel()
break break
} }
@ -314,7 +314,7 @@ func TestGetOnWaitChannel(t *testing.T) {
go func() { go func() {
defer wg.Done() defer wg.Done()
var reply string var reply string
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc, FailFast(false)); err != nil { if err := cc.Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, FailFast(false)); err != nil {
t.Errorf("grpc.Invoke(_, _, _, _, _) = %v, want <nil>", err) t.Errorf("grpc.Invoke(_, _, _, _, _) = %v, want <nil>", err)
} }
}() }()
@ -350,7 +350,7 @@ func TestOneServerDown(t *testing.T) {
var reply string var reply string
// Loop until servers[1] is up // Loop until servers[1] is up
for { for {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[1].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[1].port {
break break
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -374,7 +374,7 @@ func TestOneServerDown(t *testing.T) {
time.Sleep(sleepDuration) time.Sleep(sleepDuration)
// After sleepDuration, invoke RPC. // After sleepDuration, invoke RPC.
// server[0] is killed around the same time to make it racy between balancer and gRPC internals. // server[0] is killed around the same time to make it racy between balancer and gRPC internals.
Invoke(context.Background(), "/foo/bar", &req, &reply, cc, FailFast(false)) cc.Invoke(context.Background(), "/foo/bar", &req, &reply, FailFast(false))
wg.Done() wg.Done()
}() }()
} }
@ -403,7 +403,7 @@ func TestOneAddressRemoval(t *testing.T) {
var reply string var reply string
// Loop until servers[1] is up // Loop until servers[1] is up
for { for {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[1].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[1].port {
break break
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -433,7 +433,7 @@ func TestOneAddressRemoval(t *testing.T) {
time.Sleep(sleepDuration) time.Sleep(sleepDuration)
// After sleepDuration, invoke RPC. // After sleepDuration, invoke RPC.
// server[0] is removed around the same time to make it racy between balancer and gRPC internals. // server[0] is removed around the same time to make it racy between balancer and gRPC internals.
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc, FailFast(false)); err != nil { if err := cc.Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, FailFast(false)); err != nil {
t.Errorf("grpc.Invoke(_, _, _, _, _) = %v, want not nil", err) t.Errorf("grpc.Invoke(_, _, _, _, _) = %v, want not nil", err)
} }
wg.Done() wg.Done()
@ -452,7 +452,7 @@ func checkServerUp(t *testing.T, currentServer *server) {
defer cc.Close() defer cc.Close()
var reply string var reply string
for { for {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == port {
break break
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -469,7 +469,7 @@ func TestPickFirstEmptyAddrs(t *testing.T) {
} }
defer cc.Close() defer cc.Close()
var reply string var reply string
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc); err != nil || reply != expectedResponse { if err := cc.Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply); err != nil || reply != expectedResponse {
t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, reply = %q, want %q, <nil>", err, reply, expectedResponse) t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, reply = %q, want %q, <nil>", err, reply, expectedResponse)
} }
// Inject name resolution change to remove the server so that there is no address // Inject name resolution change to remove the server so that there is no address
@ -483,7 +483,7 @@ func TestPickFirstEmptyAddrs(t *testing.T) {
for { for {
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
if err := Invoke(ctx, "/foo/bar", &expectedRequest, &reply, cc); err != nil { if err := cc.Invoke(ctx, "/foo/bar", &expectedRequest, &reply); err != nil {
cancel() cancel()
break break
} }
@ -501,7 +501,7 @@ func TestPickFirstCloseWithPendingRPC(t *testing.T) {
} }
defer cc.Close() defer cc.Close()
var reply string var reply string
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc, FailFast(false)); err != nil { if err := cc.Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, FailFast(false)); err != nil {
t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, want %s", err, servers[0].port) t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, want %s", err, servers[0].port)
} }
// Remove the server. // Remove the server.
@ -513,7 +513,7 @@ func TestPickFirstCloseWithPendingRPC(t *testing.T) {
// Loop until the above update applies. // Loop until the above update applies.
for { for {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
if err := Invoke(ctx, "/foo/bar", &expectedRequest, &reply, cc, FailFast(false)); status.Code(err) == codes.DeadlineExceeded { if err := cc.Invoke(ctx, "/foo/bar", &expectedRequest, &reply, FailFast(false)); status.Code(err) == codes.DeadlineExceeded {
cancel() cancel()
break break
} }
@ -526,7 +526,7 @@ func TestPickFirstCloseWithPendingRPC(t *testing.T) {
go func() { go func() {
defer wg.Done() defer wg.Done()
var reply string var reply string
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc, FailFast(false)); err == nil { if err := cc.Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, FailFast(false)); err == nil {
t.Errorf("grpc.Invoke(_, _, _, _, _) = %v, want not nil", err) t.Errorf("grpc.Invoke(_, _, _, _, _) = %v, want not nil", err)
} }
}() }()
@ -534,7 +534,7 @@ func TestPickFirstCloseWithPendingRPC(t *testing.T) {
defer wg.Done() defer wg.Done()
var reply string var reply string
time.Sleep(5 * time.Millisecond) time.Sleep(5 * time.Millisecond)
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc, FailFast(false)); err == nil { if err := cc.Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, FailFast(false)); err == nil {
t.Errorf("grpc.Invoke(_, _, _, _, _) = %v, want not nil", err) t.Errorf("grpc.Invoke(_, _, _, _, _) = %v, want not nil", err)
} }
}() }()
@ -576,7 +576,7 @@ func TestPickFirstOrderAllServerUp(t *testing.T) {
req := "port" req := "port"
var reply string var reply string
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[0].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[0].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 0, err, servers[0].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 0, err, servers[0].port)
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -591,13 +591,13 @@ func TestPickFirstOrderAllServerUp(t *testing.T) {
r.w.inject([]*naming.Update{u}) r.w.inject([]*naming.Update{u})
// Loop until it changes to server[1] // Loop until it changes to server[1]
for { for {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[1].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[1].port {
break break
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
} }
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[1].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[1].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 1, err, servers[1].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 1, err, servers[1].port)
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -611,7 +611,7 @@ func TestPickFirstOrderAllServerUp(t *testing.T) {
} }
r.w.inject([]*naming.Update{u}) r.w.inject([]*naming.Update{u})
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[1].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[1].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 1, err, servers[1].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 1, err, servers[1].port)
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -624,13 +624,13 @@ func TestPickFirstOrderAllServerUp(t *testing.T) {
} }
r.w.inject([]*naming.Update{u}) r.w.inject([]*naming.Update{u})
for { for {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[2].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[2].port {
break break
} }
time.Sleep(1 * time.Second) time.Sleep(1 * time.Second)
} }
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[2].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[2].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 2, err, servers[2].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 2, err, servers[2].port)
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -643,13 +643,13 @@ func TestPickFirstOrderAllServerUp(t *testing.T) {
} }
r.w.inject([]*naming.Update{u}) r.w.inject([]*naming.Update{u})
for { for {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[0].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[0].port {
break break
} }
time.Sleep(1 * time.Second) time.Sleep(1 * time.Second)
} }
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[0].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[0].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 0, err, servers[0].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 0, err, servers[0].port)
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -689,7 +689,7 @@ func TestPickFirstOrderOneServerDown(t *testing.T) {
req := "port" req := "port"
var reply string var reply string
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[0].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[0].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 0, err, servers[0].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 0, err, servers[0].port)
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -700,13 +700,13 @@ func TestPickFirstOrderOneServerDown(t *testing.T) {
servers[0].stop() servers[0].stop()
// Loop until it changes to server[1] // Loop until it changes to server[1]
for { for {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[1].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[1].port {
break break
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
} }
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[1].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[1].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 1, err, servers[1].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 1, err, servers[1].port)
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -721,7 +721,7 @@ func TestPickFirstOrderOneServerDown(t *testing.T) {
checkServerUp(t, servers[0]) checkServerUp(t, servers[0])
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[1].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[1].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 1, err, servers[1].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 1, err, servers[1].port)
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -734,13 +734,13 @@ func TestPickFirstOrderOneServerDown(t *testing.T) {
} }
r.w.inject([]*naming.Update{u}) r.w.inject([]*naming.Update{u})
for { for {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[0].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[0].port {
break break
} }
time.Sleep(1 * time.Second) time.Sleep(1 * time.Second)
} }
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[0].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[0].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 0, err, servers[0].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 0, err, servers[0].port)
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -794,7 +794,7 @@ func TestPickFirstOneAddressRemoval(t *testing.T) {
time.Sleep(sleepDuration) time.Sleep(sleepDuration)
// After sleepDuration, invoke RPC. // After sleepDuration, invoke RPC.
// server[0] is removed around the same time to make it racy between balancer and gRPC internals. // server[0] is removed around the same time to make it racy between balancer and gRPC internals.
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc, FailFast(false)); err != nil { if err := cc.Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, FailFast(false)); err != nil {
t.Errorf("grpc.Invoke(_, _, _, _, _) = %v, want not nil", err) t.Errorf("grpc.Invoke(_, _, _, _, _) = %v, want not nil", err)
} }
wg.Done() wg.Done()

10
call_test.go
View File

@ -217,7 +217,7 @@ func TestInvoke(t *testing.T) {
defer leakcheck.Check(t) defer leakcheck.Check(t)
server, cc := setUp(t, 0, math.MaxUint32) server, cc := setUp(t, 0, math.MaxUint32)
var reply string var reply string
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc); err != nil || reply != expectedResponse { if err := cc.Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply); err != nil || reply != expectedResponse {
t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, want <nil>", err) t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, want <nil>", err)
} }
cc.Close() cc.Close()
@ -229,7 +229,7 @@ func TestInvokeLargeErr(t *testing.T) {
server, cc := setUp(t, 0, math.MaxUint32) server, cc := setUp(t, 0, math.MaxUint32)
var reply string var reply string
req := "hello" req := "hello"
err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc) err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply)
if _, ok := status.FromError(err); !ok { if _, ok := status.FromError(err); !ok {
t.Fatalf("grpc.Invoke(_, _, _, _, _) receives non rpc error.") t.Fatalf("grpc.Invoke(_, _, _, _, _) receives non rpc error.")
} }
@ -246,7 +246,7 @@ func TestInvokeErrorSpecialChars(t *testing.T) {
server, cc := setUp(t, 0, math.MaxUint32) server, cc := setUp(t, 0, math.MaxUint32)
var reply string var reply string
req := "weird error" req := "weird error"
err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc) err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply)
if _, ok := status.FromError(err); !ok { if _, ok := status.FromError(err); !ok {
t.Fatalf("grpc.Invoke(_, _, _, _, _) receives non rpc error.") t.Fatalf("grpc.Invoke(_, _, _, _, _) receives non rpc error.")
} }
@ -266,7 +266,7 @@ func TestInvokeCancel(t *testing.T) {
for i := 0; i < 100; i++ { for i := 0; i < 100; i++ {
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
cancel() cancel()
Invoke(ctx, "/foo/bar", &req, &reply, cc) cc.Invoke(ctx, "/foo/bar", &req, &reply)
} }
if canceled != 0 { if canceled != 0 {
t.Fatalf("received %d of 100 canceled requests", canceled) t.Fatalf("received %d of 100 canceled requests", canceled)
@ -285,7 +285,7 @@ func TestInvokeCancelClosedNonFailFast(t *testing.T) {
req := "hello" req := "hello"
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
cancel() cancel()
if err := Invoke(ctx, "/foo/bar", &req, &reply, cc, FailFast(false)); err == nil { if err := cc.Invoke(ctx, "/foo/bar", &req, &reply, FailFast(false)); err == nil {
t.Fatalf("canceled invoke on closed connection should fail") t.Fatalf("canceled invoke on closed connection should fail")
} }
server.stop() server.stop()

48
pickfirst_test.go
View File

@ -58,14 +58,14 @@ func TestOneBackendPickfirst(t *testing.T) {
defer cancel() defer cancel()
req := "port" req := "port"
var reply string var reply string
if err := Invoke(ctx, "/foo/bar", &req, &reply, cc); err == nil || status.Code(err) != codes.DeadlineExceeded { if err := cc.Invoke(ctx, "/foo/bar", &req, &reply); err == nil || status.Code(err) != codes.DeadlineExceeded {
t.Fatalf("EmptyCall() = _, %v, want _, DeadlineExceeded", err) t.Fatalf("EmptyCall() = _, %v, want _, DeadlineExceeded", err)
} }
r.NewAddress([]resolver.Address{{Addr: servers[0].addr}}) r.NewAddress([]resolver.Address{{Addr: servers[0].addr}})
// The second RPC should succeed. // The second RPC should succeed.
for i := 0; i < 1000; i++ { for i := 0; i < 1000; i++ {
if err = Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[0].port { if err = cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[0].port {
return return
} }
time.Sleep(time.Millisecond) time.Sleep(time.Millisecond)
@ -92,14 +92,14 @@ func TestBackendsPickfirst(t *testing.T) {
defer cancel() defer cancel()
req := "port" req := "port"
var reply string var reply string
if err := Invoke(ctx, "/foo/bar", &req, &reply, cc); err == nil || status.Code(err) != codes.DeadlineExceeded { if err := cc.Invoke(ctx, "/foo/bar", &req, &reply); err == nil || status.Code(err) != codes.DeadlineExceeded {
t.Fatalf("EmptyCall() = _, %v, want _, DeadlineExceeded", err) t.Fatalf("EmptyCall() = _, %v, want _, DeadlineExceeded", err)
} }
r.NewAddress([]resolver.Address{{Addr: servers[0].addr}, {Addr: servers[1].addr}}) r.NewAddress([]resolver.Address{{Addr: servers[0].addr}, {Addr: servers[1].addr}})
// The second RPC should succeed with the first server. // The second RPC should succeed with the first server.
for i := 0; i < 1000; i++ { for i := 0; i < 1000; i++ {
if err = Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[0].port { if err = cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[0].port {
return return
} }
time.Sleep(time.Millisecond) time.Sleep(time.Millisecond)
@ -126,7 +126,7 @@ func TestNewAddressWhileBlockingPickfirst(t *testing.T) {
defer cancel() defer cancel()
req := "port" req := "port"
var reply string var reply string
if err := Invoke(ctx, "/foo/bar", &req, &reply, cc); err == nil || status.Code(err) != codes.DeadlineExceeded { if err := cc.Invoke(ctx, "/foo/bar", &req, &reply); err == nil || status.Code(err) != codes.DeadlineExceeded {
t.Fatalf("EmptyCall() = _, %v, want _, DeadlineExceeded", err) t.Fatalf("EmptyCall() = _, %v, want _, DeadlineExceeded", err)
} }
@ -136,7 +136,7 @@ func TestNewAddressWhileBlockingPickfirst(t *testing.T) {
go func() { go func() {
defer wg.Done() defer wg.Done()
// This RPC blocks until NewAddress is called. // This RPC blocks until NewAddress is called.
Invoke(context.Background(), "/foo/bar", &req, &reply, cc) cc.Invoke(context.Background(), "/foo/bar", &req, &reply)
}() }()
} }
time.Sleep(50 * time.Millisecond) time.Sleep(50 * time.Millisecond)
@ -163,7 +163,7 @@ func TestCloseWithPendingRPCPickfirst(t *testing.T) {
defer cancel() defer cancel()
req := "port" req := "port"
var reply string var reply string
if err := Invoke(ctx, "/foo/bar", &req, &reply, cc); err == nil || status.Code(err) != codes.DeadlineExceeded { if err := cc.Invoke(ctx, "/foo/bar", &req, &reply); err == nil || status.Code(err) != codes.DeadlineExceeded {
t.Fatalf("EmptyCall() = _, %v, want _, DeadlineExceeded", err) t.Fatalf("EmptyCall() = _, %v, want _, DeadlineExceeded", err)
} }
@ -173,7 +173,7 @@ func TestCloseWithPendingRPCPickfirst(t *testing.T) {
go func() { go func() {
defer wg.Done() defer wg.Done()
// This RPC blocks until NewAddress is called. // This RPC blocks until NewAddress is called.
Invoke(context.Background(), "/foo/bar", &req, &reply, cc) cc.Invoke(context.Background(), "/foo/bar", &req, &reply)
}() }()
} }
time.Sleep(50 * time.Millisecond) time.Sleep(50 * time.Millisecond)
@ -200,14 +200,14 @@ func TestOneServerDownPickfirst(t *testing.T) {
defer cancel() defer cancel()
req := "port" req := "port"
var reply string var reply string
if err := Invoke(ctx, "/foo/bar", &req, &reply, cc); err == nil || status.Code(err) != codes.DeadlineExceeded { if err := cc.Invoke(ctx, "/foo/bar", &req, &reply); err == nil || status.Code(err) != codes.DeadlineExceeded {
t.Fatalf("EmptyCall() = _, %v, want _, DeadlineExceeded", err) t.Fatalf("EmptyCall() = _, %v, want _, DeadlineExceeded", err)
} }
r.NewAddress([]resolver.Address{{Addr: servers[0].addr}, {Addr: servers[1].addr}}) r.NewAddress([]resolver.Address{{Addr: servers[0].addr}, {Addr: servers[1].addr}})
// The second RPC should succeed with the first server. // The second RPC should succeed with the first server.
for i := 0; i < 1000; i++ { for i := 0; i < 1000; i++ {
if err = Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[0].port { if err = cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[0].port {
break break
} }
time.Sleep(time.Millisecond) time.Sleep(time.Millisecond)
@ -215,7 +215,7 @@ func TestOneServerDownPickfirst(t *testing.T) {
servers[0].stop() servers[0].stop()
for i := 0; i < 1000; i++ { for i := 0; i < 1000; i++ {
if err = Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[1].port { if err = cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[1].port {
return return
} }
time.Sleep(time.Millisecond) time.Sleep(time.Millisecond)
@ -242,14 +242,14 @@ func TestAllServersDownPickfirst(t *testing.T) {
defer cancel() defer cancel()
req := "port" req := "port"
var reply string var reply string
if err := Invoke(ctx, "/foo/bar", &req, &reply, cc); err == nil || status.Code(err) != codes.DeadlineExceeded { if err := cc.Invoke(ctx, "/foo/bar", &req, &reply); err == nil || status.Code(err) != codes.DeadlineExceeded {
t.Fatalf("EmptyCall() = _, %v, want _, DeadlineExceeded", err) t.Fatalf("EmptyCall() = _, %v, want _, DeadlineExceeded", err)
} }
r.NewAddress([]resolver.Address{{Addr: servers[0].addr}, {Addr: servers[1].addr}}) r.NewAddress([]resolver.Address{{Addr: servers[0].addr}, {Addr: servers[1].addr}})
// The second RPC should succeed with the first server. // The second RPC should succeed with the first server.
for i := 0; i < 1000; i++ { for i := 0; i < 1000; i++ {
if err = Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[0].port { if err = cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[0].port {
break break
} }
time.Sleep(time.Millisecond) time.Sleep(time.Millisecond)
@ -259,7 +259,7 @@ func TestAllServersDownPickfirst(t *testing.T) {
servers[i].stop() servers[i].stop()
} }
for i := 0; i < 1000; i++ { for i := 0; i < 1000; i++ {
if err = Invoke(context.Background(), "/foo/bar", &req, &reply, cc); status.Code(err) == codes.Unavailable { if err = cc.Invoke(context.Background(), "/foo/bar", &req, &reply); status.Code(err) == codes.Unavailable {
return return
} }
time.Sleep(time.Millisecond) time.Sleep(time.Millisecond)
@ -286,19 +286,19 @@ func TestAddressesRemovedPickfirst(t *testing.T) {
defer cancel() defer cancel()
req := "port" req := "port"
var reply string var reply string
if err := Invoke(ctx, "/foo/bar", &req, &reply, cc); err == nil || status.Code(err) != codes.DeadlineExceeded { if err := cc.Invoke(ctx, "/foo/bar", &req, &reply); err == nil || status.Code(err) != codes.DeadlineExceeded {
t.Fatalf("EmptyCall() = _, %v, want _, DeadlineExceeded", err) t.Fatalf("EmptyCall() = _, %v, want _, DeadlineExceeded", err)
} }
r.NewAddress([]resolver.Address{{Addr: servers[0].addr}, {Addr: servers[1].addr}, {Addr: servers[2].addr}}) r.NewAddress([]resolver.Address{{Addr: servers[0].addr}, {Addr: servers[1].addr}, {Addr: servers[2].addr}})
for i := 0; i < 1000; i++ { for i := 0; i < 1000; i++ {
if err = Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[0].port { if err = cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[0].port {
break break
} }
time.Sleep(time.Millisecond) time.Sleep(time.Millisecond)
} }
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[0].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[0].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 0, err, servers[0].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 0, err, servers[0].port)
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -307,13 +307,13 @@ func TestAddressesRemovedPickfirst(t *testing.T) {
// Remove server[0]. // Remove server[0].
r.NewAddress([]resolver.Address{{Addr: servers[1].addr}, {Addr: servers[2].addr}}) r.NewAddress([]resolver.Address{{Addr: servers[1].addr}, {Addr: servers[2].addr}})
for i := 0; i < 1000; i++ { for i := 0; i < 1000; i++ {
if err = Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[1].port { if err = cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[1].port {
break break
} }
time.Sleep(time.Millisecond) time.Sleep(time.Millisecond)
} }
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[1].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[1].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 1, err, servers[1].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 1, err, servers[1].port)
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -322,7 +322,7 @@ func TestAddressesRemovedPickfirst(t *testing.T) {
// Append server[0], nothing should change. // Append server[0], nothing should change.
r.NewAddress([]resolver.Address{{Addr: servers[1].addr}, {Addr: servers[2].addr}, {Addr: servers[0].addr}}) r.NewAddress([]resolver.Address{{Addr: servers[1].addr}, {Addr: servers[2].addr}, {Addr: servers[0].addr}})
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[1].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[1].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 1, err, servers[1].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 1, err, servers[1].port)
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -331,13 +331,13 @@ func TestAddressesRemovedPickfirst(t *testing.T) {
// Remove server[1]. // Remove server[1].
r.NewAddress([]resolver.Address{{Addr: servers[2].addr}, {Addr: servers[0].addr}}) r.NewAddress([]resolver.Address{{Addr: servers[2].addr}, {Addr: servers[0].addr}})
for i := 0; i < 1000; i++ { for i := 0; i < 1000; i++ {
if err = Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[2].port { if err = cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[2].port {
break break
} }
time.Sleep(time.Millisecond) time.Sleep(time.Millisecond)
} }
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[2].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[2].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 2, err, servers[2].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 2, err, servers[2].port)
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)
@ -346,13 +346,13 @@ func TestAddressesRemovedPickfirst(t *testing.T) {
// Remove server[2]. // Remove server[2].
r.NewAddress([]resolver.Address{{Addr: servers[0].addr}}) r.NewAddress([]resolver.Address{{Addr: servers[0].addr}})
for i := 0; i < 1000; i++ { for i := 0; i < 1000; i++ {
if err = Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && errorDesc(err) == servers[0].port { if err = cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err != nil && errorDesc(err) == servers[0].port {
break break
} }
time.Sleep(time.Millisecond) time.Sleep(time.Millisecond)
} }
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || errorDesc(err) != servers[0].port { if err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply); err == nil || errorDesc(err) != servers[0].port {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 0, err, servers[0].port) t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", 0, err, servers[0].port)
} }
time.Sleep(10 * time.Millisecond) time.Sleep(10 * time.Millisecond)