fix onchanged listener

meta.go

@@ -26,9 +26,9 @@ func NewMetadata() *Metadata {
 
 func (m *Metadata) Set(key string, value interface{}) {
 	m.mu.Lock()
-	defer m.mu.Unlock()
-
 	m.m[key] = value
+	m.mu.Unlock()
+
 	m.onChanged(key, value)
 }
 
@@ -44,9 +44,9 @@ func (m *Metadata) Get(key string) (interface{}, error) {
 
 func (m *Metadata) Delete(key string) {
 	m.mu.Lock()
-	defer m.mu.Unlock()
-
 	delete(m.m, key)
+	m.mu.Unlock()
+
 	m.onChanged(key, nil)
 }
 
@@ -67,17 +67,17 @@ func (m *Metadata) onChanged(key string, value interface{}) {
 
 func (m *Metadata) OnChanged(ctx context.Context, f func(key string, value interface{})) {
 	m.mu.Lock()
-	defer m.mu.Unlock()
-
 	nextIdx := len(m.onChangedCallbacks)
 	m.onChangedCallbacks = append(m.onChangedCallbacks, f)
+	m.mu.Unlock()
 
 	go func() {
 		localCtx, cancel := context.WithCancel(ctx)
 		defer cancel()
 
 		<-localCtx.Done()
-
+		m.mu.Lock()
+		defer m.mu.Unlock()
 		for i, _ := range m.onChangedCallbacks {
 			if nextIdx == i {
 				m.onChangedCallbacks = append(m.onChangedCallbacks[:i], m.onChangedCallbacks[i+1:]...)

meta_test.go

@@ -42,7 +42,7 @@ func TestMetadata(t *testing.T) {
 
 	// Test OnChanged method
 	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+
 	ch := make(chan struct{})
 	m.OnChanged(ctx, func(key string, value interface{}) {
 		t.Logf("Key: %s, Value: %v", key, value)
@@ -53,4 +53,22 @@ func TestMetadata(t *testing.T) {
 	}()
 
 	<-ch
+
+	// cancel the listener above
+	cancel()
+
+	// Test OnChanged method with cancel
+	var state = true
+	ctx1, cancel1 := context.WithCancel(context.Background())
+	m.OnChanged(ctx1, func(key string, value interface{}) {
+		t.Logf("Key: %s, Value: %v", key, value)
+		state = false
+	})
+
+	cancel1()
+
+	m.Set("key6", "value6")
+
+	require.Equal(t, true, state)
+
 }