Responding to code review comments

Author: aaron-zeisler

testservice/flag_change_listener.go

@@ -18,27 +18,36 @@ type listenerEntry struct {
 // listenerRegistry manages all active flag change listener registrations for a single
 // SDK client entity. It is safe to use from multiple goroutines.
 type listenerRegistry struct {
-	mu      sync.Mutex
-	entries map[string]*listenerEntry // keyed by listenerId
-	tracker interfaces.FlagTracker
+	mu        sync.Mutex
+	listeners map[string]*listenerEntry // keyed by listenerId
+	tracker   interfaces.FlagTracker
 }
 
 func newListenerRegistry(tracker interfaces.FlagTracker) *listenerRegistry {
 	return &listenerRegistry{
-		entries: make(map[string]*listenerEntry),
-		tracker: tracker,
+		listeners: make(map[string]*listenerEntry),
+		tracker:   tracker,
 	}
 }
 
-// registerFlagChangeListener subscribes to general flag configuration changes. If flagKey
-// is non-empty, only events for that specific flag are forwarded to the callback URI.
-func (r *listenerRegistry) registerFlagChangeListener(listenerID, flagKey, callbackURI string) {
-	ch := r.tracker.AddFlagChangeListener()
+// storeListener registers a new listener entry under listenerID, cancelling any
+// previously registered listener with the same ID. Returns the new entry's context.
+func (r *listenerRegistry) storeListener(listenerID string) context.Context {
 	ctx, cancel := context.WithCancel(context.Background())
-
 	r.mu.Lock()
-	r.entries[listenerID] = &listenerEntry{cancel: cancel}
+	if old, exists := r.listeners[listenerID]; exists {
+		old.cancel()
+	}
+	r.listeners[listenerID] = &listenerEntry{cancel: cancel}
 	r.mu.Unlock()
+	return ctx
+}
+
+// registerFlagChangeListener subscribes to general flag configuration changes.
+// All flag change events are forwarded to the callback URI.
+func (r *listenerRegistry) registerFlagChangeListener(listenerID, callbackURI string) {
+	ch := r.tracker.AddFlagChangeListener()
+	ctx := r.storeListener(listenerID)
 
 	svc := callbackService{baseURL: callbackURI}
 	go func() {
@@ -51,9 +60,6 @@ func (r *listenerRegistry) registerFlagChangeListener(listenerID, flagKey, callb
 				if !ok {
 					return
 				}
-				if flagKey != "" && event.Key != flagKey {
-					continue
-				}
 				_ = svc.post("", servicedef.ListenerNotification{
 					ListenerID: listenerID,
 					FlagKey:    event.Key,
@@ -73,11 +79,7 @@ func (r *listenerRegistry) registerFlagValueChangeListener(
 	callbackURI string,
 ) {
 	ch := r.tracker.AddFlagValueChangeListener(flagKey, evalCtx, defaultValue)
-	ctx, cancel := context.WithCancel(context.Background())
-
-	r.mu.Lock()
-	r.entries[listenerID] = &listenerEntry{cancel: cancel}
-	r.mu.Unlock()
+	ctx := r.storeListener(listenerID)
 
 	svc := callbackService{baseURL: callbackURI}
 	go func() {
@@ -107,9 +109,9 @@ func (r *listenerRegistry) registerFlagValueChangeListener(
 // registry. Returns false if no listener with that ID was found.
 func (r *listenerRegistry) unregister(listenerID string) bool {
 	r.mu.Lock()
-	entry, ok := r.entries[listenerID]
+	entry, ok := r.listeners[listenerID]
 	if ok {
-		delete(r.entries, listenerID)
+		delete(r.listeners, listenerID)
 	}
 	r.mu.Unlock()
 
@@ -122,11 +124,11 @@ func (r *listenerRegistry) unregister(listenerID string) bool {
 // closeAll stops all active listener goroutines. Called when the SDK client entity closes.
 func (r *listenerRegistry) closeAll() {
 	r.mu.Lock()
-	entries := r.entries
-	r.entries = make(map[string]*listenerEntry)
+	listeners := r.listeners
+	r.listeners = make(map[string]*listenerEntry)
 	r.mu.Unlock()
 
-	for _, entry := range entries {
+	for _, entry := range listeners {
 		entry.cancel()
 	}
 }

testservice/sdk_client_entity.go

@@ -135,7 +135,7 @@ func (c *SDKClientEntity) DoCommand(params servicedef.CommandParams) (interface{
 		return c.migrationOperation(*params.MigrationOperation)
 	case servicedef.CommandRegisterFlagChangeListener:
 		p := params.RegisterFlagChangeListener
-		c.listeners.registerFlagChangeListener(p.ListenerID, p.FlagKey, p.CallbackURI)
+		c.listeners.registerFlagChangeListener(p.ListenerID, p.CallbackURI)
 		return nil, nil
 	case servicedef.CommandRegisterFlagValueChangeListener:
 		p := params.RegisterFlagValueChangeListener

testservice/service.go

@@ -49,6 +49,7 @@ var capabilities = []string{
 	servicedef.CapabilityPersistentDataStoreConsul,
 	servicedef.CapabilityPersistentDataStoreDynamoDB,
 	servicedef.CapabilityFlagChangeListeners,
+	servicedef.CapabilityFlagValueChangeListeners,
 }
 
 // gets the specified environment variable, or the default if not set

testservice/servicedef/command_params.go

@@ -190,10 +190,9 @@ type HookExecutionTrackPayload struct {
 }
 
 // RegisterFlagChangeListenerParams defines parameters for registering a general flag change listener.
-// FlagKey may be empty to listen for changes to any flag, or non-empty to filter to a specific flag.
+// The listener will be notified whenever any flag's configuration changes.
 type RegisterFlagChangeListenerParams struct {
 	ListenerID  string `json:"listenerId"`
-	FlagKey     string `json:"flagKey"`
 	CallbackURI string `json:"callbackUri"`
 }
 

testservice/servicedef/service_params.go

@@ -30,6 +30,7 @@ const (
 	CapabilityPersistentDataStoreConsul   = "persistent-data-store-consul"
 	CapabilityPersistentDataStoreDynamoDB = "persistent-data-store-dynamodb"
 	CapabilityFlagChangeListeners         = "flag-change-listeners"
+	CapabilityFlagValueChangeListeners    = "flag-value-change-listeners"
 )
 
 type StatusRep struct {