<commit_before>package collectors

import (
	"reflect"
	"runtime"
	"sync"
	"time"

	"bosun.org/collect"
	"bosun.org/metadata"
	"bosun.org/opentsdb"
)

type StreamCollector struct {
	F      func() <-chan *opentsdb.MultiDataPoint
	Enable func() bool
	name   string
	init   func()

	// internal use
	sync.Mutex
	enabled bool
}

func (s *StreamCollector) Init() {
	if s.init != nil {
		s.init()
	}
}

func (s *StreamCollector) Run(dpchan chan<- *opentsdb.DataPoint) {
	if s.Enable != nil {
		go func() {
			for {
				next := time.After(time.Minute * 5)
				s.Lock()
				s.enabled = s.Enable()
				s.Unlock()
				<-next
			}
		}()
	}

	inputChan := s.F()
	count := 0
	for md := range inputChan {
		if s.Enabled() {
			for _, dp := range *md {
				dpchan <- dp
				count++
			}

			if !collect.DisableDefaultCollectors {
				tags := opentsdb.TagSet{"collector": s.Name(), "os": runtime.GOOS}
				Add(md, "scollector.collector.count", count, tags, metadata.Counter, metadata.Count, "Counter of metrics passed through.")
			}
		}
	}
}

func (s *StreamCollector) Enabled() bool {
	if s.Enable == nil {
		return true
	}
	s.Lock()
	defer s.Unlock()
	return s.enabled
}

func (s *StreamCollector) Name() string {
	if s.name != "" {
		return s.name
	}
	v := runtime.FuncForPC(reflect.ValueOf(s.F).Pointer())
	return v.Name()

}
<commit_msg>Adding TagOverride to stream collector.<commit_after>package collectors

import (
	"reflect"
	"runtime"
	"sync"
	"time"

	"bosun.org/collect"
	"bosun.org/metadata"
	"bosun.org/opentsdb"
)

type StreamCollector struct {
	F      func() <-chan *opentsdb.MultiDataPoint
	Enable func() bool
	name   string
	init   func()

	// internal use
	sync.Mutex
	enabled bool

	TagOverride
}

func (s *StreamCollector) Init() {
	if s.init != nil {
		s.init()
	}
}

func (s *StreamCollector) Run(dpchan chan<- *opentsdb.DataPoint) {
	if s.Enable != nil {
		go func() {
			for {
				next := time.After(time.Minute * 5)
				s.Lock()
				s.enabled = s.Enable()
				s.Unlock()
				<-next
			}
		}()
	}

	inputChan := s.F()
	count := 0
	for md := range inputChan {
		if s.Enabled() {
			for _, dp := range *md {
				dpchan <- dp
				count++
			}

			if !collect.DisableDefaultCollectors {
				tags := opentsdb.TagSet{"collector": s.Name(), "os": runtime.GOOS}
				Add(md, "scollector.collector.count", count, tags, metadata.Counter, metadata.Count, "Counter of metrics passed through.")
			}
		}
	}
}

func (s *StreamCollector) Enabled() bool {
	if s.Enable == nil {
		return true
	}
	s.Lock()
	defer s.Unlock()
	return s.enabled
}

func (s *StreamCollector) Name() string {
	if s.name != "" {
		return s.name
	}
	v := runtime.FuncForPC(reflect.ValueOf(s.F).Pointer())
	return v.Name()

}

{'fraction_non_alphanumeric': 0.15701559020044542, 'fraction_numerical': 0.0014847809948032665, 'mean_word_length': 2.980797636632201, 'pattern_counts': {'":': 4, '<': 11, '<?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': '911154', 'n_tokens_mistral': 1151, 'n_tokens_neox': 1025, 'n_words': 245}