func AsStructuredWatcher(wf cache.WatchFunc, obj runtime.Object) cache.WatchFunc {
	return func(lo metav1.ListOptions) (watch.Interface, error) {
		uw, err := wf(lo)
		if err != nil {
			return nil, err
		}
		structuredCh := make(chan watch.Event)
		go func() {
			defer close(structuredCh)
			unstructuredCh := uw.ResultChan()
			for {
				select {
				case ue, ok := <-unstructuredCh:
					if !ok {
						return
					}
					unstructuredObj, ok := ue.Object.(*unstructured.Unstructured)
					if !ok {
						structuredCh <- ue
						continue
					}
					structuredObj := obj.DeepCopyObject()
					err := FromUnstructured(unstructuredObj, structuredObj)
					if err != nil {
						structuredCh <- watch.Event{
							Type: watch.Error,
							Object: &metav1.Status{
								Status:  metav1.StatusFailure,
								Code:    http.StatusUnprocessableEntity,
								Reason:  metav1.StatusReasonInvalid,
								Message: err.Error(),
							},
						}
						continue
					}
					structuredCh <- watch.Event{
						Type:   ue.Type,
						Object: structuredObj,
					}
				}
			}
		}()
		return NewProxyWatcher(structuredCh), nil
	}
}
AsStructuredWatcher is public for testing only.
TODO(mattmoor): Move tests for this to `package duck` and make private.
package duck
import (
	"fmt"
	"net/http"
	"time"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/tools/cache"
	"github.com/knative/pkg/apis"
)
type TypedInformerFactory struct {
	Client       dynamic.Interface
	Type         apis.Listable
	ResyncPeriod time.Duration
	StopChannel  <-chan struct{}
}
var _ InformerFactory = (*TypedInformerFactory)(nil)
func (dif *TypedInformerFactory) Get(gvr schema.GroupVersionResource) (cache.SharedIndexInformer, cache.GenericLister, error) {
	listObj := dif.Type.GetListType()
	lw := &cache.ListWatch{
		ListFunc:  asStructuredLister(dif.Client.Resource(gvr).List, listObj),
		WatchFunc: AsStructuredWatcher(dif.Client.Resource(gvr).Watch, dif.Type),
	}
	inf := cache.NewSharedIndexInformer(lw, dif.Type, dif.ResyncPeriod, cache.Indexers{
		cache.NamespaceIndex: cache.MetaNamespaceIndexFunc,
	})
	lister := cache.NewGenericLister(inf.GetIndexer(), gvr.GroupResource())
	go inf.Run(dif.StopChannel)
	if ok := cache.WaitForCacheSync(dif.StopChannel, inf.HasSynced); !ok {
		return nil, nil, fmt.Errorf("Failed starting shared index informer for %v with type %T", gvr, dif.Type)
	}
	return inf, lister, nil
}
type unstructuredLister func(metav1.ListOptions) (*unstructured.UnstructuredList, error)
func asStructuredLister(ulist unstructuredLister, listObj runtime.Object) cache.ListFunc {
	return func(opts metav1.ListOptions) (runtime.Object, error) {
		ul, err := ulist(opts)
		if err != nil {
			return nil, err
		}
		res := listObj.DeepCopyObject()
		if err := FromUnstructured(ul, res); err != nil {
			return nil, err
		}
		return res, nil
	}
}