v0.0.24

syncext/atomic.go

@@ -1,16 +1,21 @@
 package dataext
 
-import "sync/atomic"
+import (
+	"context"
+	"sync/atomic"
+	"time"
+)
 
 type AtomicBool struct {
-	v int32
+	v      int32
+	waiter chan bool // unbuffered
 }
 
 func NewAtomicBool(value bool) *AtomicBool {
 	if value {
-		return &AtomicBool{v: 0}
+		return &AtomicBool{v: 0, waiter: make(chan bool)}
 	} else {
-		return &AtomicBool{v: 1}
+		return &AtomicBool{v: 1, waiter: make(chan bool)}
 	}
 }
 
@@ -24,4 +29,70 @@ func (a *AtomicBool) Set(value bool) {
 	} else {
 		atomic.StoreInt32(&a.v, 0)
 	}
+
+	select {
+	case a.waiter <- value:
+		// message sent
+	default:
+		// no receiver on channel
+	}
+}
+
+func (a *AtomicBool) Wait(waitFor bool) {
+	if a.Get() == waitFor {
+		return
+	}
+
+	for {
+		if v, ok := ReadChannelWithTimeout(a.waiter, 128*time.Millisecond); ok {
+			if v == waitFor {
+				return
+			}
+		} else {
+			if a.Get() == waitFor {
+				return
+			}
+		}
+	}
+}
+
+func (a *AtomicBool) WaitWithContext(ctx context.Context, waitFor bool) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+
+	if a.Get() == waitFor {
+		return nil
+	}
+
+	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+
+		timeOut := 128 * time.Millisecond
+
+		if dl, ok := ctx.Deadline(); ok {
+			timeOutMax := dl.Sub(time.Now())
+			if timeOutMax <= 0 {
+				timeOut = 0
+			} else if 0 < timeOutMax && timeOutMax < timeOut {
+				timeOut = timeOutMax
+			}
+		}
+
+		if v, ok := ReadChannelWithTimeout(a.waiter, timeOut); ok {
+			if v == waitFor {
+				return nil
+			}
+		} else {
+			if err := ctx.Err(); err != nil {
+				return err
+			}
+
+			if a.Get() == waitFor {
+				return nil
+			}
+		}
+	}
 }