Repository: kubernetes/kubernetes
File: staging/src/k8s.io/client-go/testing/fixture.go (Go)

Code before:
			}
			addFromRV = rv
		}
	}

	t.lock.Lock()
	defer t.lock.Unlock()

	fakewatcher := watch.NewRaceFreeFake()

	if _, exists := t.watchers[gvr]; !exists {
		t.watchers[gvr] = make(map[string][]*watch.RaceFreeFakeWatcher)
	}
	t.watchers[gvr][ns] = append(t.watchers[gvr][ns], fakewatcher)

	// Deliver all objects that match the list options, for example
	// between the initial List and the following Watch.
	if addExisting {
		objs := t.objects[gvr]
		matchingObjs, err := filterByNamespace(objs, ns)
		if err != nil {
			return nil, err
		}
		for _, obj := range matchingObjs {
			if addFromRV < obj.resourceVersion {
				fakewatcher.Add(obj.Object)
			}
		}
	}

	return fakewatcher, nil
}

func (t *tracker) Get(gvr schema.GroupVersionResource, ns, name string, opts ...metav1.GetOptions) (runtime.Object, error) {
	_, err := assertOptionalSingleArgument(opts)
	if err != nil {
		return nil, err
	}
	errNotFound := apierrors.NewNotFound(gvr.GroupResource(), name)

	t.lock.RLock()
	defer t.lock.RUnlock()

	objs, ok := t.objects[gvr]
	if !ok {
		return nil, errNotFound
	}

	matchingObj, ok := objs[types.NamespacedName{Namespace: ns, Name: name}]
	if !ok {
		return nil, errNotFound

Review comment (suggestion):
If opts.SendInitialEvents is true, do we also need to inject a `"k8s.io/initial-events-end": "true"` bookmark after sending the initial events?

```
	// The bookmark will report the ResourceVersion (RV) corresponding to the
	// set of objects, and be marked with `"k8s.io/initial-events-end": "true"` annotation.
	// Afterwards, the watch stream will proceed as usual, sending watch events
	// corresponding to changes (subsequent to the RV) to objects watched.
```

Code after:
		// The default if ResourceVersion is empty is "start at most recent",
		// which includes delivering all existing objects. addFromRV == 0
		// matches all objects below because all stored objects have addFromRV > 0.
		addExisting = true
		if opts[0].ResourceVersion != "" {
			rv, err := strconv.ParseInt(opts[0].ResourceVersion, 10, 64)
			if err != nil {
				return nil, fmt.Errorf("invalid ResourceVersion %q in ListOptions, must be int64: %w", opts[0].ResourceVersion, err)
			}
			addFromRV = rv
		}
	}

	t.lock.Lock()
	defer t.lock.Unlock()

	fakewatcher := watch.NewRaceFreeFake()

	if _, exists := t.watchers[gvr]; !exists {
		t.watchers[gvr] = make(map[string][]*watch.RaceFreeFakeWatcher)
	}
	t.watchers[gvr][ns] = append(t.watchers[gvr][ns], fakewatcher)

	// Deliver all objects that match the list options, for example
	// between the initial List and the following Watch.
	if addExisting {
		objs := t.objects[gvr]
		matchingObjs, err := filterByNamespace(objs, ns)
		if err != nil {
			return nil, err
		}
		for _, obj := range matchingObjs {
			if addFromRV < obj.resourceVersion {
				fakewatcher.Add(obj.Object)
			}
		}
	}

	return fakewatcher, nil
}

func (t *tracker) Get(gvr schema.GroupVersionResource, ns, name string, opts ...metav1.GetOptions) (runtime.Object, error) {
	_, err := assertOptionalSingleArgument(opts)
	if err != nil {
		return nil, err
	}
	errNotFound := apierrors.NewNotFound(gvr.GroupResource(), name)

	t.lock.RLock()
	defer t.lock.RUnlock()