feat: make the observers asynchronous

Author: ziccardi

async.go

@@ -1,12 +1,14 @@
 package async
 
 import (
+	"math"
 	"time"
 )
 
 const (
 	// DefaultRoutineSnapshottingInterval defines how often the routine manager checks routine status
 	DefaultRoutineSnapshottingInterval = 30 * time.Second
+	DefaultObserverTimeout             = time.Duration(math.MaxInt64)
 )
 
 // A RoutinesObserver is an object that observes the status of the executions of routines.
@@ -29,3 +31,40 @@ type RoutinesObserver interface {
 	// currently running
 	RunningRoutineByNameCount(name string, count int)
 }
+
+type routineEventType int
+
+const (
+	routineStarted routineEventType = iota
+	routineEnded
+	routineTimeboxExceeded
+	takeSnapshot
+)
+
+type routineEvent struct {
+	Type     routineEventType
+	routine  AsyncRoutine
+	snapshot Snapshot
+}
+
+func newRoutineEvent(eventType routineEventType) routineEvent {
+	return routineEvent{
+		Type: eventType,
+	}
+}
+
+func routineStartedEvent() routineEvent {
+	return newRoutineEvent(routineStarted)
+}
+
+func routineFinishedEvent() routineEvent {
+	return newRoutineEvent(routineEnded)
+}
+
+func routineTimeboxExceededEvent() routineEvent {
+	return newRoutineEvent(routineTimeboxExceeded)
+}
+
+func takeSnapshotEvent() routineEvent {
+	return newRoutineEvent(takeSnapshot)
+}

async_routine.go

@@ -145,14 +145,10 @@ func (r *asyncRoutine) run(manager AsyncRoutineManager) {
 			r.status = RoutineStatusFinished
 		}
 		manager.deregister(r)
-		manager.notify(func(observer RoutinesObserver) {
-			observer.RoutineFinished(r)
-		})
+		manager.notifyAll(r, routineFinishedEvent())
 	}
 
