1package events
2
3import (
4 "context"
5 "fmt"
6 "sync"
7 "time"
8
9 "github.com/google/uuid"
10 "github.com/rs/zerolog/log"
11)
12
13type InMemoryEventBus struct {
14 handlers []EventHandler
15 eventChan chan Event
16 done chan struct{}
17 mu sync.RWMutex
18 ctx context.Context
19 cancel context.CancelFunc
20 bufferSize int
21}
22
23func NewInMemoryEventBus(bufferSize int) *InMemoryEventBus {
24 if bufferSize <= 0 {
25 bufferSize = 100
26 }
27
28 ctx, cancel := context.WithCancel(context.Background())
29
30 return &InMemoryEventBus{
31 handlers: make([]EventHandler, 0),
32 eventChan: make(chan Event, bufferSize),
33 done: make(chan struct{}),
34 ctx: ctx,
35 cancel: cancel,
36 bufferSize: bufferSize,
37 }
38}
39
40func (bus *InMemoryEventBus) Publish(eventType EventType, payload interface{}) error {
41 event := Event{
42 ID: uuid.New().String(),
43 Type: eventType,
44 Timestamp: time.Now(),
45 Data: payload,
46 }
47
48 // Add specific fields for known event types
49 switch p := payload.(type) {
50 case ImagePushedPayload:
51 event.ImageName = p.Name
52 event.Tag = p.Reference
53 }
54
55 select {
56 case bus.eventChan <- event:
57 log.Debug().
58 Str("event_id", event.ID).
59 Str("event_type", string(event.Type)).
60 Str("image_name", event.ImageName).
61 Str("tag", event.Tag).
62 Msg("Event published")
63 return nil
64 case <-bus.ctx.Done():
65 return fmt.Errorf("event bus is stopped")
66 default:
67 return fmt.Errorf("event channel is full, dropping event %s", event.ID)
68 }
69}
70
71func (bus *InMemoryEventBus) Subscribe(handler EventHandler) error {
72 bus.mu.Lock()
73 defer bus.mu.Unlock()
74
75 bus.handlers = append(bus.handlers, handler)
76 log.Debug().
77 Str("handler_type", fmt.Sprintf("%T", handler)).
78 Int("total_handlers", len(bus.handlers)).
79 Msg("Event handler subscribed")
80
81 return nil
82}
83
84func (bus *InMemoryEventBus) Unsubscribe(handler EventHandler) error {
85 bus.mu.Lock()
86 defer bus.mu.Unlock()
87
88 for i, h := range bus.handlers {
89 if h == handler {
90 bus.handlers = append(bus.handlers[:i], bus.handlers[i+1:]...)
91 log.Debug().
92 Str("handler_type", fmt.Sprintf("%T", handler)).
93 Int("total_handlers", len(bus.handlers)).
94 Msg("Event handler unsubscribed")
95 return nil
96 }
97 }
98
99 return fmt.Errorf("handler not found")
100}
101
102func (bus *InMemoryEventBus) Start() error {
103 log.Info().
104 Int("buffer_size", bus.bufferSize).
105 Msg("Starting event bus")
106
107 go bus.processEvents()
108 return nil
109}
110
111func (bus *InMemoryEventBus) Stop() error {
112 log.Info().Msg("Stopping event bus")
113
114 bus.cancel()
115
116 select {
117 case <-bus.done:
118 log.Info().Msg("Event bus stopped")
119 return nil
120 case <-time.After(5 * time.Second):
121 log.Warn().Msg("Event bus stop timeout")
122 return fmt.Errorf("timeout waiting for event bus to stop")
123 }
124}
125
126func (bus *InMemoryEventBus) processEvents() {
127 defer close(bus.done)
128
129 for {
130 select {
131 case event := <-bus.eventChan:
132 bus.handleEvent(event)
133 case <-bus.ctx.Done():
134 log.Debug().Msg("Event bus processing stopped")
135 return
136 }
137 }
138}
139
140func (bus *InMemoryEventBus) handleEvent(event Event) {
141 bus.mu.RLock()
142 handlers := make([]EventHandler, len(bus.handlers))
143 copy(handlers, bus.handlers)
144 bus.mu.RUnlock()
145
146 for _, handler := range handlers {
147 if handler.CanHandle(event.Type) {
148 go func(h EventHandler) {
149 start := time.Now()
150
151 if err := h.Handle(event); err != nil {
152 log.Error().
153 Err(err).
154 Str("event_id", event.ID).
155 Str("event_type", string(event.Type)).
156 Str("handler_type", fmt.Sprintf("%T", h)).
157 Msg("Error handling event")
158 } else {
159 log.Debug().
160 Str("event_id", event.ID).
161 Str("event_type", string(event.Type)).
162 Str("handler_type", fmt.Sprintf("%T", h)).
163 Dur("duration", time.Since(start)).
164 Msg("Event handled successfully")
165 }
166 }(handler)
167 }
168 }
169}