<commit_before>/*
The EventBus allows publish-subscribe-style communication between
components without requiring the components to explicitly register with one
another 
*/
package eventbus

import (
	"fmt"
	"reflect"
)

type EventHandler interface {
	OnEvent(evt Event)
}

type Event interface {
	Event() string
}

type EventBus struct {
	locks    *SegmentedRWLock
	handlers map[string]map[EventHandler]None
	async    bool
}

type None struct{}

func (self *EventBus) Unsubscribe(evt Event, handler EventHandler) {
	eventType := resolveType(evt)
	locker := self.locks.locker(eventType)
	locker.Lock()
	defer locker.Unlock()

	handlers, ok := self.handlers[eventType]
	if ok {
		delete(handlers, handler)
	} else {
		panic(fmt.Sprint("the event '%s' doesn't exist", eventType))
	}
}

func (self *EventBus) Subscribe(evt Event, handler EventHandler) {
	eventType := resolveType(evt)
	locker := self.locks.locker(eventType)
	locker.Lock()
	defer locker.Unlock()

	handlers, ok := self.handlers[eventType]
	if ok {
		handlers[handler] = None{}
	} else {
		handlers := make(map[EventHandler]None)
		handlers[handler] = None{}
		self.handlers[eventType] = handlers
	}
}

func (self *EventBus) Publish(evt Event) {
	eventType := resolveType(evt)
	locker := self.locks.locker(eventType)
	locker.RLock()
	defer locker.RUnlock()

	handlers, ok := self.handlers[eventType]
	if ok {
		for handler, _ := range handlers {
			self.dispatch(evt, handler)
		}
	}
}

func (self *EventBus) dispatch(evt Event, handler EventHandler) {
	if self.async {
		go handler.OnEvent(evt)
	} else {
		handler.OnEvent(evt)
	}
}

func resolveType(evt Event) string {
	subEventType := evt.Event()
	if subEventType != "" {
		return reflect.TypeOf(evt).String() + "." + subEventType
	}
	return reflect.TypeOf(evt).String()
}

type SegmentedRWLock struct {
	segments int
	locks    []sync.RWMutex
}

func NewSegmentedRWLock(segments int) *SegmentedRWLock {
	return &SegmentedRWLock{
		segments: segments,
		locks:    make([]sync.RWMutex, segments),
	}
}

func (self *SegmentedRWLock) locker(key string) *sync.RWMutex {
	hash := abs(hash([]byte(key)))
	return &self.locks[hash%self.segments]
}

func abs(x int) int {
	if x < 0 {
		return x * -1
	}
	return x
}

func hash(bytes []byte) int {
	var h int
	for i := 0; i < len(bytes); i++ {
		h = 31*h ^ int(bytes[i])
	}
	return h
}

type EventChannel struct {
	C chan Event
}

func (self *EventChannel) OnEvent(evt Event) {
	self.C <- evt
}

var DefaultEventBus = &EventBus{
	locks:    NewSegmentedRWLock(32),
	handlers: make(map[string]map[EventHandler]None),
	async:    true,
}

func Unsubscribe(evt Event, handler EventHandler) {
	DefaultEventBus.Unsubscribe(evt, handler)
}

func Subscribe(evt Event, handler EventHandler) {
	DefaultEventBus.Subscribe(evt, handler)
}

func Publish(evt Event) {
	DefaultEventBus.Publish(evt)
}

func NewEventChannel() *EventChannel {
	return &EventChannel{
		C: make(chan Event),
	}
}
<commit_msg>add shortcut form buffered channel<commit_after>/*
The EventBus allows publish-subscribe-style communication between
components without requiring the components to explicitly register with one
another 
*/
package eventbus

import (
	"fmt"
	"reflect"
)

type EventHandler interface {
	OnEvent(evt Event)
}

type Event interface {
	Event() string
}

type EventChannel struct {
	C chan Event
}

func (self *EventChannel) OnEvent(evt Event) {
	self.C <- evt
}

type EventBus struct {
	locks    *SegmentedRWLock
	handlers map[string]map[EventHandler]None
	async    bool
}

type None struct{}

func (self *EventBus) Unsubscribe(evt Event, handler EventHandler) {
	eventType := resolveType(evt)
	locker := self.locks.locker(eventType)
	locker.Lock()
	defer locker.Unlock()

	handlers, ok := self.handlers[eventType]
	if ok {
		delete(handlers, handler)
	} else {
		panic(fmt.Sprint("the event '%s' doesn't exist", eventType))
	}
}

func (self *EventBus) Subscribe(evt Event, handler EventHandler) {
	eventType := resolveType(evt)
	locker := self.locks.locker(eventType)
	locker.Lock()
	defer locker.Unlock()

	handlers, ok := self.handlers[eventType]
	if ok {
		handlers[handler] = None{}
	} else {
		handlers := make(map[EventHandler]None)
		handlers[handler] = None{}
		self.handlers[eventType] = handlers
	}
}

func (self *EventBus) Publish(evt Event) {
	eventType := resolveType(evt)
	locker := self.locks.locker(eventType)
	locker.RLock()
	defer locker.RUnlock()

	handlers, ok := self.handlers[eventType]
	if ok {
		for handler, _ := range handlers {
			self.dispatch(evt, handler)
		}
	}
}

func (self *EventBus) dispatch(evt Event, handler EventHandler) {
	if self.async {
		go handler.OnEvent(evt)
	} else {
		handler.OnEvent(evt)
	}
}

func resolveType(evt Event) string {
	subEventType := evt.Event()
	if subEventType != "" {
		return reflect.TypeOf(evt).String() + "." + subEventType
	}
	return reflect.TypeOf(evt).String()
}

type SegmentedRWLock struct {
	segments int
	locks    []sync.RWMutex
}

func NewSegmentedRWLock(segments int) *SegmentedRWLock {
	return &SegmentedRWLock{
		segments: segments,
		locks:    make([]sync.RWMutex, segments),
	}
}

func (self *SegmentedRWLock) locker(key string) *sync.RWMutex {
	hash := abs(hash([]byte(key)))
	return &self.locks[hash%self.segments]
}

func abs(x int) int {
	if x < 0 {
		return x * -1
	}
	return x
}

func hash(bytes []byte) int {
	var h int
	for i := 0; i < len(bytes); i++ {
		h = 31*h ^ int(bytes[i])
	}
	return h
}

var DefaultEventBus = &EventBus{
	locks:    NewSegmentedRWLock(32),
	handlers: make(map[string]map[EventHandler]None),
	async:    true,
}

func Unsubscribe(evt Event, handler EventHandler) {
	DefaultEventBus.Unsubscribe(evt, handler)
}

func Subscribe(evt Event, handler EventHandler) {
	DefaultEventBus.Subscribe(evt, handler)
}

func Publish(evt Event) {
	DefaultEventBus.Publish(evt)
}

func NewEventChannel() *EventChannel {
	return &EventChannel{
		C: make(chan Event),
	}
}

func NewBufferedEventChannel(size int) *EventChannel {
	return &EventChannel{
		C: make(chan Event, size),
	}
}

{'fraction_non_alphanumeric': 0.1264539714190761, 'fraction_numerical': 0.002326354270521768, 'mean_word_length': 4.527089072543618, 'pattern_counts': {'":': 0, '<': 9, '<?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': '2697518', 'n_tokens_mistral': 2191, 'n_tokens_neox': 2163, 'n_words': 526}