<commit_before>package logtap

// Overview

import (
	"github.com/pagodabox/golang-hatchet"
	"reflect"
	"time"
)

type Collector interface {
	CollectChan() chan Message
	SetLogger(hatchet.Logger)
}

type Drain interface {
	Write(Message)
	SetLogger(hatchet.Logger)
}

type Message struct {
	Type     string
	Time     time.Time
	Priority int
	Content  string
}

type Logtap struct {
	log        hatchet.Logger
	Collectors map[string]Collector
	Drains     map[string]Drain
}

// Establishes a new logtap object
// and makes sure it has the some logger
func New(log hatchet.Logger) *Logtap {
	if log == nil {
		log = hatchet.DevNullLogger{}
	}
	return &Logtap{
		log:        log,
		Collectors: make(map[string]Collector),
		Drains:     make(map[string]Drain),
	}
}

// AddDrain addes a drain to the listeners and sets its logger
func (l *Logtap) AddDrain(tag string, d Drain) {
	d.SetLogger(l.log)
	l.Drains[tag] = d
}

// RemoveDrain drops a drain
func (l *Logtap) RemoveDrain(tag string) {
	delete(l.Drains, tag)
}

// AddCollector adds a collector and begins listening to it
// also adds logging
func (l *Logtap) AddCollector(tag string, c Collector) {
	c.SetLogger(l.log)
	l.Collectors[tag] = c
}

// RemoveCollector remove given collector
func (l *Logtap) RemoveCollector(tag string) {
	delete(l.Collectors, tag)
}

// Start begins listening to all the collectors that are registered.
// it then broadcasts all messages to all the drains that are registered
// this is backgrounded so it can be used by a parent process without getting in the way
func (l *Logtap) Start() {
	go func() {
		for {
			cases := make([]reflect.SelectCase, len(l.Collectors))
			index := 0
			for _, col := range l.Collectors {
				cases[index] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(col.CollectChan())}
				index++
			}

			_, value, ok := reflect.Select(cases)
			// ok will be true if the channel has not been closed.
			if ok {
				l.log.Debug("[LOGTAP][start][collect] %v", value.Interface().(Message))
				l.writeMessage(value.Interface().(Message))
			}

		}

	}()
}

// writeMessage broadcasts to all drains in seperate go routines
func (l *Logtap) writeMessage(msg Message) {
	for _, drain := range l.Drains {
		go drain.Write(msg)
	}
}
<commit_msg>add in a way of publishing message internally not just from a collector<commit_after>package logtap

// Overview

import (
	"github.com/pagodabox/golang-hatchet"
	"reflect"
	"time"
)

type Collector interface {
	CollectChan() chan Message
	SetLogger(hatchet.Logger)
}

type Drain interface {
	Write(Message)
	SetLogger(hatchet.Logger)
}

type Message struct {
	Type     string
	Time     time.Time
	Priority int
	Content  string
}

type Logtap struct {
	log        hatchet.Logger
	Collectors map[string]Collector
	Drains     map[string]Drain
}

// Establishes a new logtap object
// and makes sure it has the some logger
func New(log hatchet.Logger) *Logtap {
	if log == nil {
		log = hatchet.DevNullLogger{}
	}
	return &Logtap{
		log:        log,
		Collectors: make(map[string]Collector),
		Drains:     make(map[string]Drain),
	}
}

// AddDrain addes a drain to the listeners and sets its logger
func (l *Logtap) AddDrain(tag string, d Drain) {
	d.SetLogger(l.log)
	l.Drains[tag] = d
}

// RemoveDrain drops a drain
func (l *Logtap) RemoveDrain(tag string) {
	delete(l.Drains, tag)
}

// AddCollector adds a collector and begins listening to it
// also adds logging
func (l *Logtap) AddCollector(tag string, c Collector) {
	c.SetLogger(l.log)
	l.Collectors[tag] = c
}

// RemoveCollector remove given collector
func (l *Logtap) RemoveCollector(tag string) {
	delete(l.Collectors, tag)
}

// Start begins listening to all the collectors that are registered.
// it then broadcasts all messages to all the drains that are registered
// this is backgrounded so it can be used by a parent process without getting in the way
func (l *Logtap) Start() {
	go func() {
		for {
			cases := make([]reflect.SelectCase, len(l.Collectors))
			index := 0
			for _, col := range l.Collectors {
				cases[index] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(col.CollectChan())}
				index++
			}

			_, value, ok := reflect.Select(cases)
			// ok will be true if the channel has not been closed.
			if ok {
				l.log.Debug("[LOGTAP][start][collect] %v", value.Interface().(Message))
				l.writeMessage(value.Interface().(Message))
			}

		}

	}()
}

func (l *Logtap) Publish(category, content string) {
	m := Message{
		Type:     category,
		Time:     time.Now(),
		Priority: 1,
		Content:  content,
	}
	l.writeMessage(m)
}

// writeMessage broadcasts to all drains in seperate go routines
func (l *Logtap) writeMessage(msg Message) {
	for _, drain := range l.Drains {
		go drain.Write(msg)
	}
}

{'fraction_non_alphanumeric': 0.11374606505771248, 'fraction_numerical': 0.0006295907660020986, 'mean_word_length': 3.595949855351977, 'pattern_counts': {'":': 0, '<': 3, '<?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': '2386046', 'n_tokens_mistral': 1729, 'n_tokens_neox': 1589, 'n_words': 499}