Wake watchers until the var changes again

1 files changed, 6 insertions, 4 deletions

diff --git a/var.go b/var.go
index 7e50cac..f3b8d40 100644
--- a/var.go
+++ b/var.go
@@ -17,20 +17,22 @@ func (v *Var) changeValue(new interface{}) {
 	newVarValue := old.Set(new)
 	v.value.Store(newVarValue)
 	if old.Changed(newVarValue) {
-		v.wakeWatchers()
+		go v.wakeWatchers(newVarValue)
 	}
 }
 
-func (v *Var) wakeWatchers() {
+func (v *Var) wakeWatchers(new VarValue) {
 	v.watchers.Range(func(k, _ interface{}) bool {
 		tx := k.(*Tx)
-
 		// We have to lock here to ensure that the Tx is waiting before we signal it. Otherwise we
 		// could signal it before it goes to sleep and it will miss the notification.
 		tx.mu.Lock()
 		tx.cond.Broadcast()
+		for !tx.waiting && !tx.completed {
+			tx.cond.Wait()
+		}
 		tx.mu.Unlock()
-		return true
+		return !v.value.Load().(VarValue).Changed(new)
 	})
 }