Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

v3rpc: implement 'prev-kv' watch

Browse Source
This commit is contained in:
Gyu-Ho Lee 2016-10-07 14:22:19 -07:00
parent 6f89fbf8b5
commit 2862c4fa12
1 changed files with 32 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
etcdserver/api/v3rpc/watch.go
View File

@ -32,7 +32,7 @@ type watchServer struct {
clusterID int64 clusterID int64
memberID int64 memberID int64
raftTimer etcdserver.RaftTimer raftTimer etcdserver.RaftTimer
watchable mvcc.Watchable watchable mvcc.WatchableKV
} }
func NewWatchServer(s *etcdserver.EtcdServer) pb.WatchServer { func NewWatchServer(s *etcdserver.EtcdServer) pb.WatchServer {
@ -82,6 +82,8 @@ type serverWatchStream struct {
memberID int64 memberID int64
raftTimer etcdserver.RaftTimer raftTimer etcdserver.RaftTimer
watchable mvcc.WatchableKV
gRPCStream pb.Watch_WatchServer gRPCStream pb.Watch_WatchServer
watchStream mvcc.WatchStream watchStream mvcc.WatchStream
ctrlStream chan *pb.WatchResponse ctrlStream chan *pb.WatchResponse
@ -91,6 +93,7 @@ type serverWatchStream struct {
// progress tracks the watchID that stream might need to send // progress tracks the watchID that stream might need to send
// progress to. // progress to.
progress map[mvcc.WatchID]bool progress map[mvcc.WatchID]bool
prevKV map[mvcc.WatchID]bool
// closec indicates the stream is closed. // closec indicates the stream is closed.
closec chan struct{} closec chan struct{}
@ -104,11 +107,15 @@ func (ws *watchServer) Watch(stream pb.Watch_WatchServer) (err error) {
clusterID: ws.clusterID, clusterID: ws.clusterID,
memberID: ws.memberID, memberID: ws.memberID,
raftTimer: ws.raftTimer, raftTimer: ws.raftTimer,
watchable: ws.watchable,
gRPCStream: stream, gRPCStream: stream,
watchStream: ws.watchable.NewWatchStream(), watchStream: ws.watchable.NewWatchStream(),
// chan for sending control response like watcher created and canceled. // chan for sending control response like watcher created and canceled.
ctrlStream: make(chan *pb.WatchResponse, ctrlStreamBufLen), ctrlStream: make(chan *pb.WatchResponse, ctrlStreamBufLen),
progress: make(map[mvcc.WatchID]bool), progress: make(map[mvcc.WatchID]bool),
prevKV: make(map[mvcc.WatchID]bool),
closec: make(chan struct{}), closec: make(chan struct{}),
} }
@ -170,11 +177,18 @@ func (sws *serverWatchStream) recvLoop() error {
rev = wsrev + 1 rev = wsrev + 1
} }
id := sws.watchStream.Watch(creq.Key, creq.RangeEnd, rev) id := sws.watchStream.Watch(creq.Key, creq.RangeEnd, rev)
if id != -1 && creq.ProgressNotify { if id != -1 {
if creq.ProgressNotify {
sws.mu.Lock() sws.mu.Lock()
sws.progress[id] = true sws.progress[id] = true
sws.mu.Unlock() sws.mu.Unlock()
} }
if creq.PrevKv {
sws.mu.Lock()
sws.prevKV[id] = true
sws.mu.Unlock()
}
}
wr := &pb.WatchResponse{ wr := &pb.WatchResponse{
Header: sws.newResponseHeader(wsrev), Header: sws.newResponseHeader(wsrev),
WatchId: int64(id), WatchId: int64(id),
@ -198,6 +212,7 @@ func (sws *serverWatchStream) recvLoop() error {
} }
sws.mu.Lock() sws.mu.Lock()
delete(sws.progress, mvcc.WatchID(id)) delete(sws.progress, mvcc.WatchID(id))
delete(sws.prevKV, mvcc.WatchID(id))
sws.mu.Unlock() sws.mu.Unlock()
} }
} }
@ -244,8 +259,17 @@ func (sws *serverWatchStream) sendLoop() {
// or define protocol buffer with []mvccpb.Event. // or define protocol buffer with []mvccpb.Event.
evs := wresp.Events evs := wresp.Events
events := make([]*mvccpb.Event, len(evs)) events := make([]*mvccpb.Event, len(evs))
needPrevKV := sws.prevKV[wresp.WatchID]
for i := range evs { for i := range evs {
events[i] = &evs[i] events[i] = &evs[i]
if needPrevKV {
opt := mvcc.RangeOptions{Rev: evs[i].Kv.ModRevision - 1}
r, err := sws.watchable.Range(evs[i].Kv.Key, nil, opt)
if err == nil && len(r.KVs) != 0 {
events[i].PrevKv = &(r.KVs[0])
}
}
} }
wr := &pb.WatchResponse{ wr := &pb.WatchResponse{