v0.0.608

.idea/copilot.data.migration.ask.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="AskMigrationStateService">
+    <option name="migrationStatus" value="COMPLETED" />
+  </component>
+</project>

.idea/copilot.data.migration.ask2agent.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="Ask2AgentMigrationStateService">
+    <option name="migrationStatus" value="COMPLETED" />
+  </component>
+</project>

bfcodegen/enum-generate.template

@@ -117,6 +117,20 @@ func (e {{.EnumTypeName}}) DescriptionMeta() enums.EnumDescriptionMetaValue {
 }
 {{end}}
 
+{{if $hasData}}
+func (e {{.EnumTypeName}}) DataMeta() enums.EnumDataMetaValue {
+    return enums.EnumDataMetaValue{
+        VarName: e.VarName(),
+        Value: e,
+        {{if $hasDescr}} Description: langext.Ptr(e.Description()), {{else}} Description: nil, {{end}}
+        Data: map[string]any{
+            {{ range $datakey, $datatype := $enumdef | generalDataKeys  }} "{{ $datakey }}": e.Data().{{ $datakey | godatakey }},
+            {{ end }}
+        },
+    }
+}
+{{end}}
+
 func Parse{{.EnumTypeName}}(vv string) ({{.EnumTypeName}}, bool) {
 	for _, ev := range __{{.EnumTypeName}}Values {
 		if string(ev) == vv {
@@ -136,6 +150,12 @@ func {{.EnumTypeName}}ValuesMeta() []enums.EnumMetaValue {
 	}
 }
 
+func {{.EnumTypeName}}ValuesDataMeta() []enums.EnumMetaValue {
+	return []enums.EnumDataMetaValue{ {{range .Values}}
+            {{.VarName}}.DataMeta(), {{end}}
+	}
+}
+
 {{if $hasDescr}}
 func {{.EnumTypeName}}ValuesDescriptionMeta() []enums.EnumDescriptionMetaValue {
 	return []enums.EnumDescriptionMetaValue{ {{range .Values}}

dataext/broadcaster.go

@@ -0,0 +1,230 @@
+package dataext
+
+import (
+	"context"
+	"iter"
+	"sync"
+	"time"
+
+	"git.blackforestbytes.com/BlackForestBytes/goext/langext"
+	"git.blackforestbytes.com/BlackForestBytes/goext/syncext"
+	"github.com/rs/xid"
+)
+
+// Broadcaster is a simple Broadcaster channel
+// This is a simpler interface over Broadcaster - which does not have distinct namespaces
+type Broadcaster[TData any] struct {
+	masterLock *sync.Mutex
+
+	subscriptions []*broadcastSubscription[TData]
+}
+
+type BroadcastSubscription interface {
+	Unsubscribe()
+}
+
+type broadcastSubscription[TData any] struct {
+	ID string
+
+	parent *Broadcaster[TData]
+
+	subLock *sync.Mutex
+
+	Func func(TData)
+	Chan chan TData
+
+	UnsubChan chan bool
+}
+
+func (p *broadcastSubscription[TData]) Unsubscribe() {
+	p.parent.unsubscribe(p)
+}
+
+func NewBroadcaster[TData any](capacity int) *Broadcaster[TData] {
+	return &Broadcaster[TData]{
+		masterLock:    &sync.Mutex{},
+		subscriptions: make([]*broadcastSubscription[TData], 0, capacity),
+	}
+}
+
+func (bb *Broadcaster[TData]) SubscriberCount() int {
+	bb.masterLock.Lock()
+	defer bb.masterLock.Unlock()
+
+	return len(bb.subscriptions)
+}
+
+// Publish sends `data` to all subscriber
+// But unbuffered - if one is currently not listening, we skip (the actualReceiver < subscriber)
+func (bb *Broadcaster[TData]) Publish(data TData) (subscriber int, actualReceiver int) {
+	bb.masterLock.Lock()
+	subs := langext.ArrCopy(bb.subscriptions)
+	bb.masterLock.Unlock()
+
+	subscriber = len(subs)
+	actualReceiver = 0
+
+	for _, sub := range subs {
+		func() {
+			sub.subLock.Lock()
+			defer sub.subLock.Unlock()
+
+			if sub.Func != nil {
+				go func() { sub.Func(data) }()
+				actualReceiver++
+			} else if sub.Chan != nil {
+				msgSent := syncext.WriteNonBlocking(sub.Chan, data)
+				if msgSent {
+					actualReceiver++
+				}
+			}
+		}()
+	}
+
+	return subscriber, actualReceiver
+}
+
+// PublishWithContext sends `data` to all subscriber
+// buffered - if one is currently not listening, we wait (but error out when the context runs out)
+func (bb *Broadcaster[TData]) PublishWithContext(ctx context.Context, data TData) (subscriber int, actualReceiver int, err error) {
+	bb.masterLock.Lock()
+	subs := langext.ArrCopy(bb.subscriptions)
+	bb.masterLock.Unlock()
+
+	subscriber = len(subs)
+	actualReceiver = 0
+
+	for _, sub := range subs {
+		err := func() error {
+			sub.subLock.Lock()
+			defer sub.subLock.Unlock()
+
+			if err := ctx.Err(); err != nil {
+				return err
+			}
+
+			if sub.Func != nil {
+				go func() { sub.Func(data) }()
+				actualReceiver++
+			} else if sub.Chan != nil {
+				err := syncext.WriteChannelWithContext(ctx, sub.Chan, data)
+				if err != nil {
+					return err
+				}
+				actualReceiver++
+			}
+
+			return nil
+		}()
+		if err != nil {
+			return subscriber, actualReceiver, err
+		}
+	}
+
+	return subscriber, actualReceiver, nil
+}
+
+// PublishWithTimeout sends `data` to all subscriber
+// buffered - if one is currently not listening, we wait (but wait at most `timeout` - if the timeout is exceeded then actualReceiver < subscriber)
+func (bb *Broadcaster[TData]) PublishWithTimeout(data TData, timeout time.Duration) (subscriber int, actualReceiver int) {
+	bb.masterLock.Lock()
+	subs := langext.ArrCopy(bb.subscriptions)
+	bb.masterLock.Unlock()
+
+	subscriber = len(subs)
+	actualReceiver = 0
+
+	for _, sub := range subs {
+		func() {
+			sub.subLock.Lock()
+			defer sub.subLock.Unlock()
+
+			if sub.Func != nil {
+				go func() { sub.Func(data) }()
+				actualReceiver++
+			} else if sub.Chan != nil {
+				ok := syncext.WriteChannelWithTimeout(sub.Chan, data, timeout)
+				if ok {
+					actualReceiver++
+				}
+			}
+		}()
+	}
+
+	return subscriber, actualReceiver
+}
+
+func (bb *Broadcaster[TData]) SubscribeByCallback(fn func(TData)) BroadcastSubscription {
+	bb.masterLock.Lock()
+	defer bb.masterLock.Unlock()
+
+	sub := &broadcastSubscription[TData]{ID: xid.New().String(), parent: bb, subLock: &sync.Mutex{}, Func: fn, UnsubChan: nil}
+
+	bb.subscriptions = append(bb.subscriptions, sub)
+
+	return sub
+}
+
+func (bb *Broadcaster[TData]) SubscribeByChan(chanBufferSize int) (chan TData, BroadcastSubscription) {
+	bb.masterLock.Lock()
+	defer bb.masterLock.Unlock()
+
+	msgCh := make(chan TData, chanBufferSize)
+
+	sub := &broadcastSubscription[TData]{ID: xid.New().String(), parent: bb, subLock: &sync.Mutex{}, Chan: msgCh, UnsubChan: nil}
+
+	bb.subscriptions = append(bb.subscriptions, sub)
+
+	return msgCh, sub
+}
+
+func (bb *Broadcaster[TData]) SubscribeByIter(chanBufferSize int) (iter.Seq[TData], BroadcastSubscription) {
+	bb.masterLock.Lock()
+	defer bb.masterLock.Unlock()
+
+	msgCh := make(chan TData, chanBufferSize)
+	unsubChan := make(chan bool, 8)
+
+	sub := &broadcastSubscription[TData]{ID: xid.New().String(), parent: bb, subLock: &sync.Mutex{}, Chan: msgCh, UnsubChan: unsubChan}
+
+	bb.subscriptions = append(bb.subscriptions, sub)
+
+	iterFun := func(yield func(TData) bool) {
+		for {
+			select {
+			case msg := <-msgCh:
+				if !yield(msg) {
+					sub.Unsubscribe()
+					return
+				}
+			case <-sub.UnsubChan:
+				sub.Unsubscribe()
+				return
+			}
+		}
+	}
+
+	return iterFun, sub
+}
+
+func (bb *Broadcaster[TData]) unsubscribe(p *broadcastSubscription[TData]) {
+	bb.masterLock.Lock()
+	defer bb.masterLock.Unlock()
+
+	p.subLock.Lock()
+	defer p.subLock.Unlock()
+
+	if p.Chan != nil {
+		close(p.Chan)
+		p.Chan = nil
+	}
+	if p.UnsubChan != nil {
+		syncext.WriteNonBlocking(p.UnsubChan, true)
+		close(p.UnsubChan)
+		p.UnsubChan = nil
+	}
+
+	bb.subscriptions = langext.ArrFilter(bb.subscriptions, func(v *broadcastSubscription[TData]) bool {
+		return v.ID != p.ID
+	})
+}

dataext/broadcaster_test.go

@@ -0,0 +1,332 @@
+package dataext
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestNewBroadcast(t *testing.T) {
+	bb := NewBroadcaster[string](10)
+	if bb == nil {
+		t.Fatal("NewBroadcaster returned nil")
+	}
+	if bb.masterLock == nil {
+		t.Fatal("masterLock is nil")
+	}
+	if bb.subscriptions == nil {
+		t.Fatal("subscriptions is nil")
+	}
+}
+
+func TestBroadcast_SubscribeByCallback(t *testing.T) {
+	bb := NewBroadcaster[string](10)
+
+	var received string
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	callback := func(msg string) {
+		received = msg
+		wg.Done()
+	}
+
+	sub := bb.SubscribeByCallback(callback)
+	defer sub.Unsubscribe()
+
+	// Publish a message
+	subs, receivers := bb.Publish("hello")
+
+	if subs != 1 {
+		t.Fatalf("Expected 1 subscriber, got %d", subs)
+	}
+
+	if receivers != 1 {
+		t.Fatalf("Expected 1 receiver, got %d", receivers)
+	}
+
+	// Wait for the callback to be executed
+	wg.Wait()
+
+	if received != "hello" {
+		t.Fatalf("Expected to receive 'hello', got '%s'", received)
+	}
+}
+
+func TestBroadcast_SubscribeByChan(t *testing.T) {
+	bb := NewBroadcaster[string](10)
+
+	ch, sub := bb.SubscribeByChan(1)
+	defer sub.Unsubscribe()
+
+	// Publish a message
+	subs, receivers := bb.Publish("hello")
+
+	if subs != 1 {
+		t.Fatalf("Expected 1 subscriber, got %d", subs)
+	}
+
+	if receivers != 1 {
+		t.Fatalf("Expected 1 receiver, got %d", receivers)
+	}
+
+	// Read from the channel with a timeout to avoid blocking
+	select {
+	case msg := <-ch:
+		if msg != "hello" {
+			t.Fatalf("Expected to receive 'hello', got '%s'", msg)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Timed out waiting for message")
+	}
+}
+
+func TestBroadcast_SubscribeByIter(t *testing.T) {
+	bb := NewBroadcaster[string](10)
+
+	iterSeq, sub := bb.SubscribeByIter(1)
+	defer sub.Unsubscribe()
+
+	// Channel to communicate when message is received
+	done := make(chan bool)
+	received := false
+
+	// Start a goroutine to use the iterator
+	go func() {
+		for msg := range iterSeq {
+			if msg == "hello" {
+				received = true
+				done <- true
+				return // Stop iteration
+			}
+		}
+	}()
+
+	// Give time for the iterator to start
+	time.Sleep(100 * time.Millisecond)
+
+	// Publish a message
+	bb.Publish("hello")
+
+	// Wait for the message to be received or timeout
+	select {
+	case <-done:
+		if !received {
+			t.Fatal("Message was received but not 'hello'")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Timed out waiting for message")
+	}
+
+	subCount := bb.SubscriberCount()
+	if subCount != 0 {
+		t.Fatalf("Expected 0 receivers, got %d", subCount)
+	}
+}
+
+func TestBroadcast_Publish(t *testing.T) {
+	bb := NewBroadcaster[string](10)
+
+	// Test publishing with no subscribers
+	subs, receivers := bb.Publish("hello")
+	if subs != 0 {
+		t.Fatalf("Expected 0 subscribers, got %d", subs)
+	}
+	if receivers != 0 {
+		t.Fatalf("Expected 0 receivers, got %d", receivers)
+	}
+
+	// Add a subscriber
+	ch, sub := bb.SubscribeByChan(1)
+	defer sub.Unsubscribe()
+
+	// Publish a message
+	subs, receivers = bb.Publish("hello")
+	if subs != 1 {
+		t.Fatalf("Expected 1 subscriber, got %d", subs)
+	}
+	if receivers != 1 {
+		t.Fatalf("Expected 1 receiver, got %d", receivers)
+	}
+
+	// Verify the message was received
+	select {
+	case msg := <-ch:
+		if msg != "hello" {
+			t.Fatalf("Expected to receive 'hello', got '%s'", msg)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Timed out waiting for message")
+	}
+
+	// Test non-blocking behavior with a full channel
+	// First fill the channel
+	bb.Publish("fill")
+
+	// Now publish again - this should not block but may skip the receiver
+	subs, receivers = bb.Publish("overflow")
+	if subs != 1 {
+		t.Fatalf("Expected 1 subscriber, got %d", subs)
+	}
+	_ = receivers // may be 0 if channel is full
+
+	// Drain the channel
+	<-ch
+}
+
+func TestBroadcast_PublishWithTimeout(t *testing.T) {
+	bb := NewBroadcaster[string](10)
+
+	// Add a subscriber with a channel
+	ch, sub := bb.SubscribeByChan(1)
+	defer sub.Unsubscribe()
+
+	// Publish with a timeout
+	subs, receivers := bb.PublishWithTimeout("hello", 100*time.Millisecond)
+	if subs != 1 {
+		t.Fatalf("Expected 1 subscriber, got %d", subs)
+	}
+	if receivers != 1 {
+		t.Fatalf("Expected 1 receiver, got %d", receivers)
+	}
+
+	// Verify the message was received
+	select {
+	case msg := <-ch:
+		if msg != "hello" {
+			t.Fatalf("Expected to receive 'hello', got '%s'", msg)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Timed out waiting for message")
+	}
+
+	// Fill the channel
+	bb.Publish("fill")
+
+	// Test timeout behavior with a full channel
+	start := time.Now()
+	subs, receivers = bb.PublishWithTimeout("timeout-test", 50*time.Millisecond)
+	elapsed := time.Since(start)
+
+	if subs != 1 {
+		t.Fatalf("Expected 1 subscriber, got %d", subs)
+	}
+
+	// The receiver count should be 0 if the timeout occurred
+	if elapsed < 50*time.Millisecond {
+		t.Fatalf("Expected to wait at least 50ms, only waited %v", elapsed)
+	}
+
+	// Drain the channel
+	<-ch
+}
+
+func TestBroadcast_PublishWithContext(t *testing.T) {
+	bb := NewBroadcaster[string](10)
+
+	// Add a subscriber with a channel
+	ch, sub := bb.SubscribeByChan(1)
+	defer sub.Unsubscribe()
+
+	// Create a context
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	// Publish with context
+	subs, receivers, err := bb.PublishWithContext(ctx, "hello")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if subs != 1 {
+		t.Fatalf("Expected 1 subscriber, got %d", subs)
+	}
+	if receivers != 1 {
+		t.Fatalf("Expected 1 receiver, got %d", receivers)
+	}
+
+	// Verify the message was received
+	select {
+	case msg := <-ch:
+		if msg != "hello" {
+			t.Fatalf("Expected to receive 'hello', got '%s'", msg)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Timed out waiting for message")
+	}
+
+	// Fill the channel
+	bb.Publish("fill")
+
+	// Test context cancellation with a full channel
+	ctx, cancel = context.WithCancel(context.Background())
+
+	// Cancel the context after a short delay
+	go func() {
+		time.Sleep(50 * time.Millisecond)
+		cancel()
+	}()
+
+	start := time.Now()
+	subs, receivers, err = bb.PublishWithContext(ctx, "context-test")
+	elapsed := time.Since(start)
+
+	if subs != 1 {
+		t.Fatalf("Expected 1 subscriber, got %d", subs)
+	}
+
+	// Should get a context canceled error
+	if err == nil {
+		t.Fatal("Expected context canceled error, got nil")
+	}
+
+	if elapsed < 50*time.Millisecond {
+		t.Fatalf("Expected to wait at least 50ms, only waited %v", elapsed)
+	}
+
+	// Drain the channel
+	<-ch
+}
+
+func TestBroadcast_Unsubscribe(t *testing.T) {
+	bb := NewBroadcaster[string](10)
+
+	// Add a subscriber
+	ch, sub := bb.SubscribeByChan(1)
+
+	// Publish a message
+	subs, receivers := bb.Publish("hello")
+	if subs != 1 {
+		t.Fatalf("Expected 1 subscriber, got %d", subs)
+	}
+	if receivers != 1 {
+		t.Fatalf("Expected 1 receiver, got %d", receivers)
+	}
+
+	// Verify the message was received
+	select {
+	case msg := <-ch:
+		if msg != "hello" {
+			t.Fatalf("Expected to receive 'hello', got '%s'", msg)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Timed out waiting for message")
+	}
+
+	// Unsubscribe
+	sub.Unsubscribe()
+
+	// Publish again
+	subs, receivers = bb.Publish("after-unsub")
+	if subs != 0 {
+		t.Fatalf("Expected 0 subscribers after unsubscribe, got %d", subs)
+	}
+	if receivers != 0 {
+		t.Fatalf("Expected 0 receivers after unsubscribe, got %d", receivers)
+	}
+
+	// Check that the subscriber count is 0
+	if bb.SubscriberCount() != 0 {
+		t.Fatalf("Expected SubscriberCount() == 0, got %d", bb.SubscriberCount())
+	}
+}

dataext/delayedCombiningInvoker.go

@@ -2,12 +2,19 @@ package dataext
 
 import (
 	"context"
-	"git.blackforestbytes.com/BlackForestBytes/goext/langext"
-	"git.blackforestbytes.com/BlackForestBytes/goext/syncext"
 	"sync"
 	"time"
+
+	"git.blackforestbytes.com/BlackForestBytes/goext/langext"
+	"git.blackforestbytes.com/BlackForestBytes/goext/syncext"
 )
 
+// DelayedCombiningInvoker is a utility to combine multiple consecutive requests into a single execution
+//
+// Requests are made with Request(), and consecutive requests are combined during the `delay` period.
+//
+// Can be used, e.g., for search-controls, where we want to init the search when teh user stops typing
+// Or generally to queue an execution once a burst of requests is over.
 type DelayedCombiningInvoker struct {
 	syncLock           sync.Mutex
 	triggerChan        chan bool
@@ -17,8 +24,13 @@ type DelayedCombiningInvoker struct {
 	delay              time.Duration
 	maxDelay           time.Duration
 	executorRunning    *syncext.AtomicBool
+	pendingRequests    *syncext.Atomic[int]
 	lastRequestTime    time.Time
 	initialRequestTime time.Time
+
+	onExecutionStart []func(immediately bool)          // listener ( actual execution of action starts )
+	onExecutionDone  []func()                          // listener ( actual execution of action finished )
+	onRequest        []func(pending int, initial bool) // listener ( a request came in, waiting for execution )
 }
 
 func NewDelayedCombiningInvoker(action func(), delay time.Duration, maxDelay time.Duration) *DelayedCombiningInvoker {
@@ -27,11 +39,15 @@ func NewDelayedCombiningInvoker(action func(), delay time.Duration, maxDelay tim
 		delay:              delay,
 		maxDelay:           maxDelay,
 		executorRunning:    syncext.NewAtomicBool(false),
+		pendingRequests:    syncext.NewAtomic[int](0),
 		triggerChan:        make(chan bool),
 		cancelChan:         make(chan bool, 1),
 		execNowChan:        make(chan bool, 1),
 		lastRequestTime:    time.Now(),
 		initialRequestTime: time.Now(),
+		onExecutionStart:   make([]func(bool), 0),
+		onExecutionDone:    make([]func(), 0),
+		onRequest:          make([]func(int, bool), 0),
 	}
 }
 
@@ -43,6 +59,11 @@ func (d *DelayedCombiningInvoker) Request() {
 
 	if d.executorRunning.Get() {
 		d.lastRequestTime = now
+		d.pendingRequests.Update(func(v int) int { return v + 1 })
+
+		for _, fn := range d.onRequest {
+			_ = langext.RunPanicSafe(func() { fn(d.pendingRequests.Get(), true) })
+		}
 
 		d.triggerChan <- true
 	} else {
@@ -50,18 +71,28 @@ func (d *DelayedCombiningInvoker) Request() {
 		d.lastRequestTime = now
 
 		d.executorRunning.Set(true)
+		d.pendingRequests.Set(1)
 		syncext.ReadNonBlocking(d.triggerChan) // clear the channel
 		syncext.ReadNonBlocking(d.cancelChan)  // clear the channel
 		syncext.ReadNonBlocking(d.execNowChan) // clear the channel
+
+		for _, fn := range d.onRequest {
+			_ = langext.RunPanicSafe(func() { fn(d.pendingRequests.Get(), false) })
+		}
+
 		go d.run()
 	}
 }
 
 func (d *DelayedCombiningInvoker) run() {
+
+	needsExecutorRunningCleanup := true
 	defer func() {
+		if needsExecutorRunningCleanup {
 			d.syncLock.Lock()
 			d.executorRunning.Set(false)
 			d.syncLock.Unlock()
+		}
 	}()
 
 	for {
@@ -98,8 +129,25 @@ func (d *DelayedCombiningInvoker) run() {
 			continue
 		}
 
+		d.pendingRequests.Set(0)
+
+		for _, fn := range d.onExecutionStart {
+			_ = langext.RunPanicSafe(func() { fn(immediately) })
+		}
+
+		// =================================================
 		_ = langext.RunPanicSafe(d.action)
+		// =================================================
+
+		d.executorRunning.Set(false) // ensure HasPendingRequests returns fals ein onExecutionDone listener
+		needsExecutorRunningCleanup = false
+
+		for _, fn := range d.onExecutionDone {
+			_ = langext.RunPanicSafe(fn)
+		}
+
 		d.syncLock.Unlock()
+
 		return
 	}
 }
@@ -115,6 +163,10 @@ func (d *DelayedCombiningInvoker) HasPendingRequests() bool {
 	return d.executorRunning.Get()
 }
 
+func (d *DelayedCombiningInvoker) CountPendingRequests() int {
+	return d.pendingRequests.Get()
+}
+
 func (d *DelayedCombiningInvoker) ExecuteNow() bool {
 	d.syncLock.Lock()
 	defer d.syncLock.Unlock()
@@ -130,3 +182,21 @@ func (d *DelayedCombiningInvoker) ExecuteNow() bool {
 func (d *DelayedCombiningInvoker) WaitForCompletion(ctx context.Context) error {
 	return d.executorRunning.WaitWithContext(ctx, false)
 }
+
+func (d *DelayedCombiningInvoker) RegisterOnExecutionStart(fn func(immediately bool)) {
+	d.syncLock.Lock()
+	defer d.syncLock.Unlock()
+	d.onExecutionStart = append(d.onExecutionStart, fn)
+}
+
+func (d *DelayedCombiningInvoker) RegisterOnExecutionDone(fn func()) {
+	d.syncLock.Lock()
+	defer d.syncLock.Unlock()
+	d.onExecutionDone = append(d.onExecutionDone, fn)
+}
+
+func (d *DelayedCombiningInvoker) RegisterOnRequest(fn func(pending int, initial bool)) {
+	d.syncLock.Lock()
+	defer d.syncLock.Unlock()
+	d.onRequest = append(d.onRequest, fn)
+}

enums/enum.go

@@ -1,5 +1,7 @@
 package enums
 
+import "encoding/json"
+
 type Enum interface {
 	Valid() bool
 	ValuesAny() []any
@@ -31,3 +33,25 @@ type EnumDescriptionMetaValue struct {
 	Value       Enum   `json:"value"`
 	Description string `json:"description"`
 }
+
+type EnumDataMetaValue struct {
+	VarName     string  `json:"varName"`
+	Value       Enum    `json:"value"`
+	Description *string `json:"description"`
+
+	Data map[string]any `json:"-"` //handled by MarshalJSON
+}
+
+func (v EnumDataMetaValue) MarshalJSON() ([]byte, error) {
+	m := make(map[string]any, 8)
+
+	for k, dv := range v.Data {
+		m[k] = dv
+	}
+
+	m["varName"] = v.VarName
+	m["value"] = v.Value
+	m["description"] = v.Description
+
+	return json.Marshal(m)
+}

go.mod

@@ -3,7 +3,7 @@ module git.blackforestbytes.com/BlackForestBytes/goext
 go 1.24.0
 
 require (
-	github.com/gin-gonic/gin v1.10.1
+	github.com/gin-gonic/gin v1.11.0
 	github.com/glebarez/go-sqlite v1.22.0 // only needed for tests -.-
 	github.com/jmoiron/sqlx v1.4.0
 	github.com/rs/xid v1.6.0
@@ -34,6 +34,7 @@ require (
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-playground/validator/v10 v10.27.0 // indirect
 	github.com/goccy/go-json v0.10.5 // indirect
+	github.com/goccy/go-yaml v1.18.0 // indirect
 	github.com/golang/snappy v1.0.0 // indirect
 	github.com/google/uuid v1.5.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
@@ -46,6 +47,8 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/montanaflynn/stats v0.7.1 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
+	github.com/quic-go/qpack v0.5.1 // indirect
+	github.com/quic-go/quic-go v0.55.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/ugorji/go/codec v1.3.0 // indirect
@@ -53,10 +56,13 @@ require (
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect
+	go.uber.org/mock v0.6.0 // indirect
 	golang.org/x/arch v0.21.0 // indirect
 	golang.org/x/image v0.31.0 // indirect
+	golang.org/x/mod v0.28.0 // indirect
 	golang.org/x/text v0.29.0 // indirect
-	google.golang.org/protobuf v1.36.9 // indirect
+	golang.org/x/tools v0.37.0 // indirect
+	google.golang.org/protobuf v1.36.10 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	modernc.org/libc v1.37.6 // indirect
 	modernc.org/mathutil v1.6.0 // indirect

go.sum

@@ -80,6 +80,8 @@ github.com/gin-gonic/gin v1.10.0 h1:nTuyha1TYqgedzytsKYqna+DfLos46nTv2ygFy86HFU=
 github.com/gin-gonic/gin v1.10.0/go.mod h1:4PMNQiOhvDRa013RKVbsiNwoyezlm2rm0uX/T7kzp5Y=
 github.com/gin-gonic/gin v1.10.1 h1:T0ujvqyCSqRopADpgPgiTT63DUQVSfojyME59Ei63pQ=
 github.com/gin-gonic/gin v1.10.1/go.mod h1:4PMNQiOhvDRa013RKVbsiNwoyezlm2rm0uX/T7kzp5Y=
+github.com/gin-gonic/gin v1.11.0 h1:OW/6PLjyusp2PPXtyxKHU0RbX6I/l28FTdDlae5ueWk=
+github.com/gin-gonic/gin v1.11.0/go.mod h1:+iq/FyxlGzII0KHiBGjuNn4UNENUlKbGlNmc+W50Dls=
 github.com/glebarez/go-sqlite v1.22.0 h1:uAcMJhaA6r3LHMTFgP0SifzgXg46yJkgxqyuyec+ruQ=
 github.com/glebarez/go-sqlite v1.22.0/go.mod h1:PlBIdHe0+aUEFn+r2/uthrWq4FxbzugL0L8Li6yQJbc=
 github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s=
@@ -108,6 +110,8 @@ github.com/goccy/go-json v0.10.4 h1:JSwxQzIqKfmFX1swYPpUThQZp/Ka4wzJdK0LWVytLPM=
 github.com/goccy/go-json v0.10.4/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
 github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4=
 github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
+github.com/goccy/go-yaml v1.18.0 h1:8W7wMFS12Pcas7KU+VVkaiCng+kG8QiFeFwzFb+rwuw=
+github.com/goccy/go-yaml v1.18.0/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
 github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
@@ -175,6 +179,14 @@ github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/quic-go/qpack v0.5.1 h1:giqksBPnT/HDtZ6VhtFKgoLOWmlyo9Ei6u9PqzIMbhI=
+github.com/quic-go/qpack v0.5.1/go.mod h1:+PC4XFrEskIVkcLzpEkbLqq1uCoxPhQuvK5rH1ZgaEg=
+github.com/quic-go/quic-go v0.54.0 h1:6s1YB9QotYI6Ospeiguknbp2Znb/jZYjZLRXn9kMQBg=
+github.com/quic-go/quic-go v0.54.0/go.mod h1:e68ZEaCdyviluZmy44P6Iey98v/Wfz6HCjQEm+l8zTY=
+github.com/quic-go/quic-go v0.54.1 h1:4ZAWm0AhCb6+hE+l5Q1NAL0iRn/ZrMwqHRGQiFwj2eg=
+github.com/quic-go/quic-go v0.54.1/go.mod h1:e68ZEaCdyviluZmy44P6Iey98v/Wfz6HCjQEm+l8zTY=
+github.com/quic-go/quic-go v0.55.0 h1:zccPQIqYCXDt5NmcEabyYvOnomjs8Tlwl7tISjJh9Mk=
+github.com/quic-go/quic-go v0.55.0/go.mod h1:DR51ilwU1uE164KuWXhinFcKWGlEjzys2l8zUl5Ss1U=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
@@ -224,6 +236,8 @@ go.mongodb.org/mongo-driver v1.17.3 h1:TQyXhnsWfWtgAhMtOgtYHMTkZIfBTpMTsMnd9ZBeH
 go.mongodb.org/mongo-driver v1.17.3/go.mod h1:Hy04i7O2kC4RS06ZrhPRqj/u4DTYkFDAAccj+rVKqgQ=
 go.mongodb.org/mongo-driver v1.17.4 h1:jUorfmVzljjr0FLzYQsGP8cgN/qzzxlY9Vh0C9KFXVw=
 go.mongodb.org/mongo-driver v1.17.4/go.mod h1:Hy04i7O2kC4RS06ZrhPRqj/u4DTYkFDAAccj+rVKqgQ=
+go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y=
+go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU=
 golang.org/x/arch v0.11.0 h1:KXV8WWKCXm6tRpLirl2szsO5j/oOODwZf4hATmGVNs4=
 golang.org/x/arch v0.11.0/go.mod h1:FEVrYAQjsQXMVJ1nsMoVVXPZg6p2JE2mx8psSWTDQys=
 golang.org/x/arch v0.12.0 h1:UsYJhbzPYGsT0HbEdmYcqtCv8UNGvnaL561NnIUvaKg=
@@ -301,6 +315,8 @@ golang.org/x/image v0.30.0/go.mod h1:SAEUTxCCMWSrJcCy/4HwavEsfZZJlYxeHLc6tTiAe/c
 golang.org/x/image v0.31.0 h1:mLChjE2MV6g1S7oqbXC0/UcKijjm5fnJLUYKIYrLESA=
 golang.org/x/image v0.31.0/go.mod h1:R9ec5Lcp96v9FTF+ajwaH3uGxPH4fKfHHAVbUILxghA=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+golang.org/x/mod v0.28.0 h1:gQBtGhjxykdjY9YhZpSlZIsbnaE2+PgjfLWUQTnoZ1U=
+golang.org/x/mod v0.28.0/go.mod h1:yfB/L0NOf/kmEbXjzCPOx1iK1fRutOydrCMsqRhEBxI=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
@@ -438,6 +454,8 @@ golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+golang.org/x/tools v0.37.0 h1:DVSRzp7FwePZW356yEAChSdNcQo6Nsp+fex1SUW09lE=
+golang.org/x/tools v0.37.0/go.mod h1:MBN5QPQtLMHVdvsbtarmTNukZDdgwdwlO5qGacAzF0w=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
 google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
@@ -461,6 +479,8 @@ google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyM
 google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
 google.golang.org/protobuf v1.36.9 h1:w2gp2mA27hUeUzj9Ex9FBjsBm40zfaDtEWow293U7Iw=
 google.golang.org/protobuf v1.36.9/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
+google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
+google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

goextVersion.go

@@ -1,5 +1,5 @@
 package goext
 
-const GoextVersion = "0.0.598"
+const GoextVersion = "0.0.608"
 
-const GoextVersionTimestamp = "2025-09-11T11:17:34+0200"
+const GoextVersionTimestamp = "2025-10-04T00:27:41+0200"

gojson/gionic.go

@@ -27,11 +27,11 @@ func (r GoJsonRender) Render(w http.ResponseWriter) error {
 	}
 	jsonBytes, err := MarshalSafeCollections(r.Data, r.NilSafeSlices, r.NilSafeMaps, r.Indent, r.Filter)
 	if err != nil {
-		panic(err)
+		return err
 	}
 	_, err = w.Write(jsonBytes)
 	if err != nil {
-		panic(err)
+		return err
 	}
 	return nil
 }
@@ -39,7 +39,7 @@ func (r GoJsonRender) Render(w http.ResponseWriter) error {
 func (r GoJsonRender) RenderString() (string, error) {
 	jsonBytes, err := MarshalSafeCollections(r.Data, r.NilSafeSlices, r.NilSafeMaps, r.Indent, r.Filter)
 	if err != nil {
-		panic(err)
+		return "", err
 	}
 	return string(jsonBytes), nil
 }

rfctime/rfc3339Nano_test.go

@@ -2,10 +2,11 @@ package rfctime
 
 import (
 	"encoding/json"
-	"git.blackforestbytes.com/BlackForestBytes/goext/timeext"
-	"git.blackforestbytes.com/BlackForestBytes/goext/tst"
 	"testing"
 	"time"
+
+	"git.blackforestbytes.com/BlackForestBytes/goext/timeext"
+	"git.blackforestbytes.com/BlackForestBytes/goext/tst"
 )
 
 func TestRoundtrip(t *testing.T) {
@@ -23,8 +24,8 @@ func TestRoundtrip(t *testing.T) {
 	}
 
 	if string(jstr1) != "{\"v\":\"2023-02-09T15:05:56.820915171+01:00\"}" {
-		t.Errorf(string(jstr1))
+		t.Error(string(jstr1))
-		t.Errorf("repr differs")
+		t.Error("repr differs")
 	}
 
 	w2 := Wrap{}

syncext/atomic.go

@@ -1,17 +1,23 @@
 package syncext
 
 import (
+	"context"
 	"sync"
+	"time"
+
+	"git.blackforestbytes.com/BlackForestBytes/goext/langext"
 )
 
-type Atomic[T any] struct {
+type Atomic[T comparable] struct {
 	v        T
 	lock     sync.RWMutex
+	listener map[string]chan T
 }
 
-func NewAtomic[T any](value T) *Atomic[T] {
+func NewAtomic[T comparable](value T) *Atomic[T] {
 	return &Atomic[T]{
 		v:        value,
+		listener: make(map[string]chan T),
 		lock:     sync.RWMutex{},
 	}
 }
@@ -26,9 +32,138 @@ func (a *Atomic[T]) Set(value T) T {
 	a.lock.Lock()
 	defer a.lock.Unlock()
 
+	return a.setInternal(value)
+}
+
+func (a *Atomic[T]) setInternal(value T) T {
+	// not locked !!
+	// only call from locked context
+
 	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 *Atomic[T]) WaitForChange() chan T {
+	outChan := make(chan T)
+	inChan := make(chan T)
+
+	uuid, _ := langext.NewHexUUID()
+
+	a.lock.Lock()
+	a.listener[uuid] = inChan
+	a.lock.Unlock()
+
+	go func() {
+
+		v := <-inChan
+
+		a.lock.Lock()
+		delete(a.listener, uuid)
+		a.lock.Unlock()
+
+		outChan <- v
+
+		close(outChan)
+	}()
+
+	return outChan
+}
+
+func (a *Atomic[T]) Wait(waitFor T) {
+	_ = a.WaitWithContext(context.Background(), waitFor)
+}
+
+func (a *Atomic[T]) WaitWithTimeout(timeout time.Duration, waitFor T) error {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	return a.WaitWithContext(ctx, waitFor)
+}
+
+func (a *Atomic[T]) WaitWithContext(ctx context.Context, waitFor T) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+
+	if a.Get() == waitFor {
+		return nil
+	}
+
+	uuid, _ := langext.NewHexUUID()
+
+	waitchan := make(chan T)
+
+	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
+			}
+		}
+	}
+}
+
+func (a *Atomic[T]) Update(fn func(old T) T) {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+
+	oldValue := a.v
+	newValue := fn(oldValue)
+
+	a.setInternal(newValue)
+}
+
+func (a *Atomic[T]) CompareAndSwap(old, new T) bool {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+
+	if a.v == old {
+		a.setInternal(new)
+		return true
+	} else {
+		return false
+	}
+}