Apply Code Coverage Badge

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

broker/broker.go

@@ -21,7 +21,7 @@ var (
 	// ErrInvalidMessage returns when invalid Message passed
 	ErrInvalidMessage = errors.New("invalid message")
 	// ErrInvalidHandler returns when subscriber passed to Subscribe
-	ErrInvalidHandler = errors.New("invalid handler")
+	ErrInvalidHandler = errors.New("invalid handler, ony func(Message) error and func([]Message) error supported")
 	// DefaultGracefulTimeout
 	DefaultGracefulTimeout = 5 * time.Second
 )

broker/context.go

@@ -42,6 +42,16 @@ func SetSubscribeOption(k, v interface{}) SubscribeOption {
 	}
 }
 
+// SetPublishOption returns a function to setup a context with given value
+func SetPublishOption(k, v interface{}) PublishOption {
+	return func(o *PublishOptions) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, k, v)
+	}
+}
+
 // SetOption returns a function to setup a context with given value
 func SetOption(k, v interface{}) Option {
 	return func(o *Options) {

broker/options.go

@@ -79,11 +79,15 @@ type PublishOptions struct {
 	// BodyOnly flag says the message contains raw body bytes and don't need
 	// codec Marshal method
 	BodyOnly bool
+	// Context holds custom options
+	Context context.Context
 }
 
 // NewPublishOptions creates PublishOptions struct
 func NewPublishOptions(opts ...PublishOption) PublishOptions {
-	options := PublishOptions{}
+	options := PublishOptions{
+		Context: context.Background(),
+	}
 	for _, o := range opts {
 		o(&options)
 	}

broker/subscriber.go

@@ -1,87 +1,14 @@
 package broker
 
-import (
-	"fmt"
-	"reflect"
-	"unicode"
-	"unicode/utf8"
-)
-
-const (
-	messageSig  = "func(broker.Message) error"
-	messagesSig = "func([]broker.Message) error"
-)
-
-// Precompute the reflect type for error. Can't use error directly
-// because Typeof takes an empty interface value. This is annoying.
-var typeOfError = reflect.TypeOf((*error)(nil)).Elem()
-
-// Is this an exported - upper case - name?
-func isExported(name string) bool {
-	r, _ := utf8.DecodeRuneInString(name)
-	return unicode.IsUpper(r)
-}
-
-// Is this type exported or a builtin?
-func isExportedOrBuiltinType(t reflect.Type) bool {
-	for t.Kind() == reflect.Ptr {
-		t = t.Elem()
-	}
-	// PkgPath will be non-empty even for an exported type,
-	// so we need to check the type name as well.
-	return isExported(t.Name()) || t.PkgPath() == ""
-}
-
 // IsValidHandler func signature
 func IsValidHandler(sub interface{}) error {
-	typ := reflect.TypeOf(sub)
-	var argType reflect.Type
-	switch typ.Kind() {
-	case reflect.Func:
-		name := "Func"
-		switch typ.NumIn() {
-		case 1:
-			argType = typ.In(0)
-		default:
-			return fmt.Errorf("subscriber %v takes wrong number of args: %v required signature %s", name, typ.NumIn(), messageSig)
-		}
-		if !isExportedOrBuiltinType(argType) {
-			return fmt.Errorf("subscriber %v argument type not exported: %v", name, argType)
-		}
-		if typ.NumOut() != 1 {
-			return fmt.Errorf("subscriber %v has wrong number of return values: %v require signature %s",
-				name, typ.NumOut(), messageSig)
-		}
-		if returnType := typ.Out(0); returnType != typeOfError {
-			return fmt.Errorf("subscriber %v returns %v not error", name, returnType.String())
-		}
+	switch sub.(type) {
 	default:
-		hdlr := reflect.ValueOf(sub)
-		name := reflect.Indirect(hdlr).Type().Name()
-
-		for m := 0; m < typ.NumMethod(); m++ {
-			method := typ.Method(m)
-			switch method.Type.NumIn() {
-			case 3:
-				argType = method.Type.In(2)
-			default:
-				return fmt.Errorf("subscriber %v.%v takes wrong number of args: %v required signature %s",
-					name, method.Name, method.Type.NumIn(), messageSig)
-			}
-
-			if !isExportedOrBuiltinType(argType) {
-				return fmt.Errorf("%v argument type not exported: %v", name, argType)
-			}
-			if method.Type.NumOut() != 1 {
-				return fmt.Errorf(
-					"subscriber %v.%v has wrong number of return values: %v require signature %s",
-					name, method.Name, method.Type.NumOut(), messageSig)
-			}
-			if returnType := method.Type.Out(0); returnType != typeOfError {
-				return fmt.Errorf("subscriber %v.%v returns %v not error", name, method.Name, returnType.String())
-			}
-		}
+		return ErrInvalidHandler
+	case func(Message) error:
+		break
+	case func([]Message) error:
+		break
 	}
-
 	return nil
 }

logger/options.go

@@ -2,7 +2,6 @@ package logger
 
 import (
 	"context"
-	"fmt"
 	"io"
 	"log/slog"
 	"os"
@@ -100,11 +99,8 @@ func WithAddFields(fields ...interface{}) Option {
 					iv, iok := o.Fields[i].(string)
 					jv, jok := fields[j].(string)
 					if iok && jok && iv == jv {
-						fmt.Printf("AAAA o.Fields:%v old:%v new:%v\n", o.Fields[i], o.Fields[i+1], fields[j+1])
 						o.Fields[i+1] = fields[j+1]
-						fmt.Printf("BBBB o.Fields:%v old:%v new:%v\n", o.Fields[i], o.Fields[i+1], fields[j+1])
 						fields = slices.Delete(fields, j, j+2)
-
 					}
 				}
 			}

tracer/memory/memory.go

@@ -89,6 +89,10 @@ func (s *Span) Tracer() tracer.Tracer {
 	return s.tracer
 }
 
+func (s *Span) IsRecording() bool {
+	return true
+}
+
 type Event struct {
 	name   string
 	labels []interface{}

tracer/noop.go

@@ -120,6 +120,10 @@ func (s *noopSpan) SetStatus(st SpanStatus, msg string) {
 	s.statusMsg = msg
 }
 
+func (s *noopSpan) IsRecording() bool {
+	return false
+}
+
 // NewTracer returns new memory tracer
 func NewTracer(opts ...Option) Tracer {
 	return &noopTracer{

tracer/tracer.go

@@ -78,4 +78,6 @@ type Span interface {
 	TraceID() string
 	// SpanID returns span id
 	SpanID() string
+	// IsRecording returns the recording state of the Span.
+	IsRecording() bool
 }