Compare commits
2 Commits
e0571eef88
...
2e8d9e8570
| Author | SHA1 | Date |
|---|---|---|
 Domagoj Zecevic
|
2e8d9e8570 | 3 weeks ago |
|
Domagoj Zecevic
|
ef46fc849a | 3 weeks ago |
15 changed files with 682 additions and 89 deletions
@ -0,0 +1,41 @@ |
|||
package sensor |
|||
|
|||
import "sync" |
|||
|
|||
// Hub broadcasts refresh notifications. Its zero value is ready to use, and all
|
|||
// methods are concurrency-safe. Bursts coalesce so slow subscribers never block
|
|||
// the sender or other subscribers.
|
|||
type Hub struct { |
|||
mu sync.Mutex |
|||
subscribers map[chan struct{}]struct{} |
|||
} |
|||
|
|||
// Subscribe registers a buffered notification channel. The caller must cancel
|
|||
// when finished; cancellation is idempotent and leaves the channel open.
|
|||
func (h *Hub) Subscribe() (<-chan struct{}, func()) { |
|||
events := make(chan struct{}, 1) |
|||
h.mu.Lock() |
|||
if h.subscribers == nil { |
|||
h.subscribers = make(map[chan struct{}]struct{}) |
|||
} |
|||
h.subscribers[events] = struct{}{} |
|||
h.mu.Unlock() |
|||
return events, func() { |
|||
h.mu.Lock() |
|||
delete(h.subscribers, events) |
|||
h.mu.Unlock() |
|||
} |
|||
} |
|||
|
|||
// Notify queues at most one pending notification per subscriber without waiting
|
|||
// for readers. Notifications carry no data; clients fetch the current snapshot.
|
|||
func (h *Hub) Notify() { |
|||
h.mu.Lock() |
|||
defer h.mu.Unlock() |
|||
for events := range h.subscribers { |
|||
select { |
|||
case events <- struct{}{}: |
|||
default: |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,75 @@ |
|||
package sensor |
|||
|
|||
import ( |
|||
"sync" |
|||
"testing" |
|||
"time" |
|||
) |
|||
|
|||
func TestHubDeliveryCoalescingAndCancel(t *testing.T) { |
|||
var hub Hub |
|||
slow, cancelSlow := hub.Subscribe() |
|||
fast, cancelFast := hub.Subscribe() |
|||
defer cancelFast() |
|||
done := make(chan struct{}) |
|||
go func() { |
|||
for range 1000 { |
|||
hub.Notify() |
|||
} |
|||
close(done) |
|||
}() |
|||
select { |
|||
case <-done: |
|||
case <-time.After(time.Second): |
|||
t.Fatal("slow subscribers blocked notification") |
|||
} |
|||
for _, events := range []<-chan struct{}{slow, fast} { |
|||
select { |
|||
case <-events: |
|||
default: |
|||
t.Fatal("subscriber missed notification") |
|||
} |
|||
select { |
|||
case <-events: |
|||
t.Fatal("burst was not coalesced") |
|||
default: |
|||
} |
|||
} |
|||
cancelSlow() |
|||
cancelSlow() |
|||
hub.Notify() |
|||
select { |
|||
case _, ok := <-slow: |
|||
if ok { |
|||
t.Fatal("cancelled subscriber received a new event") |
|||
} |
|||
default: |
|||
} |
|||
select { |
|||
case <-fast: |
|||
default: |
|||
t.Fatal("remaining subscriber missed notification") |
|||
} |
|||
} |
|||
|
|||
func TestHubConcurrentUse(t *testing.T) { |
|||
var hub Hub |
|||
hub.Notify() // The zero value also supports notification without subscribers.
|
|||
var workers sync.WaitGroup |
|||
for range 10 { |
|||
workers.Go(func() { |
|||
for range 100 { |
|||
events, cancel := hub.Subscribe() |
|||
hub.Notify() |
|||
select { |
|||
case <-events: |
|||
default: |
|||
t.Error("subscriber missed notification") |
|||
} |
|||
cancel() |
|||
cancel() |
|||
} |
|||
}) |
|||
} |
|||
workers.Wait() |
|||
} |
|||
Loading…
Reference in new issue