scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-05 16:15:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Requested changes.

Browse source 
Signed-off-by: Antonio Navarro Perez <antnavper@gmail.com>
This commit is contained in:
Antonio Navarro Perez 2021-03-05 18:40:00 +01:00
parent 85a43c519d
commit feea0fa347
1 changed files with 19 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
etcd/leasedkv.go
View file

@ -55,34 +55,38 @@ func (l *leasedKV) Start(initValue string) error {
l.mu.Lock()
defer l.mu.Unlock()
return l.create(initValue)
kaChann, err := l.create(l.value)
if err != nil {
return err
}
go l.consumeLease(kaChann)
return nil
}
func (l *leasedKV) create(initValue string) error {
func (l *leasedKV) create(initValue string) (<-chan *clientv3.LeaseKeepAliveResponse, error) {
ctx, cancel := context.WithCancel(context.Background())
l.cancel = cancel
leaseResp, err := l.cli.Grant(ctx, l.ttlSeconds)
if err != nil {
return errors.Wrap(err, "creating a lease")
return nil, errors.Wrap(err, "creating a lease")
}
if _, err := l.cli.Txn(ctx).
Then(clientv3.OpPut(l.key, initValue, clientv3.WithLease(leaseResp.ID))).
Commit(); err != nil {
return errors.Wrapf(err, "creating key %s with value [%s]", l.key, initValue)
return nil, errors.Wrapf(err, "creating key %s with value [%s]", l.key, initValue)
}
kaChann, err := l.cli.KeepAlive(ctx, leaseResp.ID)
if err != nil {
return errors.Wrapf(err, "keeping alive the lease for the key %s with value %s", l.key, l.value)
return nil, errors.Wrapf(err, "keeping alive the lease for the key %s with value %s", l.key, l.value)
}
l.value = initValue
go l.consumeLease(kaChann)
return nil
return kaChann, nil
}
func (l *leasedKV) consumeLease(ch <-chan *clientv3.LeaseKeepAliveResponse) {
@ -97,7 +101,13 @@ func (l *leasedKV) consumeLease(ch <-chan *clientv3.LeaseKeepAliveResponse) {
}
if ok := retry(1*time.Second, func() error {
return l.create(l.value)
kaChann, err := l.create(l.value)
if err != nil {
return err
}
go l.consumeLease(kaChann)
return nil
}); !ok {
log.Println("lease cannot be recreated. Key:", l.key)
l.mu.Unlock()
@ -108,10 +118,6 @@ func (l *leasedKV) consumeLease(ch <-chan *clientv3.LeaseKeepAliveResponse) {
l.mu.Unlock()
return
}
if kresp == nil {
continue
}
}
}