}
		w = func(options metav1.ListOptions) (watch.Interface, error) {
			return c.{{.ClientGetter}}().{{ .ClientGroupPath }}().{{ .ClientTypePath }}({{if not .Resource.ClusterScoped}}opts.Namespace{{end}}).Watch(context.Background(), options)
		}
	{{- end }}
{{- end }}
  default:
    panic(fmt.Sprintf("Unknown type %v", g))
	}

	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

	}

	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.

				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

	// 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 {

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

      "storageVersionHash": "l/iqIbDgFyQ=",
      "verbs": [
        "create",
        "delete",
        "deletecollection",
        "get",
        "list",
        "patch",
        "update",
        "watch"
      ]
    },
    {
      "kind": "Ingress",
      "name": "ingresses",
      "namespaced": true,
      "shortNames": [
        "ing"
      ],
      "singularName": "ingress",

        },
        {
          "$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.",