63 lines
1.6 KiB
Go
63 lines
1.6 KiB
Go
package worker
|
|
|
|
import "sync"
|
|
|
|
// SnapshotRefreshNotifier requests every subscribed Worker snapshot stream to
|
|
// issue a new full snapshot. Notifications are coalesced per subscriber.
|
|
type SnapshotRefreshNotifier interface {
|
|
NotifySnapshotRefresh()
|
|
}
|
|
|
|
// SnapshotRefreshSubscriber receives controller-local full-snapshot refresh
|
|
// requests. The cancel function releases the subscription.
|
|
type SnapshotRefreshSubscriber interface {
|
|
SubscribeSnapshotRefresh() (<-chan struct{}, func())
|
|
}
|
|
|
|
// SnapshotRefreshBroker fans a management-plane change out to every local
|
|
// Worker snapshot stream. It contains no proxy or client data.
|
|
type SnapshotRefreshBroker struct {
|
|
mu sync.Mutex
|
|
subscribers map[chan struct{}]struct{}
|
|
}
|
|
|
|
func NewSnapshotRefreshBroker() *SnapshotRefreshBroker {
|
|
return &SnapshotRefreshBroker{subscribers: make(map[chan struct{}]struct{})}
|
|
}
|
|
|
|
func (broker *SnapshotRefreshBroker) NotifySnapshotRefresh() {
|
|
if broker == nil {
|
|
return
|
|
}
|
|
broker.mu.Lock()
|
|
defer broker.mu.Unlock()
|
|
for subscriber := range broker.subscribers {
|
|
select {
|
|
case subscriber <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
}
|
|
|
|
func (broker *SnapshotRefreshBroker) SubscribeSnapshotRefresh() (<-chan struct{}, func()) {
|
|
if broker == nil {
|
|
return nil, func() {}
|
|
}
|
|
subscriber := make(chan struct{}, 1)
|
|
broker.mu.Lock()
|
|
broker.subscribers[subscriber] = struct{}{}
|
|
broker.mu.Unlock()
|
|
|
|
var once sync.Once
|
|
return subscriber, func() {
|
|
once.Do(func() {
|
|
broker.mu.Lock()
|
|
delete(broker.subscribers, subscriber)
|
|
broker.mu.Unlock()
|
|
})
|
|
}
|
|
}
|
|
|
|
var _ SnapshotRefreshNotifier = (*SnapshotRefreshBroker)(nil)
|
|
var _ SnapshotRefreshSubscriber = (*SnapshotRefreshBroker)(nil)
|