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(event Event) error {
41 if event.ID == "" {
42 event.ID = uuid.New().String()
43 }
44
45 if event.Timestamp.IsZero() {
46 event.Timestamp = time.Now()
47 }
48
49 select {
50 case bus.eventChan <- event:
51 log.Debug().
52 Str("event_id", event.ID).
53 Str("event_type", string(event.Type)).
54 Str("image_name", event.ImageName).
55 Str("tag", event.Tag).
56 Msg("Event published")
57 return nil
58 case <-bus.ctx.Done():
59 return fmt.Errorf("event bus is stopped")
60 default:
61 return fmt.Errorf("event channel is full, dropping event %s", event.ID)
62 }
63}
64
65func (bus *InMemoryEventBus) Subscribe(handler EventHandler) error {
66 bus.mu.Lock()
67 defer bus.mu.Unlock()
68
69 bus.handlers = append(bus.handlers, handler)
70 log.Debug().
71 Str("handler_type", fmt.Sprintf("%T", handler)).
72 Int("total_handlers", len(bus.handlers)).
73 Msg("Event handler subscribed")
74
75 return nil
76}
77
78func (bus *InMemoryEventBus) Unsubscribe(handler EventHandler) error {
79 bus.mu.Lock()
80 defer bus.mu.Unlock()
81
82 for i, h := range bus.handlers {
83 if h == handler {
84 bus.handlers = append(bus.handlers[:i], bus.handlers[i+1:]...)
85 log.Debug().
86 Str("handler_type", fmt.Sprintf("%T", handler)).
87 Int("total_handlers", len(bus.handlers)).
88 Msg("Event handler unsubscribed")
89 return nil
90 }
91 }
92
93 return fmt.Errorf("handler not found")
94}
95
96func (bus *InMemoryEventBus) Start() error {
97 log.Info().
98 Int("buffer_size", bus.bufferSize).
99 Msg("Starting event bus")
100
101 go bus.processEvents()
102 return nil
103}
104
105func (bus *InMemoryEventBus) Stop() error {
106 log.Info().Msg("Stopping event bus")
107
108 bus.cancel()
109
110 select {
111 case <-bus.done:
112 log.Info().Msg("Event bus stopped")
113 return nil
114 case <-time.After(5 * time.Second):
115 log.Warn().Msg("Event bus stop timeout")
116 return fmt.Errorf("timeout waiting for event bus to stop")
117 }
118}
119
120func (bus *InMemoryEventBus) processEvents() {
121 defer close(bus.done)
122
123 for {
124 select {
125 case event := <-bus.eventChan:
126 bus.handleEvent(event)
127 case <-bus.ctx.Done():
128 log.Debug().Msg("Event bus processing stopped")
129 return
130 }
131 }
132}
133
134func (bus *InMemoryEventBus) handleEvent(event Event) {
135 bus.mu.RLock()
136 handlers := make([]EventHandler, len(bus.handlers))
137 copy(handlers, bus.handlers)
138 bus.mu.RUnlock()
139
140 for _, handler := range handlers {
141 if handler.CanHandle(event.Type) {
142 go func(h EventHandler) {
143 start := time.Now()
144
145 if err := h.Handle(event); err != nil {
146 log.Error().
147 Err(err).
148 Str("event_id", event.ID).
149 Str("event_type", string(event.Type)).
150 Str("handler_type", fmt.Sprintf("%T", h)).
151 Msg("Error handling event")
152 } else {
153 log.Debug().
154 Str("event_id", event.ID).
155 Str("event_type", string(event.Type)).
156 Str("handler_type", fmt.Sprintf("%T", h)).
157 Dur("duration", time.Since(start)).
158 Msg("Event handled successfully")
159 }
160 }(handler)
161 }
162 }
163}