Fix ticker leak and improve readme (#6)

* Add MIT License and enhance README with usage details; refactor event dispatcher for improved subscription handling

* Refactor consumer's Listen method to optimize queue handling and improve readability

* Add defer statement to stop ticker in group.Process method

* Update default_test.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: kelindar

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2025 Roman Atachiants
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -7,6 +7,26 @@
 <a href="https://coveralls.io/github/kelindar/event"><img src="https://coveralls.io/repos/github/kelindar/event/badge.svg" alt="Coverage"></a>
 </p>
 
+## Fast, In-Process Event Dispatcher
+
+This package offers a high-performance, **in-process event dispatcher** for Go, ideal for decoupling modules and enabling asynchronous event handling. It supports both synchronous and asynchronous processing, focusing on speed and simplicity.
+- **High Performance:** Processes millions of events per second, about **10x faster** than channels.
+- **Generic:** Works with any type implementing the `Event` interface.
+- **Asynchronous:** Each subscriber runs in its own goroutine, ensuring non-blocking event handling.
+
+**Use When:**
+- ✅ Decoupling modules within a single Go process.
+- ✅ Implementing lightweight pub/sub or event-driven patterns.
+- ✅ Needing high-throughput, low-latency event dispatching.
+- ✅ Preferring a simple, dependency-free solution.
+
+**Not For:**
+- ❌ Inter-process/service communication (use Kafka, NATS, etc.).
+- ❌ Event persistence, durability, or advanced routing/filtering.
+- ❌ Cross-language/platform scenarios.
+- ❌ Event replay, dead-letter queues, or deduplication.
+- ❌ Heavy subscribe/unsubscribe churn or massive dynamic subscriber counts.
+
 ## Generic In-Process Pub/Sub
 
 This repository contains a **simple, in-process event dispatcher** to be used to decouple internal modules. It provides a generic way to define events, publish and subscribe to them.
@@ -78,12 +98,18 @@ It should output something along these lines, where order is not guaranteed give
 
 ## Benchmarks
 
+Please note that the benchmarks are run on a 13th Gen Intel(R) Core(TM) i7-13700K CPU, and results may vary based on the machine and environment. This one demonstrates the publishing throughput of the event dispatcher, at different number of event types and subscribers.
+
 ```
 cpu: 13th Gen Intel(R) Core(TM) i7-13700K
-BenchmarkEvent/1x1-24          164887946       14.55 ns/op     68.70 million/s      0 B/op     0 allocs/op
-BenchmarkEvent/1x10-24          28896586       80.06 ns/op     88.02 million/s    204 B/op     0 allocs/op
-BenchmarkEvent/1x100-24          1535168        1397 ns/op     71.57 million/s     26 B/op     0 allocs/op
-BenchmarkEvent/10x1-24          14288467       256.1 ns/op     39.03 million/s      0 B/op     0 allocs/op
-BenchmarkEvent/10x10-24          1624722        1265 ns/op     78.59 million/s     65 B/op     0 allocs/op
-BenchmarkEvent/10x100-24          164623       12767 ns/op     78.33 million/s    456 B/op     0 allocs/op
+BenchmarkEvent/1x1-24          14.72 ns/op     67.95 million/s        0 B/op      0 allocs/op
+BenchmarkEvent/1x10-24         84.61 ns/op     90.93 million/s      196 B/op      0 allocs/op
+BenchmarkEvent/1x100-24         1409 ns/op     70.95 million/s       10 B/op      0 allocs/op
+BenchmarkEvent/10x1-24         155.3 ns/op     64.38 million/s        0 B/op      0 allocs/op
+BenchmarkEvent/10x10-24         1315 ns/op     76.05 million/s       30 B/op      0 allocs/op
+BenchmarkEvent/10x100-24       13541 ns/op     73.84 million/s      210 B/op      0 allocs/op
 ```
+
+# License
+
+This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.

default_test.go

@@ -14,13 +14,14 @@ import (
 )
 
 /*
+go test -bench=. -benchmem -benchtime=10s
 cpu: 13th Gen Intel(R) Core(TM) i7-13700K
-BenchmarkEvent/1x1-24         	38709926	        31.94 ns/op	        30.89 million/s	       1 B/op	       0 allocs/op
-BenchmarkEvent/1x10-24        	 8107938	       133.7 ns/op	        74.76 million/s	      45 B/op	       0 allocs/op
-BenchmarkEvent/1x100-24       	  774168	      1341 ns/op	        72.65 million/s	     373 B/op	       0 allocs/op
-BenchmarkEvent/10x1-24        	 5755402	       301.1 ns/op	        32.98 million/s	       7 B/op	       0 allocs/op
-BenchmarkEvent/10x10-24       	  750022	      1503 ns/op	        64.47 million/s	     438 B/op	       0 allocs/op
-BenchmarkEvent/10x100-24      	   69363	     14878 ns/op	        67.11 million/s	    3543 B/op	       0 allocs/op
+BenchmarkEvent/1x1-24                   814403188               14.72 ns/op             67.95 million/s        0 B/op          0 allocs/op
+BenchmarkEvent/1x10-24                  161012098               84.61 ns/op             90.93 million/s      196 B/op          0 allocs/op
+BenchmarkEvent/1x100-24                  7890922              1409 ns/op                70.95 million/s       10 B/op          0 allocs/op
+BenchmarkEvent/10x1-24                  72358305               155.3 ns/op              64.38 million/s        0 B/op          0 allocs/op
+BenchmarkEvent/10x10-24                  7632547              1315 ns/op                76.05 million/s       30 B/op          0 allocs/op
+BenchmarkEvent/10x100-24                  832560             13541 ns/op                73.84 million/s      210 B/op          0 allocs/op
 */
 func BenchmarkEvent(b *testing.B) {
 	for _, topics := range []int{1, 10} {
@@ -52,6 +53,23 @@ func BenchmarkEvent(b *testing.B) {
 	}
 }
 
+/*
+cpu: 13th Gen Intel(R) Core(TM) i7-13700K
+BenchmarkSubcribeConcurrent-24    	 1826686	       606.3 ns/op	    1648 B/op	       5 allocs/op
+*/
+func BenchmarkSubscribeConcurrent(b *testing.B) {
+	d := NewDispatcher()
+	b.ReportAllocs()
+	b.ResetTimer()
+
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			unsub := Subscribe(d, func(ev MyEvent1) {})
+			unsub()
+		}
+	})
+}
+
 func TestDefaultPublish(t *testing.T) {
 	var wg sync.WaitGroup
 

event.go

@@ -22,7 +22,7 @@ type Event interface {
 // registry holds an immutable sorted array of event mappings
 type registry struct {
 	keys []uint32 // Event types (sorted)
-	subs []any    // Corresponding subscribers
+	grps []any    // Corresponding subscribers
 }
 
 // ------------------------------------- Dispatcher -------------------------------------
@@ -44,7 +44,7 @@ func NewDispatcher() *Dispatcher {
 
 	d.subs.Store(&registry{
 		keys: make([]uint32, 0, 16),
-		subs: make([]any, 0, 16),
+		grps: make([]any, 0, 16),
 	})
 	return d
 }
@@ -82,7 +82,7 @@ func (d *Dispatcher) findGroup(eventType uint32) any {
 	}
 
 	if left < len(keys) && keys[left] == eventType {
-		return reg.subs[left]
+		return reg.grps[left]
 	}
 	return nil
 }
@@ -116,35 +116,31 @@ func SubscribeTo[T Event](broker *Dispatcher, eventType uint32, handler func(T))
 	grp := &group[T]{cond: sync.NewCond(new(sync.Mutex))}
 	sub := grp.Add(handler)
 
-	// Copy-on-write with CAS loop: insert new entry in sorted position
-	for {
-		old := broker.subs.Load()
-		idx := sort.Search(len(old.keys), func(i int) bool {
-			return old.keys[i] >= eventType
-		})
-
-		// Create new arrays with space for one more element
-		newKeys := make([]uint32, len(old.keys)+1)
-		newSubs := make([]any, len(old.subs)+1)
-
-		// Copy elements before insertion point
-		copy(newKeys[:idx], old.keys[:idx])
-		copy(newSubs[:idx], old.subs[:idx])
-
-		// Insert new element
-		newKeys[idx] = eventType
-		newSubs[idx] = grp
-
-		// Copy elements after insertion point
-		copy(newKeys[idx+1:], old.keys[idx:])
-		copy(newSubs[idx+1:], old.subs[idx:])
-
-		// Atomically swap the registry
-		newReg := &registry{keys: newKeys, subs: newSubs}
-		if broker.subs.CompareAndSwap(old, newReg) {
-			break
-		}
-	}
+	// Copy-on-write: insert new entry in sorted position
+	old := broker.subs.Load()
+	idx := sort.Search(len(old.keys), func(i int) bool {
+		return old.keys[i] >= eventType
+	})
+
+	// Create new arrays with space for one more element
+	newKeys := make([]uint32, len(old.keys)+1)
+	newGrps := make([]any, len(old.grps)+1)
+
+	// Copy elements before insertion point
+	copy(newKeys[:idx], old.keys[:idx])
+	copy(newGrps[:idx], old.grps[:idx])
+
+	// Insert new element
+	newKeys[idx] = eventType
+	newGrps[idx] = grp
+
+	// Copy elements after insertion point
+	copy(newKeys[idx+1:], old.keys[idx:])
+	copy(newGrps[idx+1:], old.grps[idx:])
+
+	// Atomically store the new registry (mutex ensures no concurrent writers)
+	newReg := &registry{keys: newKeys, grps: newGrps}
+	broker.subs.Store(newReg)
 
 	// Start processing
 	go grp.Process(broker.df, broker.done)
@@ -205,14 +201,13 @@ func (s *consumer[T]) Listen(c *sync.Cond, fn func(T)) {
 
 		// Swap buffers and reset the current queue
 		temp := s.queue
-		s.queue = pending
+		s.queue = pending[:0]
 		pending = temp
-		s.queue = s.queue[:0]
 		c.L.Unlock()
 
 		// Outside of the critical section, process the work
-		for i := 0; i < len(pending); i++ {
-			fn(pending[i])
+		for _, event := range pending {
+			fn(event)
 		}
 	}
 }
@@ -228,6 +223,7 @@ type group[T Event] struct {
 // Process periodically broadcasts events
 func (s *group[T]) Process(interval time.Duration, done chan struct{}) {
 	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
 	for {
 		select {
 		case <-done:
@@ -250,7 +246,7 @@ func (s *group[T]) Broadcast(ev T) {
 // Add adds a subscriber to the list
 func (s *group[T]) Add(handler func(T)) *consumer[T] {
 	sub := &consumer[T]{
-		queue: make([]T, 0, 128),
+		queue: make([]T, 0, 64),
 	}
 
 	// Add the consumer to the list of active consumers
@@ -270,13 +266,13 @@ func (s *group[T]) Del(sub *consumer[T]) {
 
 	// Search and remove the subscriber
 	sub.stop = true
-	subs := make([]*consumer[T], 0, len(s.subs))
-	for _, v := range s.subs {
-		if v != sub {
-			subs = append(subs, v)
+	for i, v := range s.subs {
+		if v == sub {
+			copy(s.subs[i:], s.subs[i+1:])
+			s.subs = s.subs[:len(s.subs)-1]
+			break
 		}
 	}
-	s.subs = subs
 }
 
 // ------------------------------------- Debugging -------------------------------------