cache: make cache progress-aware

Signed-off-by: Peter Chang <[email protected]>

Author: apullo777

cache/cache.go

@@ -147,7 +147,7 @@ func (c *Cache) Watch(ctx context.Context, key string, opts ...clientv3.OpOption
 }
 
 func (c *Cache) Get(ctx context.Context, key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error) {
-	if c.store.LatestRev() == 0 {
+	if c.store.AppliedRev() == 0 {
 		if err := c.WaitReady(ctx); err != nil {
 			return nil, err
 		}
@@ -186,7 +186,10 @@ func (c *Cache) WaitReady(ctx context.Context) error {
 
 func (c *Cache) WaitForRevision(ctx context.Context, rev int64) error {
 	for {
-		if c.store.LatestRev() >= rev {
+		if c.store.AppliedRev() >= rev {
+			return nil
+		}
+		if c.demux.ObservedRev() >= rev {
 			return nil
 		}
 		select {
@@ -313,7 +316,7 @@ func (c *Cache) watchEvents(watchCh clientv3.WatchChan, applyErr <-chan error, r
 				}
 				return err
 			}
-			c.demux.Broadcast(resp.Events)
+			c.demux.Broadcast(resp)
 		case err := <-applyErr:
 			c.ready.Reset()
 			c.demux.Purge()