Fix unsubscribe

engine.go

@@ -436,13 +436,12 @@ func (e *Engine) Unsubscribe(o chan Reply, id int64) {
 			return
 		}
 
-		newUnObs := []chan<- Reply{}
-		for _, existing := range e.unObservers {
-			if existing != o {
-				newUnObs = append(newUnObs, o)
+		for i, existing := range e.unObservers {
+			if existing == o {
+				e.unObservers = append(e.unObservers[:i], e.unObservers[i+1:]...)
+				break
 			}
 		}
-		e.unObservers = newUnObs
 	})
 	close(terminate)
 }
@@ -460,13 +459,12 @@ func (e *Engine) UnsubscribeAll(o chan Reply) {
 		}
 	}()
 	e.sendCommand(func() {
-		newUnObs := []chan<- Reply{}
-		for _, existing := range e.allObservers {
-			if existing != o {
-				newUnObs = append(newUnObs, o)
+		for i, existing := range e.allObservers {
+			if existing == o {
+				e.allObservers = append(e.allObservers[:i], e.allObservers[i+1:]...)
+				break
 			}
 		}
-		e.allObservers = newUnObs
 	})
 	close(terminate)
 }

engine_test.go

@@ -144,3 +144,28 @@ func logreply(t *testing.T, reply Reply, err error) {
 	}
 	t.Logf("\n")
 }
+
+func TestUnsubscribeAllAndUnmatched(t *testing.T) {
+	engine := NewTestEngine(t)
+	defer engine.ConditionalStop(t)
+	dummy := make(chan Reply)
+	rc := make(chan Reply)
+	engine.SubscribeAll(dummy)
+	engine.SubscribeAll(rc)
+	engine.UnsubscribeAll(rc)
+	for _, v := range engine.allObservers {
+		if v == rc {
+			t.Log("rc should be unsubscribed from allObservers")
+			t.Fail()
+		}
+	}
+	engine.Subscribe(dummy, UnmatchedReplyID)
+	engine.Subscribe(rc, UnmatchedReplyID)
+	engine.Unsubscribe(rc, UnmatchedReplyID)
+	for _, v := range engine.unObservers {
+		if v == rc {
+			t.Log("rc should be unsubscribed from unObservers")
+			t.Fail()
+		}
+	}
+}