Skip to content

Commit

Permalink
Merge pull request #7852 from heyitsanthony/revert-lease-err-ka
Browse files Browse the repository at this point in the history 
Revert "Merge pull request #7732 from heyitsanthony/lease-err-ka"
  • Loading branch information
Anthony Romano authored May 3, 2017
2 parents 27b3bf2 + 6dd8fb6 commit 24e85b2
Show file tree
Hide file tree
Showing 9 changed files with 145 additions and 138 deletions.
7 changes: 5 additions & 2 deletions clientv3/concurrency/session.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,9 +51,12 @@ func NewSession(client *v3.Client, opts ...SessionOption) (*Session, error) {
}

ctx, cancel := context.WithCancel(ops.ctx)
keepAlive := client.KeepAlive(ctx, id)
donec := make(chan struct{})
keepAlive, err := client.KeepAlive(ctx, id)
if err != nil || keepAlive == nil {
return nil, err
}

donec := make(chan struct{})
s := &Session{client: client, opts: ops, id: id, cancel: cancel, donec: donec}

// keep the lease alive until client error or cancelled context
Expand Down
15 changes: 7 additions & 8 deletions clientv3/example_lease_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -100,13 +100,12 @@ func ExampleLease_keepAlive() {
}

// the key 'foo' will be kept forever
ch := cli.KeepAlive(context.TODO(), resp.ID)

ka := <-ch
if ka.Err != nil {
log.Fatal(ka.Err)
ch, kaerr := cli.KeepAlive(context.TODO(), resp.ID)
if kaerr != nil {
log.Fatal(kaerr)
}

ka := <-ch
fmt.Println("ttl:", ka.TTL)
// Output: ttl: 5
}
Expand All @@ -132,9 +131,9 @@ func ExampleLease_keepAliveOnce() {
}

