<commit_before>package main

import (
	"fmt"
	"time"

	"golang.org/x/net/context"
	"k8s.io/kubernetes/pkg/api"
	"k8s.io/kubernetes/pkg/fields"
	"k8s.io/kubernetes/pkg/labels"
	"k8s.io/kubernetes/pkg/watch"
)

type watchFunc func() (watch.Interface, error)

type event struct {
	watch.Event
}

func (e event) String() string {
	m, er := getMeta(e.Object)
	if er != nil {
		return fmt.Sprintf("Event: type=%v object=Unknown", e.Type)
	}
	return fmt.Sprintf(
		"Event: [%v] object={Kind: %q, Name: %q, Namespace: %q} registerable=%v",
		e.Type, m.kind, m.name, m.ns, registerable(e.Object),
	)
}

func acquireWatch(fn watchFunc, out chan<- watch.Interface, c context.Context) {
	retry := 2 * time.Second
	t := time.NewTicker(retry)
	defer t.Stop()

	w, e := fn()
	if e == nil {
		out <- w
		return
	}

	for {
		debugf("Setting watch failed, retry in (%v): %v", retry, e)
		select {
		case <-c.Done():
			return
		case <-t.C:
			w, e := fn()
			if e == nil {
				out <- w
				return
			}
		}
	}
}

func startWatches(c context.Context) (chan event, error) {
	resourceVersion = "0"
	out := make(chan event, 100)
	kc, er := kubeClient()
	if er != nil {
		return out, er
	}
	sv := func() (watch.Interface, error) {
		debugf("Attempting to set watch on Services")
		return kc.Services(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), resourceVersion)
	}
	en := func() (watch.Interface, error) {
		debugf("Attempting to set watch on Endpoints")
		return kc.Endpoints(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), resourceVersion)
	}

	go watcher("Services", sv, out, c)
	go watcher("Endpoints", en, out, c)
	return out, nil
}

func watcher(name string, fn watchFunc, out chan<- event, c context.Context) {
	var w watch.Interface
	var wc = make(chan watch.Interface, 1)
	defer close(wc)

Acquire:
	go acquireWatch(fn, wc, c)
	select {
	case <-c.Done():
		infof("Closing %s watch channel", name)
		return
	case w = <-wc:
		debugf("%s watch set", name)
	}

	for {
		select {
		case <-c.Done():
			infof("Closing %s watch channel", name)
			return
		case e := <-w.ResultChan():
			if isClosed(e) {
				warnf("%s watch closed: %+v", name, e)
				goto Acquire
			}
			out <- event{e}
		}
	}
}

func isClosed(e watch.Event) bool {
	return e.Type == watch.Error || e == watch.Event{}
}<commit_msg>Try and protect against sending on closed channel. Don't reacquire watch on error<commit_after>package main

import (
	"fmt"
	"time"

	"golang.org/x/net/context"
	"k8s.io/kubernetes/pkg/api"
	uapi "k8s.io/kubernetes/pkg/api/unversioned"
	"k8s.io/kubernetes/pkg/fields"
	"k8s.io/kubernetes/pkg/labels"
	"k8s.io/kubernetes/pkg/watch"
)

type watchFunc func() (watch.Interface, error)

type event struct {
	watch.Event
}

func (e event) String() string {
	m, er := getMeta(e.Object)
	if er != nil {
		return fmt.Sprintf("Event: type=%v object=Unknown", e.Type)
	}
	if s, ok := e.Object.(*uapi.Status); ok {
		return fmt.Sprintf("Status: [%s] code=%d %q", s.Status, s.Code, s.Message)
	}
	return fmt.Sprintf(
		"Event: [%v] object={Kind: %q, Name: %q, Namespace: %q} registerable=%v",
		e.Type, m.kind, m.name, m.ns, registerable(e.Object),
	)
}

func startWatches(c context.Context) (chan event, error) {
	resourceVersion = "0"
	out := make(chan event, 100)
	kc, er := kubeClient()
	if er != nil {
		return out, er
	}
	sv := func() (watch.Interface, error) {
		debugf("Attempting to set watch on Services")
		return kc.Services(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), resourceVersion)
	}
	en := func() (watch.Interface, error) {
		debugf("Attempting to set watch on Endpoints")
		return kc.Endpoints(api.NamespaceAll).Watch(labels.Everything(), fields.Everything(), resourceVersion)
	}

	go watcher("Services", sv, out, c)
	go watcher("Endpoints", en, out, c)
	return out, nil
}

func acquireWatch(fn watchFunc, out chan<- watch.Interface, c context.Context) {
	retry := 2 * time.Second
	t := time.NewTicker(retry)
	defer t.Stop()

	w, e := fn()
	if e == nil && c.Err() == nil {
		out <- w
		return
	}

	for {
		debugf("Setting watch failed, retry in (%v): %v", retry, e)
		select {
		case <-c.Done():
			return
		case <-t.C:
			w, e := fn()
			if e == nil && c.Err() == nil {
				out <- w
				return
			}
		}
	}
}

func watcher(name string, fn watchFunc, out chan<- event, c context.Context) {
	var w watch.Interface
	var wc = make(chan watch.Interface, 1)
	defer close(wc)

Acquire:
	go acquireWatch(fn, wc, c)
	select {
	case <-c.Done():
		infof("Closing %s watch channel", name)
		return
	case w = <-wc:
		debugf("%s watch set", name)
	}

EventLoop:
	for {
		select {
		case <-c.Done():
			infof("Closing %s watch channel", name)
			return
		case e := event{<-w.ResultChan()}:
			switch {
			case isClosed(e):
				warnf("%s watch closed: %v", name, e)
				goto Acquire
			case isError(e):
				errorf("%s watch error: %v", name, e)
				goto EventLoop
			case c.Err() == nil:
				out <- e
			}
		}
	}
}

func isClosed(e event) bool {
	return e.Event == watch.Event{}
}

func isError(e event) bool {
	return e.Type == watch.Error
}
{'fraction_non_alphanumeric': 0.1660422007493591, 'fraction_numerical': 0.004141195030565963, 'mean_word_length': 3.472663139329806, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6469040', 'n_tokens_mistral': 2105, 'n_tokens_neox': 1910, 'n_words': 525}