21
0
Fork 0

v0.0.315 atomic
Build Docker and Deploy / Run goext test-suite (push) Failing after 1m26s Details

This commit is contained in:
Mike Schwörer 2023-11-12 03:10:55 +01:00
parent bfe62799d3
commit fa6d73301e
Signed by: Mikescher
GPG Key ID: D3C7172E0A70F8CF
4 changed files with 130 additions and 92 deletions

View File

@ -1,5 +1,5 @@
package goext
const GoextVersion = "0.0.314"
const GoextVersion = "0.0.315"
const GoextVersionTimestamp = "2023-11-10T13:37:54+0100"
const GoextVersionTimestamp = "2023-11-12T03:10:55+0100"

View File

@ -5,6 +5,7 @@ import (
"go.mongodb.org/mongo-driver/bson/bsoncodec"
"go.mongodb.org/mongo-driver/bson/bsontype"
"go.mongodb.org/mongo-driver/bson/primitive"
"gogs.mikescher.com/BlackForestBytes/goext/langext"
"gogs.mikescher.com/BlackForestBytes/goext/rfctime"
"reflect"
)
@ -30,6 +31,9 @@ func CreateGoExtBsonRegistry() *bsoncodec.Registry {
rb.RegisterTypeDecoder(reflect.TypeOf(rfctime.Date{}), rfctime.Date{})
rb.RegisterTypeDecoder(reflect.TypeOf(&rfctime.Date{}), rfctime.Date{})
rb.RegisterTypeDecoder(reflect.TypeOf(rfctime.SecondsF64(0)), rfctime.SecondsF64(0))
rb.RegisterTypeDecoder(reflect.TypeOf(langext.Ptr(rfctime.SecondsF64(0))), rfctime.SecondsF64(0))
bsoncodec.DefaultValueEncoders{}.RegisterDefaultEncoders(rb)
bsoncodec.DefaultValueDecoders{}.RegisterDefaultDecoders(rb)

View File

@ -1,33 +1,28 @@
package syncext
import (
"context"
"gogs.mikescher.com/BlackForestBytes/goext/langext"
"sync"
"time"
)
type AtomicBool struct {
v bool
listener map[string]chan bool
lock sync.Mutex
type Atomic[T any] struct {
v T
lock sync.RWMutex
}
func NewAtomicBool(value bool) *AtomicBool {
return &AtomicBool{
v: value,
listener: make(map[string]chan bool),
lock: sync.Mutex{},
func NewAtomic[T any](value T) *Atomic[T] {
return &Atomic[T]{
v: value,
lock: sync.RWMutex{},
}
}
func (a *AtomicBool) Get() bool {
a.lock.Lock()
defer a.lock.Unlock()
func (a *Atomic[T]) Get() T {
a.lock.RLock()
defer a.lock.RUnlock()
return a.v
}
func (a *AtomicBool) Set(value bool) bool {
func (a *Atomic[T]) Set(value T) T {
a.lock.Lock()
defer a.lock.Unlock()
@ -35,79 +30,5 @@ func (a *AtomicBool) Set(value bool) bool {
a.v = value
for k, v := range a.listener {
select {
case v <- value:
// message sent
default:
// no receiver on channel
delete(a.listener, k)
}
}
return oldValue
}
func (a *AtomicBool) Wait(waitFor bool) {
_ = a.WaitWithContext(context.Background(), waitFor)
}
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
}
uuid, _ := langext.NewHexUUID()
waitchan := make(chan bool)
a.lock.Lock()
a.listener[uuid] = waitchan
a.lock.Unlock()
defer func() {
a.lock.Lock()
delete(a.listener, uuid)
a.lock.Unlock()
}()
for {
if err := ctx.Err(); err != nil {
return err
}
timeOut := 1024 * 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(waitchan, timeOut); ok {
if v == waitFor {
return nil
}
} else {
if err := ctx.Err(); err != nil {
return err
}
if a.Get() == waitFor {
return nil
}
}
}
}

113
syncext/bool.go Normal file
View File

@ -0,0 +1,113 @@
package syncext
import (
"context"
"gogs.mikescher.com/BlackForestBytes/goext/langext"
"sync"
"time"
)
type AtomicBool struct {
v bool
listener map[string]chan bool
lock sync.Mutex
}
func NewAtomicBool(value bool) *AtomicBool {
return &AtomicBool{
v: value,
listener: make(map[string]chan bool),
lock: sync.Mutex{},
}
}
func (a *AtomicBool) Get() bool {
a.lock.Lock()
defer a.lock.Unlock()
return a.v
}
func (a *AtomicBool) Set(value bool) bool {
a.lock.Lock()
defer a.lock.Unlock()
oldValue := a.v
a.v = value
for k, v := range a.listener {
select {
case v <- value:
// message sent
default:
// no receiver on channel
delete(a.listener, k)
}
}
return oldValue
}
func (a *AtomicBool) Wait(waitFor bool) {
_ = a.WaitWithContext(context.Background(), waitFor)
}
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
}
uuid, _ := langext.NewHexUUID()
waitchan := make(chan bool)
a.lock.Lock()
a.listener[uuid] = waitchan
a.lock.Unlock()
defer func() {
a.lock.Lock()
delete(a.listener, uuid)
a.lock.Unlock()
}()
for {
if err := ctx.Err(); err != nil {
return err
}
timeOut := 1024 * 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(waitchan, timeOut); ok {
if v == waitFor {
return nil
}
} else {
if err := ctx.Err(); err != nil {
return err
}
if a.Get() == waitFor {
return nil
}
}
}
}