Merge pull request #2974 from aduffeck/async-events

Handle events asynchronously

services/clientlog/pkg/service/service.go

@@ -83,7 +83,8 @@ EventLoop:
 			if !ok {
 				break EventLoop
 			}
-			cl.processEvent(event)
+
+			go cl.processEvent(event)
 
 			if cl.stopped.Load() {
 				break EventLoop

services/sse/pkg/service/service.go

@@ -49,17 +49,19 @@ func (s SSE) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 // ListenForEvents listens for events
 func (s SSE) ListenForEvents() {
 	for e := range s.evChannel {
-		switch ev := e.Event.(type) {
-		default:
-			s.l.Error().Interface("event", ev).Msg("unhandled event")
-		case events.SendSSE:
-			for _, uid := range ev.UserIDs {
-				s.sse.Publish(uid, &sse.Event{
-					Event: []byte(ev.Type),
-					Data:  ev.Message,
-				})
+		go func() {
+			switch ev := e.Event.(type) {
+			default:
+				s.l.Error().Interface("event", ev).Msg("unhandled event")
+			case events.SendSSE:
+				for _, uid := range ev.UserIDs {
+					s.sse.Publish(uid, &sse.Event{
+						Event: []byte(ev.Type),
+						Data:  ev.Message,
+					})
+				}
 			}
-		}
+		}()
 	}
 }
 

services/userlog/pkg/service/service.go

@@ -102,7 +102,7 @@ func (ul *UserlogService) MemorizeEvents(ch <-chan events.Event) {
 	for i := 0; i < ul.cfg.MaxConcurrency; i++ {
 		go func(ch <-chan events.Event) {
 			for event := range ch {
-				ul.processEvent(event)
+				go ul.processEvent(event)
 			}
 		}(ch)
 	}