-	manager.notify(func(observer RoutinesObserver) {
-		observer.RoutineStarted(r)
-	})
+	manager.notifyAll(r, routineStartedEvent())
 
 	if r.errGroup != nil {
 		r.errGroup.Go(func() error {

async_routine_manager.go

@@ -16,7 +16,7 @@ type AsyncRoutineManager interface {
 	RemoveObserver(observerId string)
 	IsEnabled() bool
 	GetSnapshot() Snapshot
-	notify(eventSource func(observer RoutinesObserver))
+	notifyAll(src AsyncRoutine, evt routineEvent)
 	Monitor() AsyncRoutineMonitor
 	register(routine AsyncRoutine)
 	deregister(routine AsyncRoutine)
@@ -30,7 +30,7 @@ type asyncRoutineManager struct {
 	snapshottingToggle   Toggle
 	snapshottingInterval time.Duration
 	routines             cmap.ConcurrentMap[string, AsyncRoutine]
-	observers            cmap.ConcurrentMap[string, RoutinesObserver]
+	observers            cmap.ConcurrentMap[string, *observerProxy]
 
 	monitorLock sync.Mutex // user to sync the `Start` and `Stop` methods that are used to start the
 	// snapshotting routine
@@ -47,12 +47,26 @@ func (arm *asyncRoutineManager) IsEnabled() bool {
 // Assigns and returns an observer ID to the RoutineObserver
 func (arm *asyncRoutineManager) AddObserver(observer RoutinesObserver) string {
 	uid := uuid.New().String()
-	arm.observers.Set(uid, observer)
+	arm.observers.Set(uid, newObserverProxy(uid, observer, arm, DefaultObserverTimeout))
+	return uid
+}
+
+// AddObserverWithTimeout registers a new RoutinesObserver with the asyncRoutineManager,
+// associating it with a unique identifier and a specified timeout duration.
+// The function returns the unique ID assigned to the observer.
+func (arm *asyncRoutineManager) AddObserverWithTimeout(observer RoutinesObserver, timeout time.Duration) string {
+	uid := uuid.New().String()
+	arm.observers.Set(uid, newObserverProxy(uid, observer, arm, timeout))
 	return uid
 }
 
 // RemoveObserver removes the given RoutineObserver from the list of observers
 func (arm *asyncRoutineManager) RemoveObserver(observerId string) {
+	observer, ok := arm.observers.Get(observerId)
+	if !ok {
+		return
+	}
+	close(observer.channel)
 	arm.observers.Remove(observerId)
 }
 
@@ -64,9 +78,10 @@ func (arm *asyncRoutineManager) GetSnapshot() Snapshot {
 	return snapshot
 }
 
-func (arm *asyncRoutineManager) notify(eventSource func(observer RoutinesObserver)) {
+// notifyAll notifies all the observers of the event evt received from the routine src
+func (arm *asyncRoutineManager) notifyAll(src AsyncRoutine, evt routineEvent) {
 	for _, observer := range arm.observers.Items() {
-		eventSource(observer)
+		observer.notify(src, evt)
 	}
 }
 
@@ -113,7 +128,7 @@ var lock sync.RWMutex
 func newAsyncRoutineManager(options ...AsyncManagerOption) AsyncRoutineManager {
 	mgr := &asyncRoutineManager{
 		routines:             cmap.New[AsyncRoutine](),
-		observers:            cmap.New[RoutinesObserver](),
+		observers:            cmap.New[*observerProxy](),
 		snapshottingInterval: DefaultRoutineSnapshottingInterval,
 		ctx:                  context.Background(),
 		managerToggle:        func() bool { return true }, // manager is enabled by default

async_routine_monitor.go

@@ -66,15 +66,8 @@ func (arm *asyncRoutineManager) snapshot() {
 	snapshot := arm.GetSnapshot()
 
 	for _, r := range snapshot.GetTimedOutRoutines() {
-		arm.notify(func(observer RoutinesObserver) {
-			observer.RoutineExceededTimebox(r)
-		})
+		arm.notifyAll(r, routineTimeboxExceededEvent())
 	}
 
-	arm.notify(func(observer RoutinesObserver) {
-		observer.RunningRoutineCount(snapshot.totalRoutineCount)
-		for _, name := range snapshot.GetRunningRoutinesNames() {
-			observer.RunningRoutineByNameCount(name, snapshot.GetRunningRoutinesCount(name))
-		}
-	})
+	arm.notifyAll(nil, takeSnapshotEvent())
 }

async_routine_observer_proxy.go

@@ -0,0 +1,75 @@
+package async
+
+import (
+	"context"
+	"time"
+)
+
+// observerProxy acts as an intermediary between the AsyncRoutineManager and a RoutinesObserver.
+// It receives routine events via a channel and dispatches them to the observer's callback methods.
+// The proxy manages event notification asynchronously and can enforce a timeout on the observer's lifecycle.
+type observerProxy struct {
+	manager  AsyncRoutineManager
+	observer RoutinesObserver
+	channel  chan routineEvent
+}
+
+// newObserverProxy creates and initializes a new observerProxy instance.
+// It sets up an asynchronous routine that listens for routine events on the proxy's channel
+// and forwards them to the appropriate methods of the provided RoutinesObserver.
+//
+// Parameters:
+//   - observerId: a unique identifier for the observer instance.
+//   - observer: the RoutinesObserver to be notified of routine events.
+//   - manager: the AsyncRoutineManager responsible for managing routines.
+//   - timeout: the duration after which the observer routine is considered 'exceeding the timebox'.
+//
+// Returns:
+//   - A pointer to the initialized observerProxy.
+func newObserverProxy(observerId string, observer RoutinesObserver, manager AsyncRoutineManager, timeout time.Duration) *observerProxy {
+	proxy := &observerProxy{
+		manager:  manager,
+		observer: observer,
+		channel:  make(chan routineEvent),
+	}
+	NewAsyncRoutine("async-observer-notifier", context.Background(), func() {
+		for evt := range proxy.channel {
+			switch evt.Type {
+			case routineStarted:
+				observer.RoutineStarted(evt.routine)
+			case routineEnded:
+				observer.RoutineFinished(evt.routine)
+			case routineTimeboxExceeded:
+				observer.RoutineExceededTimebox(evt.routine)
+			case takeSnapshot:
+				observer.RunningRoutineCount(evt.snapshot.GetTotalRoutineCount())
+				for _, routineName := range evt.snapshot.GetRunningRoutinesNames() {
+					observer.RunningRoutineByNameCount(routineName, evt.snapshot.GetRunningRoutinesCount(routineName))
+				}
+			}
+		}
+	}).
+		Timebox(timeout).
+		WithData("observer-id", observerId).
+		Run()
+
+	return proxy
+}
+
+// notify sends a routine event to the observerProxy's channel.
+// Depending on the event type, it either forwards the routine information
+// or triggers a snapshot retrieval from the manager.
+func (proxy *observerProxy) notify(routine AsyncRoutine, evt routineEvent) {
+	switch evt.Type {
+	case routineStarted, routineEnded, routineTimeboxExceeded:
+		proxy.channel <- routineEvent{
+			Type:    evt.Type,
+			routine: routine,
+		}
+	case takeSnapshot:
+		proxy.channel <- routineEvent{
+			Type:     takeSnapshot,
+			snapshot: proxy.manager.GetSnapshot(),
+		}
+	}
+}