goext/syncext/atomic.go

package syncext

import (
	"context"
	"sync/atomic"
	"time"
)

type AtomicBool struct {
	v      int32
	waiter chan bool // unbuffered
}

func NewAtomicBool(value bool) *AtomicBool {
	if value {
		return &AtomicBool{v: 0, waiter: make(chan bool)}
	} else {
		return &AtomicBool{v: 1, waiter: make(chan bool)}
	}
}

func (a *AtomicBool) Get() bool {
	return atomic.LoadInt32(&a.v) == 1
}

func (a *AtomicBool) Set(value bool) {
	if value {
		atomic.StoreInt32(&a.v, 1)
	} 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) WaitWithTimeout(timeout time.Duration, waitFor bool) error {
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	return a.WaitWithContext(ctx, waitFor)
}

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
			}
		}
	}
}