66 lines
1.2 KiB
Go
66 lines
1.2 KiB
Go
package events
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
|
|
"github.com/containerd/containerd/api/types/event"
|
|
"github.com/containerd/containerd/namespaces"
|
|
goevents "github.com/docker/go-events"
|
|
)
|
|
|
|
const (
|
|
EventVersion = "v1"
|
|
)
|
|
|
|
type Emitter struct {
|
|
sinks map[string]*eventSink
|
|
broadcaster *goevents.Broadcaster
|
|
m sync.Mutex
|
|
}
|
|
|
|
func NewEmitter() *Emitter {
|
|
return &Emitter{
|
|
sinks: make(map[string]*eventSink),
|
|
broadcaster: goevents.NewBroadcaster(),
|
|
m: sync.Mutex{},
|
|
}
|
|
}
|
|
|
|
func (e *Emitter) Post(ctx context.Context, evt Event) error {
|
|
if err := e.broadcaster.Write(&sinkEvent{
|
|
ctx: ctx,
|
|
event: evt,
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (e *Emitter) Events(ctx context.Context, clientID string) chan *event.Envelope {
|
|
e.m.Lock()
|
|
if _, ok := e.sinks[clientID]; !ok {
|
|
ns, _ := namespaces.Namespace(ctx)
|
|
s := &eventSink{
|
|
ch: make(chan *event.Envelope),
|
|
ns: ns,
|
|
}
|
|
e.sinks[clientID] = s
|
|
e.broadcaster.Add(s)
|
|
}
|
|
ch := e.sinks[clientID].ch
|
|
e.m.Unlock()
|
|
|
|
return ch
|
|
}
|
|
|
|
func (e *Emitter) Remove(clientID string) {
|
|
e.m.Lock()
|
|
if v, ok := e.sinks[clientID]; ok {
|
|
e.broadcaster.Remove(v)
|
|
delete(e.sinks, clientID)
|
|
}
|
|
e.m.Unlock()
|
|
}
|