cs.AddReactor("*", "*", testing.ObjectReaction(o))
	cs.AddWatchReactor("*", func(action testing.Action) (handled bool, ret watch.Interface, err error) {
		gvr := action.GetResource()
		ns := action.GetNamespace()
		watch, err := o.Watch(gvr, ns)
		if err != nil {
			return false, nil, err
		}
		return true, watch, nil
	})

	return cs
}

// Clientset implements clientset.Interface. Meant to be embedded into a

	fs.BoolVar(&s.EnableWatchCache, "watch-cache", s.EnableWatchCache,
		"Enable watch caching in the apiserver")

	fs.IntVar(&s.DefaultWatchCacheSize, "default-watch-cache-size", s.DefaultWatchCacheSize,
		"Default watch cache size. If zero, watch cache will be disabled for resources that do not have a default watch size set.")

	fs.MarkDeprecated("default-watch-cache-size",

	// that we're effectively serializing the whole watch event, but storing
	// it in serializations of the Object within the watch event.
	identifier := watchEncoderIdentifier{
		Name:            "watch",
		EmbeddedEncoder: string(e.embeddedEncoder.Identifier()),
		Encoder:         string(e.encoder.Identifier()),
		EventType:       string(eventType),
	}

	result, err := json.Marshal(identifier)
	if err != nil {

				gvr := action.GetResource()
				ns := action.GetNamespace()
				watch, err := fc.Tracker().Watch(gvr, ns)
				if err != nil {
					return false, nil, err
				}
				// Kubernetes Fake watches do not add initial state, but real server does
				// https://kubernetes.io/docs/reference/using-api/api-concepts/#semantics-for-watch

	}

	m.Shutdown()

	wg.Wait()
}

func TestBroadcasterWatcherClose(t *testing.T) {
	m := NewBroadcaster(0, WaitIfChannelFull)
	w, err := m.Watch()
	if err != nil {
		t.Fatalf("Unable start event watcher: '%v' (will not retry!)", err)
	}
	w2, err := m.Watch()
	if err != nil {
		t.Fatalf("Unable start event watcher: '%v' (will not retry!)", err)
	}
	w.Stop()
	m.Shutdown()

      "storageVersionHash": "6OxvlMmQ6is=",
      "verbs": [
        "create",
        "delete",
        "deletecollection",
        "get",
        "list",
        "patch",
        "update",
        "watch"
      ]
    },
    {
      "kind": "ValidatingAdmissionPolicy",
      "name": "validatingadmissionpolicies/status",
      "namespaced": false,
      "singularName": "",
      "verbs": [
        "get",

		// Add grpc context metadata to watch and progress notify requests done by cacher to:
		// * Prevent starvation of watch opened by cacher, by moving it to separate Watch RPC than watch request that bypass cacher.
		// * Ensure that progress notification requests are executed on the same Watch RPC as their watch, which is required for it to work.

        },
        {
          "$ref": "#/parameters/watch--CFxmTDM"
        }
      ]
    },
    "/api/v1/watch/configmaps": {
      "get": {
        "consumes": [
          "*/*"
        ],
        "description": "watch individual changes to a list of ConfigMap. deprecated: use the 'watch' parameter with a list operation instead.",

	}
	prober.eventsMap[eventDirAbs] = op
}

// Recursively adds to watch all directories inside and including the file specified by the given filename.
// If the file is a symlink to a directory, it will watch the symlink but not any of the subdirectories.
//
// Each file or directory change triggers two events: one from the watch on itself, another from the watch
// on its parent directory.

      "storageVersionHash": "GJVAJZSZBIw=",
      "verbs": [
        "create",
        "delete",
        "deletecollection",
        "get",
        "list",
        "patch",
        "update",
        "watch"
      ]
    },
    {
      "kind": "FlowSchema",
      "name": "flowschemas/status",
      "namespaced": false,
      "singularName": "",
      "verbs": [
        "get",
        "patch",