// Mesos spawns a new pod source that watches API server for changes and collaborates with
// executor.Registry to generate api.Pod objects in a fashion that's very Mesos-aware.
func Mesos(
	stop <-chan struct{},
	out chan<- interface{},
	podWatch *cache.ListWatch,
	registry executor.Registry,
	options ...Option,
) {
	source := &Source{
		stop: stop,
		out:  out,
		filters: []Filter{
			FilterFunc(filterMirrorPod),
			&registeredPodFilter{registry: registry},
		},
	}
	// note: any filters added by options should be applied after the defaults
	for _, opt := range options {
		opt(source)
	}
	// reflect changes from the watch into a chan, filtered to include only mirror pods
	// (have an ConfigMirrorAnnotationKey attr)
	cache.NewReflector(
		podWatch,
		&api.Pod{},
		cache.NewUndeltaStore(source.send, cache.MetaNamespaceKeyFunc),
		0,
	).RunUntil(stop)
}

// newSourceApiserverFromLW holds creates a config source that watches and pulls from the apiserver.
func newSourceApiserverFromLW(lw cache.ListerWatcher, updates chan<- interface{}) {
	send := func(objs []interface{}) {
		var pods []*api.Pod
		for _, o := range objs {
			pods = append(pods, o.(*api.Pod))
		}
		updates <- kubelet.PodUpdate{Pods: pods, Op: kubelet.SET, Source: kubelet.ApiserverSource}
	}
	cache.NewReflector(lw, &api.Pod{}, cache.NewUndeltaStore(send, cache.MetaNamespaceKeyFunc), 0).Run()
}

func newEndpointsSourceApiFromLW(endpointsLW cache.ListerWatcher, period time.Duration, endpointsChan chan<- EndpointsUpdate) {
	endpointsPush := func(objs []interface{}) {
		var endpoints []api.Endpoints
		for _, o := range objs {
			endpoints = append(endpoints, *(o.(*api.Endpoints)))
		}
		endpointsChan <- EndpointsUpdate{Op: SET, Endpoints: endpoints}
	}

	endpointQueue := cache.NewUndeltaStore(endpointsPush, cache.MetaNamespaceKeyFunc)
	cache.NewReflector(endpointsLW, &api.Endpoints{}, endpointQueue, period).Run()
}

func newServicesSourceApiFromLW(servicesLW cache.ListerWatcher, period time.Duration, servicesChan chan<- ServiceUpdate) {
	servicesPush := func(objs []interface{}) {
		var services []api.Service
		for _, o := range objs {
			services = append(services, *(o.(*api.Service)))
		}
		servicesChan <- ServiceUpdate{Op: SET, Services: services}
	}

	serviceQueue := cache.NewUndeltaStore(servicesPush, cache.MetaNamespaceKeyFunc)
	cache.NewReflector(servicesLW, &api.Service{}, serviceQueue, period).Run()
}

// newSourceMesos creates a pod config source that merges pod updates from
// mesos (via execUpdates), and mirror pod updates from the apiserver (via
// podWatch) writing the merged update stream to the out chan. It is expected
// that execUpdates will only ever contain SET operations. The source takes
// ownership of the sourceFinished chan, closing it when the source terminates.
// Source termination happens when the execUpdates chan is closed and fully
// drained of updates.
func newSourceMesos(
	sourceFinished chan struct{},
	execUpdates <-chan kubetypes.PodUpdate,
	out chan<- interface{},
	podWatch *cache.ListWatch,
) {
	source := &sourceMesos{
		sourceFinished: sourceFinished,
		mirrorPods:     make(chan []*api.Pod),
		execUpdates:    execUpdates,
		out:            out,
	}
	// reflect changes from the watch into a chan, filtered to include only mirror pods (have an ConfigMirrorAnnotationKey attr)
	cache.NewReflector(podWatch, &api.Pod{}, cache.NewUndeltaStore(source.send, cache.MetaNamespaceKeyFunc), 0).RunUntil(sourceFinished)
	go source.mergeAndForward()
}

func new(path string, nodeName types.NodeName, period time.Duration, updates chan<- interface{}) *sourceFile {
	send := func(objs []interface{}) {
		var pods []*v1.Pod
		for _, o := range objs {
			pods = append(pods, o.(*v1.Pod))
		}
		updates <- kubetypes.PodUpdate{Pods: pods, Op: kubetypes.SET, Source: kubetypes.FileSource}
	}
	store := cache.NewUndeltaStore(send, cache.MetaNamespaceKeyFunc)
	return &sourceFile{
		path:           path,
		nodeName:       nodeName,
		store:          store,
		fileKeyMapping: map[string]string{},
		updates:        updates,
	}
}

func newSourceMesos(
	stop <-chan struct{},
	out chan<- interface{},
	podWatch *cache.ListWatch,
	registry executor.Registry,
) {
	source := &sourceMesos{
		stop:          stop,
		out:           out,
		registry:      registry,
		priorPodNames: make(map[podName]string),
	}
	// reflect changes from the watch into a chan, filtered to include only mirror pods
	// (have an ConfigMirrorAnnotationKey attr)
	cache.NewReflector(
		podWatch,
		&api.Pod{},
		cache.NewUndeltaStore(source.send, cache.MetaNamespaceKeyFunc),
		0,
	).RunUntil(stop)
}