v0.0.594 Add exerr OutputRaw(http.ResponseWriter) method

dataext/pubsub.go

@@ -10,49 +10,52 @@ import (
 	"time"
 )
 
+// PubSub is a simple Pub/Sub Broker
+// Clients can subscribe to a namespace and receive published messages on this namespace
+// Messages are broadcast to all subscribers
+type PubSub[TNamespace comparable, TData any] struct {
+	masterLock *sync.Mutex
+
+	subscriptions map[TNamespace][]*pubSubSubscription[TNamespace, TData]
+}
+
 type PubSubSubscription interface {
 	Unsubscribe()
 }
 
-type pubSubSubscription[T any] struct {
+type pubSubSubscription[TNamespace comparable, TData any] struct {
 	ID string
 
-	parent    *PubSub[T]
-	namespace string
+	parent    *PubSub[TNamespace, TData]
+	namespace TNamespace
 
 	subLock *sync.Mutex
 
-	Func func(T)
-	Chan chan T
+	Func func(TData)
+	Chan chan TData
 
 	UnsubChan chan bool
 }
 
-func (p *pubSubSubscription[T]) Unsubscribe() {
+func (p *pubSubSubscription[TNamespace, TData]) Unsubscribe() {
 	p.parent.unsubscribe(p)
 }
 
-type PubSub[T any] struct {
-	masterLock *sync.Mutex
-
-	subscriptions map[string][]*pubSubSubscription[T]
-}
-
-func NewPubSub[T any](capacity int) *PubSub[T] {
-	return &PubSub[T]{
+func NewPubSub[TNamespace comparable, TData any](capacity int) *PubSub[TNamespace, TData] {
+	return &PubSub[TNamespace, TData]{
 		masterLock:    &sync.Mutex{},
-		subscriptions: make(map[string][]*pubSubSubscription[T], capacity),
+		subscriptions: make(map[TNamespace][]*pubSubSubscription[TNamespace, TData], capacity),
 	}
 }
 
-func (ps *PubSub[T]) Namespaces() []string {
+func (ps *PubSub[TNamespace, TData]) Namespaces() []TNamespace {
 	ps.masterLock.Lock()
 	defer ps.masterLock.Unlock()
 
 	return langext.MapKeyArr(ps.subscriptions)
 }
 
-func (ps *PubSub[T]) SubscriberCount(ns string) int {
+func (ps *PubSub[TNamespace, TData]) SubscriberCount(ns TNamespace) int {
 	ps.masterLock.Lock()
 	defer ps.masterLock.Unlock()
 
@@ -61,7 +64,7 @@ func (ps *PubSub[T]) SubscriberCount(ns string) int {
 
 // Publish sends `data` to all subscriber
 // But unbuffered - if one is currently not listening, we skip (the actualReceiver < subscriber)
-func (ps *PubSub[T]) Publish(ns string, data T) (subscriber int, actualReceiver int) {
+func (ps *PubSub[TNamespace, TData]) Publish(ns TNamespace, data TData) (subscriber int, actualReceiver int) {
 	ps.masterLock.Lock()
 	subs := langext.ArrCopy(ps.subscriptions[ns])
 	ps.masterLock.Unlock()
@@ -91,7 +94,7 @@ func (ps *PubSub[T]) Publish(ns string, data T) (subscriber int, 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 (ps *PubSub[T]) PublishWithContext(ctx context.Context, ns string, data T) (subscriber int, actualReceiver int, err error) {
+func (ps *PubSub[TNamespace, TData]) PublishWithContext(ctx context.Context, ns TNamespace, data TData) (subscriber int, actualReceiver int, err error) {
 	ps.masterLock.Lock()
 	subs := langext.ArrCopy(ps.subscriptions[ns])
 	ps.masterLock.Unlock()
@@ -131,7 +134,7 @@ func (ps *PubSub[T]) PublishWithContext(ctx context.Context, ns string, data T)
 
 // 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 (ps *PubSub[T]) PublishWithTimeout(ns string, data T, timeout time.Duration) (subscriber int, actualReceiver int) {
+func (ps *PubSub[TNamespace, TData]) PublishWithTimeout(ns TNamespace, data TData, timeout time.Duration) (subscriber int, actualReceiver int) {
 	ps.masterLock.Lock()
 	subs := langext.ArrCopy(ps.subscriptions[ns])
 	ps.masterLock.Unlock()
@@ -159,42 +162,42 @@ func (ps *PubSub[T]) PublishWithTimeout(ns string, data T, timeout time.Duration
 	return subscriber, actualReceiver
 }
 
-func (ps *PubSub[T]) SubscribeByCallback(ns string, fn func(T)) PubSubSubscription {
+func (ps *PubSub[TNamespace, TData]) SubscribeByCallback(ns TNamespace, fn func(TData)) PubSubSubscription {
 	ps.masterLock.Lock()
 	defer ps.masterLock.Unlock()
 
-	sub := &pubSubSubscription[T]{ID: xid.New().String(), namespace: ns, parent: ps, subLock: &sync.Mutex{}, Func: fn, UnsubChan: nil}
+	sub := &pubSubSubscription[TNamespace, TData]{ID: xid.New().String(), namespace: ns, parent: ps, subLock: &sync.Mutex{}, Func: fn, UnsubChan: nil}
 
 	ps.subscriptions[ns] = append(ps.subscriptions[ns], sub)
 
 	return sub
 }
 
-func (ps *PubSub[T]) SubscribeByChan(ns string, chanBufferSize int) (chan T, PubSubSubscription) {
+func (ps *PubSub[TNamespace, TData]) SubscribeByChan(ns TNamespace, chanBufferSize int) (chan TData, PubSubSubscription) {
 	ps.masterLock.Lock()
 	defer ps.masterLock.Unlock()
 
-	msgCh := make(chan T, chanBufferSize)
+	msgCh := make(chan TData, chanBufferSize)
 
-	sub := &pubSubSubscription[T]{ID: xid.New().String(), namespace: ns, parent: ps, subLock: &sync.Mutex{}, Chan: msgCh, UnsubChan: nil}
+	sub := &pubSubSubscription[TNamespace, TData]{ID: xid.New().String(), namespace: ns, parent: ps, subLock: &sync.Mutex{}, Chan: msgCh, UnsubChan: nil}
 
 	ps.subscriptions[ns] = append(ps.subscriptions[ns], sub)
 
 	return msgCh, sub
 }
 
-func (ps *PubSub[T]) SubscribeByIter(ns string, chanBufferSize int) (iter.Seq[T], PubSubSubscription) {
+func (ps *PubSub[TNamespace, TData]) SubscribeByIter(ns TNamespace, chanBufferSize int) (iter.Seq[TData], PubSubSubscription) {
 	ps.masterLock.Lock()
 	defer ps.masterLock.Unlock()
 
-	msgCh := make(chan T, chanBufferSize)
+	msgCh := make(chan TData, chanBufferSize)
 	unsubChan := make(chan bool, 8)
 
-	sub := &pubSubSubscription[T]{ID: xid.New().String(), namespace: ns, parent: ps, subLock: &sync.Mutex{}, Chan: msgCh, UnsubChan: unsubChan}
+	sub := &pubSubSubscription[TNamespace, TData]{ID: xid.New().String(), namespace: ns, parent: ps, subLock: &sync.Mutex{}, Chan: msgCh, UnsubChan: unsubChan}
 
 	ps.subscriptions[ns] = append(ps.subscriptions[ns], sub)
 
-	iterFun := func(yield func(T) bool) {
+	iterFun := func(yield func(TData) bool) {
 		for {
 			select {
 			case msg := <-msgCh:
@@ -212,7 +215,7 @@ func (ps *PubSub[T]) SubscribeByIter(ns string, chanBufferSize int) (iter.Seq[T]
 	return iterFun, sub
 }
 
-func (ps *PubSub[T]) unsubscribe(p *pubSubSubscription[T]) {
+func (ps *PubSub[TNamespace, TData]) unsubscribe(p *pubSubSubscription[TNamespace, TData]) {
 	ps.masterLock.Lock()
 	defer ps.masterLock.Unlock()
 
@@ -229,7 +232,7 @@ func (ps *PubSub[T]) unsubscribe(p *pubSubSubscription[T]) {
 		p.UnsubChan = nil
 	}
 
-	ps.subscriptions[p.namespace] = langext.ArrFilter(ps.subscriptions[p.namespace], func(v *pubSubSubscription[T]) bool {
+	ps.subscriptions[p.namespace] = langext.ArrFilter(ps.subscriptions[p.namespace], func(v *pubSubSubscription[TNamespace, TData]) bool {
 		return v.ID != p.ID
 	})
 	if len(ps.subscriptions[p.namespace]) == 0 {

dataext/pubsub_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestNewPubSub(t *testing.T) {
-	ps := NewPubSub[string](10)
+	ps := NewPubSub[string, string](10)
 	if ps == nil {
 		t.Fatal("NewPubSub returned nil")
 	}
@@ -21,7 +21,7 @@ func TestNewPubSub(t *testing.T) {
 }
 
 func TestPubSub_Namespaces(t *testing.T) {
-	ps := NewPubSub[string](10)
+	ps := NewPubSub[string, string](10)
 
 	// Initially no namespaces
 	namespaces := ps.Namespaces()
@@ -60,7 +60,7 @@ func TestPubSub_Namespaces(t *testing.T) {
 }
 
 func TestPubSub_SubscribeByCallback(t *testing.T) {
-	ps := NewPubSub[string](10)
+	ps := NewPubSub[string, string](10)
 
 	var received string
 	var wg sync.WaitGroup
@@ -94,7 +94,7 @@ func TestPubSub_SubscribeByCallback(t *testing.T) {
 }
 
 func TestPubSub_SubscribeByChan(t *testing.T) {
-	ps := NewPubSub[string](10)
+	ps := NewPubSub[string, string](10)
 
 	ch, sub := ps.SubscribeByChan("test-ns", 1)
 	defer sub.Unsubscribe()
@@ -122,7 +122,7 @@ func TestPubSub_SubscribeByChan(t *testing.T) {
 }
 
 func TestPubSub_SubscribeByIter(t *testing.T) {
-	ps := NewPubSub[string](10)
+	ps := NewPubSub[string, string](10)
 
 	iterSeq, sub := ps.SubscribeByIter("test-ns", 1)
 	defer sub.Unsubscribe()
@@ -165,7 +165,7 @@ func TestPubSub_SubscribeByIter(t *testing.T) {
 }
 
 func TestPubSub_Publish(t *testing.T) {
-	ps := NewPubSub[string](10)
+	ps := NewPubSub[string, string](10)
 
 	// Test publishing to a namespace with no subscribers
 	subs, receivers := ps.Publish("empty-ns", "hello")
@@ -215,7 +215,7 @@ func TestPubSub_Publish(t *testing.T) {
 }
 
 func TestPubSub_PublishWithTimeout(t *testing.T) {
-	ps := NewPubSub[string](10)
+	ps := NewPubSub[string, string](10)
 
 	// Add a subscriber with a channel
 	ch, sub := ps.SubscribeByChan("test-ns", 1)
@@ -262,7 +262,7 @@ func TestPubSub_PublishWithTimeout(t *testing.T) {
 }
 
 func TestPubSub_PublishWithContext(t *testing.T) {
-	ps := NewPubSub[string](10)
+	ps := NewPubSub[string, string](10)
 
 	// Add a subscriber with a channel
 	ch, sub := ps.SubscribeByChan("test-ns", 1)
@@ -328,7 +328,7 @@ func TestPubSub_PublishWithContext(t *testing.T) {
 }
 
 func TestPubSub_Unsubscribe(t *testing.T) {
-	ps := NewPubSub[string](10)
+	ps := NewPubSub[string, string](10)
 
 	// Add a subscriber
 	ch, sub := ps.SubscribeByChan("test-ns", 1)
@@ -372,7 +372,7 @@ func TestPubSub_Unsubscribe(t *testing.T) {
 }
 
 func TestPubSub_MultipleSubscribers(t *testing.T) {
-	ps := NewPubSub[string](10)
+	ps := NewPubSub[string, string](10)
 
 	// Add multiple subscribers
 	ch1, sub1 := ps.SubscribeByChan("test-ns", 1)

exerr/builder.go

@@ -456,10 +456,19 @@ func (b *Builder) Output(ctx context.Context, g *gin.Context) {
 
 	// this is only here to add one level to the trace
 	// so that .Build() and .Output() and .Print() have the same depth and our stack-skip logger can have the same skip-count
-	b.doOutput(ctx, g)
+	b.doGinOutput(ctx, g)
 }
 
-func (b *Builder) doOutput(ctx context.Context, g *gin.Context) {
+// OutputRaw works teh same as Output() - but does not depend on gin and works with a raw http.ResponseWriter
+func (b *Builder) OutputRaw(w http.ResponseWriter) {
+	warnOnPkgConfigNotInitialized()
+
+	// this is only here to add one level to the trace
+	// so that .Build() and .Output() and .Print() have the same depth and our stack-skip logger can have the same skip-count
+	b.doRawOutput(w)
+}
+
+func (b *Builder) doGinOutput(ctx context.Context, g *gin.Context) {
 	b.errorData.Output(g)
 
 	if (b.errorData.Severity == SevErr || b.errorData.Severity == SevFatal) && (pkgconfig.ZeroLogErrGinOutput || pkgconfig.ZeroLogAllGinOutput) {
@@ -471,6 +480,18 @@ func (b *Builder) doOutput(ctx context.Context, g *gin.Context) {
 	b.errorData.CallListener(MethodOutput, ListenerOpt{NoLog: b.noLog})
 }
 
+func (b *Builder) doRawOutput(w http.ResponseWriter) {
+	b.errorData.OutputRaw(w)
+
+	if (b.errorData.Severity == SevErr || b.errorData.Severity == SevFatal) && (pkgconfig.ZeroLogErrGinOutput || pkgconfig.ZeroLogAllGinOutput) {
+		b.errorData.Log(pkgconfig.ZeroLogger.Error())
+	} else if (b.errorData.Severity == SevWarn) && (pkgconfig.ZeroLogAllGinOutput) {
+		b.errorData.Log(pkgconfig.ZeroLogger.Warn())
+	}
+
+	b.errorData.CallListener(MethodOutput, ListenerOpt{NoLog: b.noLog})
+}
+
 // Print prints the error
 // If the error is SevErr we also send it to the error-service
 func (b *Builder) Print(ctxs ...context.Context) Proxy {

exerr/gin.go

@@ -1,9 +1,9 @@
 package exerr
 
 import (
-	"github.com/gin-gonic/gin"
 	json "git.blackforestbytes.com/BlackForestBytes/goext/gojson"
 	"git.blackforestbytes.com/BlackForestBytes/goext/langext"
+	"github.com/gin-gonic/gin"
 	"net/http"
 	"time"
 )
@@ -68,7 +68,6 @@ func (ee *ExErr) ToDefaultAPIJson() (string, error) {
 	gjr := json.GoJsonRender{Data: ee.ToAPIJson(true, pkgconfig.ExtendedGinOutput, pkgconfig.IncludeMetaInGinOutput), NilSafeSlices: true, NilSafeMaps: true}
 
 	r, err := gjr.RenderString()
-
 	if err != nil {
 		return "", err
 	}
@@ -143,3 +142,34 @@ func (ee *ExErr) Output(g *gin.Context) {
 
 	g.Render(statuscode, json.GoJsonRender{Data: ginOutput, NilSafeSlices: true, NilSafeMaps: true})
 }
+
+func (ee *ExErr) OutputRaw(w http.ResponseWriter) {
+
+	warnOnPkgConfigNotInitialized()
+
+	var statuscode = http.StatusInternalServerError
+
+	var baseCat = ee.RecursiveCategory()
+	var baseType = ee.RecursiveType()
+	var baseStatuscode = ee.RecursiveStatuscode()
+
+	if baseCat == CatUser {
+		statuscode = http.StatusBadRequest
+	} else if baseCat == CatSystem {
+		statuscode = http.StatusInternalServerError
+	}
+
+	if baseStatuscode != nil {
+		statuscode = *ee.StatusCode
+	} else if baseType.DefaultStatusCode != nil {
+		statuscode = *baseType.DefaultStatusCode
+	}
+
+	ginOutput, err := ee.ToDefaultAPIJson()
+	if err != nil {
+		panic(err) // cannot happen
+	}
+
+	w.WriteHeader(statuscode)
+	_, _ = w.Write([]byte(ginOutput))
+}

goextVersion.go

@@ -1,5 +1,5 @@
 package goext
 
-const GoextVersion = "0.0.591"
+const GoextVersion = "0.0.594"
 
-const GoextVersionTimestamp = "2025-07-16T12:44:55+0200"
+const GoextVersionTimestamp = "2025-07-16T17:13:07+0200"