clawbot
5100b9a665
check / check (push) Canceled after 0s
The IRC relay registers a broker wait channel on each pass of its loop, and when the 100 ms ticker fired instead of a notification it looped without removing that channel. An idle IRC listener connection added ten channels a second to the broker, and all but the last stayed after the connection closed. The ticker case now removes its channel, as the HTTP long-poll does after its timer. Broker.ListenerCount returns how many wait channels a user holds, so the test can check that an idle connection holds at most one, and none after it closes. Model: opus-5-5 Co-authored-by: clawbot <sneak+clawbot@sneak.cloud>
84 lines
1.7 KiB
Go
84 lines
1.7 KiB
Go
// Package broker provides an in-memory pub/sub for long-poll notifications.
|
|
package broker
|
|
|
|
import (
|
|
"sync"
|
|
)
|
|
|
|
// Broker notifies waiting clients when new messages are available.
|
|
type Broker struct {
|
|
mu sync.Mutex
|
|
listeners map[int64][]chan struct{}
|
|
}
|
|
|
|
// New creates a new Broker.
|
|
func New() *Broker {
|
|
return &Broker{ //nolint:exhaustruct // mu has zero-value default
|
|
listeners: make(map[int64][]chan struct{}),
|
|
}
|
|
}
|
|
|
|
// Wait returns a channel that will be closed when a message
|
|
// is available for the user.
|
|
func (b *Broker) Wait(userID int64) chan struct{} {
|
|
waitCh := make(chan struct{}, 1)
|
|
|
|
b.mu.Lock()
|
|
b.listeners[userID] = append(
|
|
b.listeners[userID], waitCh,
|
|
)
|
|
b.mu.Unlock()
|
|
|
|
return waitCh
|
|
}
|
|
|
|
// Notify wakes up all waiting clients for a user.
|
|
func (b *Broker) Notify(userID int64) {
|
|
b.mu.Lock()
|
|
waiters := b.listeners[userID]
|
|
delete(b.listeners, userID)
|
|
b.mu.Unlock()
|
|
|
|
for _, waiter := range waiters {
|
|
select {
|
|
case waiter <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
}
|
|
|
|
// ListenerCount returns how many channels from Wait are still
|
|
// registered for the user: those that neither Notify nor Remove
|
|
// has taken out.
|
|
func (b *Broker) ListenerCount(userID int64) int {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
|
|
return len(b.listeners[userID])
|
|
}
|
|
|
|
// Remove removes a specific wait channel (for cleanup on timeout).
|
|
func (b *Broker) Remove(
|
|
userID int64,
|
|
waitCh chan struct{},
|
|
) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
|
|
waiters := b.listeners[userID]
|
|
|
|
for i, waiter := range waiters {
|
|
if waiter == waitCh {
|
|
b.listeners[userID] = append(
|
|
waiters[:i], waiters[i+1:]...,
|
|
)
|
|
|
|
break
|
|
}
|
|
}
|
|
|
|
if len(b.listeners[userID]) == 0 {
|
|
delete(b.listeners, userID)
|
|
}
|
|
}
|