Merge pull request #92 from mutablelogic/v5

Marshaler fixes

Author: djthorpe

go.mod

@@ -4,7 +4,7 @@ go 1.23.5
 
 require (
 	github.com/alecthomas/kong v1.10.0
-	github.com/djthorpe/go-marshaler v0.0.15
+	github.com/djthorpe/go-marshaler v1.0.0
 	github.com/djthorpe/go-pg v1.0.5
 	github.com/golang-jwt/jwt/v5 v5.2.2
 	github.com/mutablelogic/go-client v1.0.12

pkg/pgqueue/config/task.go

@@ -3,6 +3,7 @@ package config
 import (
 	"context"
 	"errors"
+	"reflect"
 	"strings"
 	"sync"
 	"time"
@@ -29,6 +30,7 @@ type task struct {
 }
 
 var _ server.Task = (*task)(nil)
+var _ server.PGQueue = (*task)(nil)
 
 ////////////////////////////////////////////////////////////////////////////////
 // LIFECYCLE
@@ -38,7 +40,13 @@ func NewTask(manager *pgqueue.Manager, threads uint) (server.Task, error) {
 	self.manager = manager
 	self.taskpool = pgqueue.NewTaskPool(threads)
 	self.callbacks = make(map[string]server.PGCallback, 100)
-	self.decoder = marshaler.NewDecoder("json", marshaler.ConvertTime, marshaler.ConvertDuration, marshaler.ConvertIntUint)
+	self.decoder = marshaler.NewDecoder("json",
+		convertPtr,
+		convertFloatToIntUint,
+		marshaler.ConvertTime,
+		marshaler.ConvertDuration,
+		marshaler.ConvertIntUint,
+	)
 	return self, nil
 }
 
@@ -154,6 +162,11 @@ FOR_LOOP:
 ////////////////////////////////////////////////////////////////////////////////
 // PUBLIC METHODS
 
+// Conn returns the underlying connection pool object.
+func (t *task) Conn() pg.PoolConn {
+	return t.manager.Conn()
+}
+
 // RegisterTicker registers a periodic task (ticker) with a callback function.
 // It returns the metadata of the registered ticker.
 func (t *task) RegisterTicker(ctx context.Context, meta schema.TickerMeta, fn server.PGCallback) (*schema.Ticker, error) {
@@ -283,3 +296,53 @@ func joinName(parts ...string) string {
 func splitName(name string, n int) []string {
 	return strings.SplitN(name, namespaceSeparator, n)
 }
+
+// //////////////////////////////////////////////////////////////////////////////
+// PRIVATE METHODS
+var (
+	nilValue = reflect.ValueOf(nil)
+)
+
+// convertPtr returns value if pointer
+func convertPtr(src reflect.Value, dest reflect.Type) (reflect.Value, error) {
+	// Pass value through
+	if src.Type() == dest {
+		return src, nil
+	}
+
+	// Convert src to elem
+	if dest.Kind() == reflect.Ptr {
+		if dest.Elem() == src.Type() {
+			if src.CanAddr() {
+				return src.Addr(), nil
+			} else {
+				src_ := reflect.New(dest.Elem())
+				src_.Elem().Set(src)
+				return src_, nil
+			}
+		}
+	}
+
+	// Skip
+	return nilValue, nil
+}
+
+// convert float types to int or uint
+func convertFloatToIntUint(src reflect.Value, dest reflect.Type) (reflect.Value, error) {
+	// Pass value through
+	if src.Type() == dest {
+		return src, nil
+	}
+	switch dest.Kind() {
+	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
+		if src.Kind() == reflect.Float64 || src.Kind() == reflect.Float32 {
+			return reflect.ValueOf(int64(src.Float())), nil
+		}
+	case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
+		if src.Kind() == reflect.Float64 || src.Kind() == reflect.Float32 {
+			return reflect.ValueOf(uint64(src.Float())), nil
+		}
+	}
+	// Skip
+	return nilValue, nil
+}

pkg/pgqueue/manager.go

@@ -90,6 +90,10 @@ func (manager *Manager) Worker() string {
 	return manager.worker
 }
 
+func (manager *Manager) Conn() pg.PoolConn {
+	return manager.conn
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 // PUBLIC METHODS - TICKER
 

pkg/provider/meta/meta.go

@@ -168,7 +168,7 @@ func newMetaField(rf reflect.StructField) (*Meta, error) {
 
 var (
 	timeType     = reflect.TypeOf(time.Time{})
-	urlType      = reflect.TypeOf((*url.URL)(nil)).Elem()
+	urlType      = reflect.TypeOf((*url.URL)(nil))
 	durationType = reflect.TypeOf(time.Duration(0))
 )
 
@@ -219,8 +219,16 @@ func setValue(rv reflect.Value, str string) error {
 	case reflect.String:
 		// String
 		rv.SetString(str)
+	case reflect.Ptr:
+		if rv.Type() == urlType {
+			if v, err := url.Parse(str); err == nil {
+				rv.Set(reflect.ValueOf(v))
+				return nil
+			}
+		}
+		fallthrough
 	default:
-		// TODO URL and Datetime
+		// TODO: Datetime
 		return httpresponse.ErrBadRequest.Withf("invalid value for %s: %q", rv.Type(), str)
 	}
 
@@ -254,10 +262,11 @@ func typeName(rt reflect.Type) string {
 		if subtype := typeName(rt.Elem()); subtype != "" && rt.Key().Kind() == reflect.String {
 			return "map(" + subtype + ")"
 		}
-	case reflect.Struct:
+	case reflect.Ptr:
 		if rt == urlType {
 			return "url"
 		}
+	case reflect.Struct:
 		if rt == timeType {
 			return "datetime"
 		}

plugin.go

@@ -116,6 +116,9 @@ type PGCallback func(context.Context, any) error
 
 // PGQueue defines methods for interacting with a PostgreSQL-backed task queue.
 type PGQueue interface {
+	// Conn returns the underlying connection pool object.
+	Conn() pg.PoolConn
+
 	// RegisterTicker registers a periodic task (ticker) with a callback function.
 	// It returns the metadata of the registered ticker.
 	RegisterTicker(context.Context, pgschema.TickerMeta, PGCallback) (*pgschema.Ticker, error)