feat: add ExecCollectValuesWatch

Signed-off-by: Christian Stewart <[email protected]>

Author: paralin

bus/multi.go

@@ -164,3 +164,165 @@ func ExecCollectValues[T directive.Value](
 		}
 	}
 }
+
+// ExecCollectValuesWatch collects values and calls a callback whenever the slice changes.
+// The callback is called with a snapshot of the current values whenever values are added or removed or the error changes.
+// The callback runs in a separate goroutine and will be called again if values change while it's running.
+// If waitIdle is true, waits until the directive is idle before emitting values after additions.
+// Removals always emit immediately (unless no values have been emitted yet).
+// errorCb is called with any error returned by the callback (optional).
+// If the callback returns any error, it will not be called again.
+// Returns the directive instance and reference for cleanup.
+// If err != nil, ref == nil.
+func ExecCollectValuesWatch[T directive.Value](
+	ctx context.Context,
+	bus Bus,
+	dir directive.Directive,
+	waitIdle bool,
+	callback func(resErr error, vals []T) error,
+	errorCb func(err error),
+) (directive.Instance, directive.Reference, error) {
+	// bcast guards these variables
+	var bcast broadcast.Broadcast
+
+	var vals []T
+	var valIDs []uint32
+	var resErr error
+	var idle bool
+	var emittedOnce bool
+	var pendingEmit bool // Track if we need to emit
+
+	di, ref, err := bus.AddDirective(
+		dir,
+		NewCallbackHandler(
+			func(v directive.AttachedValue) { // Add handler
+				val, valOk := v.GetValue().(T)
+				if !valOk {
+					return
+				}
+				bcast.HoldLock(func(broadcast func(), getWaitCh func() <-chan struct{}) {
+					vals = append(vals, val)
+					valIDs = append(valIDs, v.GetValueID())
+
+					// Condition 1: Value added and (!waitIdle || idle)
+					if !waitIdle || idle {
+						pendingEmit = true
+						broadcast()
+					}
+				})
+			},
+			func(v directive.AttachedValue) { // Remove handler
+				_, valOk := v.GetValue().(T)
+				if !valOk {
+					return
+				}
+				bcast.HoldLock(func(broadcast func(), getWaitCh func() <-chan struct{}) {
+					id := v.GetValueID()
+					for i, valID := range valIDs {
+						if valID == id {
+							valIDs = slices.Clone(valIDs)
+							valIDs[i] = valIDs[len(valIDs)-1]
+							valIDs = valIDs[:len(valIDs)-1]
+							vals = slices.Clone(vals)
+							vals[i] = vals[len(vals)-1]
+							var empty T
+							vals[len(vals)-1] = empty
+							vals = vals[:len(vals)-1]
+
+							// Condition 3: Value removed and already emitted
+							if emittedOnce {
+								pendingEmit = true
+								broadcast()
+							}
+							break
+						}
+					}
+				})
+			},
+			func() { // Dispose handler
+				bcast.HoldLock(func(broadcast func(), getWaitCh func() <-chan struct{}) {
+					if resErr == nil {
+						resErr = directive.ErrDirectiveDisposed
+						pendingEmit = true
+						broadcast()
+					}
+				})
+			},
+		),
+	)
+	if err != nil {
+		if ref != nil {
+			ref.Release()
+		}
+		return nil, nil, err
+	}
+
+	defer di.AddIdleCallback(func(isIdle bool, errs []error) {
+		bcast.HoldLock(func(broadcast func(), getWaitCh func() <-chan struct{}) {
+			if resErr != nil {
+				return
+			}
+
+			wasIdle := idle
+			idle = isIdle
+
+			// Check for errors
+			for _, err := range errs {
+				if err != nil {
+					resErr = err
+					pendingEmit = true
+					broadcast()
+					return
+				}
+			}
+
+			// Condition 2: Became idle and haven't emitted yet
+			if !wasIdle && idle && !emittedOnce && len(vals) > 0 {
+				pendingEmit = true
+				broadcast()
+			}
+		})
+	})()
+
+	// Start goroutine to handle callbacks
+	go func() {
+		for {
+			var currVals []T
+			var currErr error
+			var shouldEmit bool
+			var waitCh <-chan struct{}
+
+			bcast.HoldLock(func(broadcast func(), getWaitCh func() <-chan struct{}) {
+				currErr = resErr
+				shouldEmit = pendingEmit
+				pendingEmit = false
+				waitCh = getWaitCh()
+
+				if shouldEmit && currErr == nil {
+					currVals = slices.Clone(vals)
+					emittedOnce = true
+				}
+			})
+
+			// Call callback if we should emit
+			if shouldEmit {
+				if callback != nil {
+					// Call with values
+					if err := callback(currErr, currVals); err != nil && errorCb != nil {
+						errorCb(err)
+						return
+					}
+				}
+			}
+
+			select {
+			case <-ctx.Done():
+				return
+			case <-waitCh:
+				// Continue to check for changes
+			}
+		}
+	}()
+
+	return di, ref, nil
+}