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