<commit_before>package subscriber

import (
	"sync"
	"sync/atomic"

	"golang.org/x/sync/syncmap"
)

// Manager ...
type Manager struct {
	s syncmap.Map
	sync.Mutex
	sync.WaitGroup
	count int64
}

// Subscribe ...
func (m *Manager) Subscribe(key string) chan interface{} {
	m.Lock()
	defer m.Unlock()

	// WaitGroup
	m.Add(1)

	// Counter
	atomic.AddInt64(&m.count, 1)

	r, _ := m.s.LoadOrStore(key, make(chan interface{}, 0))
	return r.(chan interface{})
}

// Subscribed ...
func (m *Manager) Subscribed(key string) bool {
	_, ok := m.s.Load(key)
	return ok
}

// Unsubscribe ...
func (m *Manager) Unsubscribe(key string) {
	m.Lock()
	defer m.Unlock()

	chInterface, ok := m.s.Load(key)
	if !ok {
		return
	}

	// WaitGroup
	m.Done()

	// Counter
	atomic.AddInt64(&m.count, -1)

	ch := chInterface.(chan interface{})
	close(ch)
	m.s.Delete(key)
}

// Clear the list
func (m *Manager) Clear() {
	m.s.Range(func(k, v interface{}) bool {
		m.Unsubscribe(k.(string))
		return true
	})
}

// Subscriptions ...
func (m *Manager) Subscriptions() []string {
	m.Lock()
	defer m.Unlock()
<commit_msg>postpone waitgroup.done() until the other cleanup has been done<commit_after>package subscriber

import (
	"sync"
	"sync/atomic"

	"golang.org/x/sync/syncmap"
)

// Manager ...
type Manager struct {
	s syncmap.Map
	sync.Mutex
	sync.WaitGroup
	count int64
}

// Subscribe ...
func (m *Manager) Subscribe(key string) chan interface{} {
	m.Lock()
	defer m.Unlock()

	// WaitGroup
	m.Add(1)

	// Counter
	atomic.AddInt64(&m.count, 1)

	r, _ := m.s.LoadOrStore(key, make(chan interface{}, 0))
	return r.(chan interface{})
}

// Subscribed ...
func (m *Manager) Subscribed(key string) bool {
	_, ok := m.s.Load(key)
	return ok
}

// Unsubscribe ...
func (m *Manager) Unsubscribe(key string) {
	m.Lock()
	defer m.Unlock()

	chInterface, ok := m.s.Load(key)
	if !ok {
		return
	}

	// WaitGroup
	defer m.Done()

	// Counter
	atomic.AddInt64(&m.count, -1)

	ch := chInterface.(chan interface{})
	close(ch)
	m.s.Delete(key)
}

// Clear the list
func (m *Manager) Clear() {
	m.s.Range(func(k, v interface{}) bool {
		m.Unsubscribe(k.(string))
		return true
	})
}

// Subscriptions ...
func (m *Manager) Subscriptions() []string {
	m.Lock()
	defer m.Unlock()

{'fraction_non_alphanumeric': 0.17649687220732796, 'fraction_numerical': 0.008936550491510277, 'mean_word_length': 3.200750469043152, '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': '3375379', 'n_tokens_mistral': 902, 'n_tokens_neox': 864, 'n_words': 207}