Add Signal to context provider interface (elastic#4368)

internal/pkg/composable/controller.go

@@ -300,6 +300,21 @@ type contextProviderState struct {
 	signal   chan bool
 }
 
+// Signal signals that something has changed in the provider.
+//
+// Note: This should only be used by fetch context providers, standard context
+// providers should use Set to update the overall state.
+func (c *contextProviderState) Signal() {
+	// Notify the controller Run loop that a state has changed. The notification
+	// channel has buffer size 1 so this ensures that an update will always
+	// happen after this change, while coalescing multiple simultaneous changes
+	// into a single controller update.
+	select {
+	case c.signal <- true:
+	default:
+	}
+}
+
 // Set sets the current mapping.
 func (c *contextProviderState) Set(mapping map[string]interface{}) error {
 	var err error
@@ -321,15 +336,7 @@ func (c *contextProviderState) Set(mapping map[string]interface{}) error {
 		return nil
 	}
 	c.mapping = mapping
-
-	// Notify the controller Run loop that a state has changed. The notification
-	// channel has buffer size 1 so this ensures that an update will always
-	// happen after this change, while coalescing multiple simultaneous changes
-	// into a single controller update.
-	select {
-	case c.signal <- true:
-	default:
-	}
+	c.Signal()
 	return nil
 }
 

internal/pkg/composable/testing/context.go

@@ -16,6 +16,8 @@ type ContextComm struct {
 	lock     sync.Mutex
 	previous map[string]interface{}
 	current  map[string]interface{}
+	// Signal calls onSignal.
+	onSignal func()
 	// Set calls onSet with the new value, so tests can
 	// verify data from the provider.
 	onSet func(map[string]interface{})
@@ -28,6 +30,16 @@ func NewContextComm(ctx context.Context) *ContextComm {
 	}
 }
 
+// Signal signals that something has changed in the provider.
+func (t *ContextComm) Signal() {
+	t.lock.Lock()
+	onSignal := t.onSignal
+	t.lock.Unlock()
+	if onSignal != nil {
+		onSignal()
+	}
+}
+
 // Set sets the current mapping for the context.
 func (t *ContextComm) Set(mapping map[string]interface{}) error {
 	var err error
@@ -45,6 +57,7 @@ func (t *ContextComm) Set(mapping map[string]interface{}) error {
 	if onSet != nil {
 		onSet(mapping)
 	}
+	t.Signal()
 
 	return nil
 }
@@ -63,6 +76,13 @@ func (t *ContextComm) Current() map[string]interface{} {
 	return t.current
 }
 
+// CallOnSignal sets the OnSignal callback.
+func (t *ContextComm) CallOnSignal(f func()) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	t.onSignal = f
+}
+
 // CallOnSet sets the OnSet callback.
 func (t *ContextComm) CallOnSet(f func(map[string]interface{})) {
 	t.lock.Lock()

internal/pkg/core/composable/providers.go

@@ -19,6 +19,12 @@ type FetchContextProvider interface {
 type ContextProviderComm interface {
 	context.Context
 
+	// Signal signals that something has changed in the provider.
+	//
+	// Note: This should only be used by fetch context providers, standard context
+	// providers should use Set to update the overall state.
+	Signal()
+
 	// Set sets the current mapping for this context.
 	Set(map[string]interface{}) error
 }