// to renew the lease only once
ka := cli.KeepAliveOnce(context.TODO(), resp.ID)
if ka.Err != nil {
log.Fatal(ka.Err)
ka, kaerr := cli.KeepAliveOnce(context.TODO(), resp.ID)
if kaerr != nil {
log.Fatal(kaerr)
}

fmt.Println("ttl:", ka.TTL)
Expand Down
76 changes: 53 additions & 23 deletions clientv3/integration/lease_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -104,14 +104,14 @@ func TestLeaseKeepAliveOnce(t *testing.T) {
t.Errorf("failed to create lease %v", err)
}

ka := lapi.KeepAliveOnce(context.Background(), resp.ID)
if ka.Err != nil {
t.Errorf("failed to keepalive lease %v", ka.Err)
_, err = lapi.KeepAliveOnce(context.Background(), resp.ID)
if err != nil {
t.Errorf("failed to keepalive lease %v", err)
}

ka = lapi.KeepAliveOnce(context.Background(), clientv3.LeaseID(0))
if ka.Err != rpctypes.ErrLeaseNotFound {
t.Errorf("expected %v, got %v", rpctypes.ErrLeaseNotFound, ka.Err)
_, err = lapi.KeepAliveOnce(context.Background(), clientv3.LeaseID(0))
if err != rpctypes.ErrLeaseNotFound {
t.Errorf("expected %v, got %v", rpctypes.ErrLeaseNotFound, err)
}
}

Expand All @@ -129,7 +129,10 @@ func TestLeaseKeepAlive(t *testing.T) {
t.Errorf("failed to create lease %v", err)
}

rc := lapi.KeepAlive(context.Background(), resp.ID)
rc, kerr := lapi.KeepAlive(context.Background(), resp.ID)
if kerr != nil {
t.Errorf("failed to keepalive lease %v", kerr)
}

kresp, ok := <-rc
if !ok {
Expand Down Expand Up @@ -160,7 +163,11 @@ func TestLeaseKeepAliveOneSecond(t *testing.T) {
if err != nil {
t.Errorf("failed to create lease %v", err)
}
rc := cli.KeepAlive(context.Background(), resp.ID)
rc, kerr := cli.KeepAlive(context.Background(), resp.ID)
if kerr != nil {
t.Errorf("failed to keepalive lease %v", kerr)
}

for i := 0; i < 3; i++ {
if _, ok := <-rc; !ok {
t.Errorf("chan is closed, want not closed")
Expand All @@ -186,7 +193,10 @@ func TestLeaseKeepAliveHandleFailure(t *testing.T) {
t.Errorf("failed to create lease %v", err)
}

rc := lapi.KeepAlive(context.Background(), resp.ID)
rc, kerr := lapi.KeepAlive(context.Background(), resp.ID)
if kerr != nil {
t.Errorf("failed to keepalive lease %v", kerr)
}

kresp := <-rc
if kresp.ID != resp.ID {
Expand Down Expand Up @@ -220,7 +230,7 @@ func TestLeaseKeepAliveHandleFailure(t *testing.T) {

type leaseCh struct {
lid clientv3.LeaseID
ch clientv3.LeaseKeepAliveChan
ch <-chan *clientv3.LeaseKeepAliveResponse
}

// TestLeaseKeepAliveNotFound ensures a revoked lease won't stop other keep alives
Expand All @@ -237,7 +247,10 @@ func TestLeaseKeepAliveNotFound(t *testing.T) {
if rerr != nil {
t.Fatal(rerr)
}
kach := cli.KeepAlive(context.Background(), resp.ID)
kach, kaerr := cli.KeepAlive(context.Background(), resp.ID)
if kaerr != nil {
t.Fatal(kaerr)
}
lchs = append(lchs, leaseCh{resp.ID, kach})
}

Expand Down Expand Up @@ -362,7 +375,10 @@ func TestLeaseKeepAliveCloseAfterDisconnectRevoke(t *testing.T) {
if err != nil {
t.Fatal(err)
}
rc := cli.KeepAlive(context.Background(), resp.ID)
rc, kerr := cli.KeepAlive(context.Background(), resp.ID)
if kerr != nil {
t.Fatal(kerr)
}
kresp := <-rc
if kresp.ID != resp.ID {
t.Fatalf("ID = %x, want %x", kresp.ID, resp.ID)
Expand All @@ -381,10 +397,9 @@ func TestLeaseKeepAliveCloseAfterDisconnectRevoke(t *testing.T) {

// some keep-alives may still be buffered; drain until close
timer := time.After(time.Duration(kresp.TTL) * time.Second)
loop := true
for loop {
for kresp != nil {
select {
case _, loop = <-rc:
case kresp = <-rc:
case <-timer:
t.Fatalf("keepalive channel did not close")
}
Expand All @@ -408,7 +423,10 @@ func TestLeaseKeepAliveInitTimeout(t *testing.T) {
}
// keep client disconnected
clus.Members[0].Stop(t)
rc := cli.KeepAlive(context.Background(), resp.ID)
rc, kerr := cli.KeepAlive(context.Background(), resp.ID)
if kerr != nil {
t.Fatal(kerr)
}
select {
case ka, ok := <-rc:
if ok {
Expand Down Expand Up @@ -436,7 +454,10 @@ func TestLeaseKeepAliveTTLTimeout(t *testing.T) {
if err != nil {
t.Fatal(err)
}
rc := cli.KeepAlive(context.Background(), resp.ID)
rc, kerr := cli.KeepAlive(context.Background(), resp.ID)
if kerr != nil {
t.Fatal(kerr)
}
if kresp := <-rc; kresp.ID != resp.ID {
t.Fatalf("ID = %x, want %x", kresp.ID, resp.ID)
}
Expand Down Expand Up @@ -559,7 +580,10 @@ func TestLeaseRenewLostQuorum(t *testing.T) {

kctx, kcancel := context.WithCancel(context.Background())
defer kcancel()
ka := cli.KeepAlive(kctx, r.ID)
ka, err := cli.KeepAlive(kctx, r.ID)
if err != nil {
t.Fatal(err)
}
// consume first keepalive so next message sends when cluster is down
<-ka
lastKa := time.Now()
Expand Down Expand Up @@ -606,9 +630,9 @@ func TestLeaseKeepAliveLoopExit(t *testing.T) {
}
cli.Close()

ka := cli.KeepAlive(ctx, resp.ID)
if resp, ok := <-ka; ok {
t.Fatalf("expected closed channel, got response %+v", resp)
_, err = cli.KeepAlive(ctx, resp.ID)
if _, ok := err.(clientv3.ErrKeepAliveHalted); !ok {
t.Fatalf("expected %T, got %v(%T)", clientv3.ErrKeepAliveHalted{}, err, err)
}
}

Expand Down Expand Up @@ -683,9 +707,15 @@ func TestLeaseWithRequireLeader(t *testing.T) {
t.Fatal(err2)
}
// kaReqLeader close if the leader is lost
kaReqLeader := c.KeepAlive(clientv3.WithRequireLeader(context.TODO()), lid1.ID)
kaReqLeader, kerr1 := c.KeepAlive(clientv3.WithRequireLeader(context.TODO()), lid1.ID)
if kerr1 != nil {
t.Fatal(kerr1)
}
// kaWait will wait even if the leader is lost
kaWait := c.KeepAlive(context.TODO(), lid2.ID)
kaWait, kerr2 := c.KeepAlive(context.TODO(), lid2.ID)
if kerr2 != nil {
t.Fatal(kerr2)
}

select {
case <-kaReqLeader:
Expand Down
Loading

0 comments on commit 24e85b2

Please sign in to comment.