code: func (wc *watchChan) startWatching(watchClosedCh chan struct{}) {
  	if wc.initialRev == 0 {
  		if err := wc.sync(); err != nil {
  			klog.Errorf("failed to sync with latest state: %v", err)
  			wc.sendError(err)
  			return
  		}
  	}
  	opts := []clientv3.OpOption{clientv3.WithRev(wc.initialRev + 1), clientv3.WithPrevKV()}
  	if wc.recursive {
  		opts = append(opts, clientv3.WithPrefix())
  	}
  	wch := wc.watcher.client.Watch(wc.ctx, wc.key, opts...)
  	for wres := range wch {
  		if wres.Err() != nil {
  			err := wres.Err()
  			// If there is an error on server (e.g. compaction), the channel will return it before closed.
  			klog.Errorf("watch chan error: %v", err)
  			wc.sendError(err)
  			return
  		}
  		for _, e := range wres.Events {
  			parsedEvent, err := parseEvent(e)
  			if err != nil {
  				klog.Errorf("watch chan error: %v", err)
  				wc.sendError(err)
  				return
  			}
  			wc.sendEvent(parsedEvent)
  		}
  	}
  	// When we come to this point, it's only possible that client side ends the watch.
  	// e.g. cancel the context, close the client.
  	// If this watch chan is broken and context isn't cancelled, other goroutines will still hang.
  	// We should notify the main thread that this goroutine has exited.
  	close(watchClosedCh)
  }
docs: // startWatching does:
  // - get current objects if initialRev=0; set initialRev to current rev
  // - watch on given key and send events to process.