From f11aa2755f62c9696806352058cf8389c6fd25b0 Mon Sep 17 00:00:00 2001 From: Chris Miles Date: Sun, 23 Aug 2026 01:28:30 +0000 Subject: [PATCH] feat: add generic payload decoding helpers --- .github/workflows/test.yml | 2 +- README.md | 38 +++++++- docs/go.mod | 2 +- docs/readme/testcounts/integration_count.json | 2 +- driver/mysqlqueue/go.mod | 2 +- driver/natsqueue/go.mod | 2 +- driver/postgresqueue/go.mod | 2 +- driver/rabbitmqqueue/go.mod | 2 +- driver/redisqueue/go.mod | 2 +- driver/sqlitequeue/go.mod | 2 +- driver/sqlqueuecore/go.mod | 2 +- driver/sqsqueue/go.mod | 2 +- examples/go.mod | 2 +- examples/job-payloadas/main.go | 24 +++++ examples/message-payloadas/main.go | 24 +++++ go.mod | 2 +- go.work | 2 +- integration/go.mod | 2 +- job.go | 18 ++++ payload_as_test.go | 96 +++++++++++++++++++ scripts/module-go-versions.tsv | 24 ++--- workflow_models.go | 18 ++++ 22 files changed, 240 insertions(+), 32 deletions(-) create mode 100644 examples/job-payloadas/main.go create mode 100644 examples/message-payloadas/main.go create mode 100644 payload_as_test.go diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 2aa76b9..3ccedd0 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -101,7 +101,7 @@ jobs: strategy: fail-fast: false matrix: - minimum_go_version: ["1.24.4", "1.25.0"] + minimum_go_version: ["1.27.0"] steps: - name: Checkout uses: actions/checkout@v4 diff --git a/README.md b/README.md index fbf06fc..74de9e8 100644 --- a/README.md +++ b/README.md @@ -10,11 +10,11 @@ Go Reference License: MIT Go Test - Go version + Go version Latest tag - Unit tests (executed count) + Unit tests (executed count) Integration tests (executed count)

@@ -29,7 +29,7 @@ Existing deployments upgrading to application-type direct delivery must replace Applications upgrading from the retired `bus` or `queuefake` packages should follow the [legacy API migration guide](docs/legacy-api-migration.md). The removed Temporal compatibility adapter has no root-package replacement. -The root queue module and non-PostgreSQL driver modules require Go 1.24.4 or newer. The PostgreSQL driver requires Go 1.25 or newer so it can use pgx 5.9.2, the first release containing the [GO-2026-5004 security fix](https://pkg.go.dev/vuln/GO-2026-5004). +Queue and its driver modules require Go 1.27 or newer. Projects on older Go releases can pin the v0.2 release line. The PostgreSQL driver continues to use pgx 5.9.2 or newer, the first release containing the [GO-2026-5004 security fix](https://pkg.go.dev/vuln/GO-2026-5004). ## Quick Start @@ -536,10 +536,10 @@ The API section below is autogenerated; do not edit between the markers. |------:|:-----------| | **Admin** | [CancelJob](#queue-canceljob) · [Queue.CancelJob](#queue-queue-canceljob) · [ClearQueue](#queue-clearqueue) · [Queue.ClearQueue](#queue-queue-clearqueue) · [DeleteJob](#queue-deletejob) · [Queue.DeleteJob](#queue-queue-deletejob) · [History](#queue-queue-history) · [ListJobs](#queue-listjobs) · [Queue.ListJobs](#queue-queue-listjobs) · [Normalize](#queue-listjobsoptions-normalize) · [QueueHistory](#queue-queuehistory) · [RetryJob](#queue-retryjob) · [Queue.RetryJob](#queue-queue-retryjob) · [SinglePointHistory](#queue-singlepointhistory) · [SupportsQueueAdmin](#queue-supportsqueueadmin) · [TimelineHistoryFromSnapshot](#queue-timelinehistoryfromsnapshot) | | **Constructors** | [New](#queue-new) · [NewMemoryStore](#queue-newmemorystore) · [NewMessage](#queue-newmessage) · [NewNull](#queue-newnull) · [NewSQLStore](#queue-newsqlstore) · [NewSQLStoreWithManagedSchema](#queue-newsqlstorewithmanagedschema) · [NewStatsCollector](#queue-newstatscollector) · [NewSync](#queue-newsync) · [NewWorkerpool](#queue-newworkerpool) | -| **Job** | [Backoff](#queue-job-backoff) · [Bind](#queue-job-bind) · [Delay](#queue-job-delay) · [NewJob](#queue-newjob) · [OnQueue](#queue-job-onqueue) · [Payload](#queue-job-payload) · [PayloadBytes](#queue-job-payloadbytes) · [PayloadJSON](#queue-job-payloadjson) · [Retry](#queue-job-retry) · [Timeout](#queue-job-timeout) · [UniqueFor](#queue-job-uniquefor) | +| **Job** | [Backoff](#queue-job-backoff) · [Bind](#queue-job-bind) · [Delay](#queue-job-delay) · [NewJob](#queue-newjob) · [OnQueue](#queue-job-onqueue) · [Payload](#queue-job-payload) · [PayloadAs](#queue-job-payloadas) · [PayloadBytes](#queue-job-payloadbytes) · [PayloadJSON](#queue-job-payloadjson) · [Retry](#queue-job-retry) · [Timeout](#queue-job-timeout) · [UniqueFor](#queue-job-uniquefor) | | **Observability** | [Active](#queue-statssnapshot-active) · [Archived](#queue-statssnapshot-archived) · [Failed](#queue-statssnapshot-failed) · [MultiObserver](#queue-multiobserver) · [ChannelObserver.Observe](#queue-channelobserver-observe) · [Observer.Observe](#queue-observer-observe) · [ObserverFunc.Observe](#queue-observerfunc-observe) · [StatsCollector.Observe](#queue-statscollector-observe) · [Pause](#queue-pause) · [Paused](#queue-statssnapshot-paused) · [Pending](#queue-statssnapshot-pending) · [Processed](#queue-statssnapshot-processed) · [Queue](#queue-statssnapshot-queue) · [Queues](#queue-statssnapshot-queues) · [Ready](#queue-ready) · [Resume](#queue-resume) · [RetryCount](#queue-statssnapshot-retrycount) · [SafeObserve](#queue-safeobserve) · [Scheduled](#queue-statssnapshot-scheduled) · [Snapshot](#queue-snapshot) · [StatsCollector.Snapshot](#queue-statscollector-snapshot) · [SupportsNativeStats](#queue-supportsnativestats) · [SupportsPause](#queue-supportspause) · [Throughput](#queue-statssnapshot-throughput) | | **Other** | [Acquire](#queue-locker-acquire) · [AdvanceChain](#queue-workflowstore-advancechain) · [Allow](#queue-ratelimiter-allow) · [AllowFailures](#queue-batchbuilder-allowfailures) · [CancelBatch](#queue-workflowstore-cancelbatch) · [BatchBuilder.Catch](#queue-batchbuilder-catch) · [ChainBuilder.Catch](#queue-chainbuilder-catch) · [CreateBatch](#queue-workflowstore-createbatch) · [CreateChain](#queue-workflowstore-createchain) · [BatchBuilder.Dispatch](#queue-batchbuilder-dispatch) · [ChainBuilder.Dispatch](#queue-chainbuilder-dispatch) · [FailChain](#queue-workflowstore-failchain) · [FailChainNode](#queue-workflowoutcomestore-failchainnode) · [BatchBuilder.Finally](#queue-batchbuilder-finally) · [ChainBuilder.Finally](#queue-chainbuilder-finally) · [GetBatch](#queue-workflowstore-getbatch) · [GetChain](#queue-workflowstore-getchain) · [FailOnError.Handle](#queue-failonerror-handle) · [Middleware.Handle](#queue-middleware-handle) · [MiddlewareFunc.Handle](#queue-middlewarefunc-handle) · [RateLimit.Handle](#queue-ratelimit-handle) · [RetryPolicy.Handle](#queue-retrypolicy-handle) · [SkipWhen.Handle](#queue-skipwhen-handle) · [WithoutOverlapping.Handle](#queue-withoutoverlapping-handle) · [MarkBatchJobFailed](#queue-workflowstore-markbatchjobfailed) · [MarkBatchJobStarted](#queue-workflowstore-markbatchjobstarted) · [MarkBatchJobSucceeded](#queue-workflowstore-markbatchjobsucceeded) · [MarkCallbackInvoked](#queue-workflowstore-markcallbackinvoked) · [Name](#queue-batchbuilder-name) · [BatchBuilder.OnQueue](#queue-batchbuilder-onqueue) · [ChainBuilder.OnQueue](#queue-chainbuilder-onqueue) · [PhysicalQueueName](#queue-physicalqueuename) · [PhysicalQueueWeights](#queue-physicalqueueweights) · [Progress](#queue-batchbuilder-progress) · [Prune](#queue-workflowstore-prune) · [Release](#queue-lock-release) · [ResolveObservedJobType](#queue-resolveobservedjobtype) · [SettleBatchJob](#queue-workflowoutcomestore-settlebatchjob) · [Then](#queue-batchbuilder-then) | -| **Queue** | [Batch](#queue-queue-batch) · [Bind](#queue-message-bind) · [Chain](#queue-queue-chain) · [Dispatch](#queue-queue-dispatch) · [Driver](#queue-queue-driver) · [FindBatch](#queue-queue-findbatch) · [FindChain](#queue-queue-findchain) · [IsPermanent](#queue-ispermanent) · [Pause](#queue-queue-pause) · [PayloadBytes](#queue-message-payloadbytes) · [Permanent](#queue-permanent) · [Prune](#queue-queue-prune) · [Ready](#queue-queue-ready) · [Register](#queue-queue-register) · [Resume](#queue-queue-resume) · [Run](#queue-queue-run) · [Shutdown](#queue-queue-shutdown) · [StartWorkers](#queue-queue-startworkers) · [Stats](#queue-queue-stats) · [WithClock](#queue-withclock) · [WithContext](#queue-queue-withcontext) · [WithHandlerContextDecorator](#queue-withhandlercontextdecorator) · [WithLegacyDirectEnvelope](#queue-withlegacydirectenvelope) · [WithMiddleware](#queue-withmiddleware) · [WithObserver](#queue-withobserver) · [WithStore](#queue-withstore) · [WithWorkers](#queue-withworkers) · [Queue.WithWorkers](#queue-queue-withworkers) | +| **Queue** | [Batch](#queue-queue-batch) · [Bind](#queue-message-bind) · [Chain](#queue-queue-chain) · [Dispatch](#queue-queue-dispatch) · [Driver](#queue-queue-driver) · [FindBatch](#queue-queue-findbatch) · [FindChain](#queue-queue-findchain) · [IsPermanent](#queue-ispermanent) · [Pause](#queue-queue-pause) · [PayloadAs](#queue-message-payloadas) · [PayloadBytes](#queue-message-payloadbytes) · [Permanent](#queue-permanent) · [Prune](#queue-queue-prune) · [Ready](#queue-queue-ready) · [Register](#queue-queue-register) · [Resume](#queue-queue-resume) · [Run](#queue-queue-run) · [Shutdown](#queue-queue-shutdown) · [StartWorkers](#queue-queue-startworkers) · [Stats](#queue-queue-stats) · [WithClock](#queue-withclock) · [WithContext](#queue-queue-withcontext) · [WithHandlerContextDecorator](#queue-withhandlercontextdecorator) · [WithLegacyDirectEnvelope](#queue-withlegacydirectenvelope) · [WithMiddleware](#queue-withmiddleware) · [WithObserver](#queue-withobserver) · [WithStore](#queue-withstore) · [WithWorkers](#queue-withworkers) · [Queue.WithWorkers](#queue-queue-withworkers) | | **Driver Constructors** | [mysqlqueue.New](#mysqlqueue-new) · [mysqlqueue.NewWithConfig](#mysqlqueue-newwithconfig) · [natsqueue.New](#natsqueue-new) · [natsqueue.NewWithConfig](#natsqueue-newwithconfig) · [postgresqueue.New](#postgresqueue-new) · [postgresqueue.NewWithConfig](#postgresqueue-newwithconfig) · [rabbitmqqueue.New](#rabbitmqqueue-new) · [rabbitmqqueue.NewWithConfig](#rabbitmqqueue-newwithconfig) · [redisqueue.New](#redisqueue-new) · [redisqueue.NewWithConfig](#redisqueue-newwithconfig) · [sqlitequeue.New](#sqlitequeue-new) · [sqlitequeue.NewWithConfig](#sqlitequeue-newwithconfig) · [sqsqueue.New](#sqsqueue-new) · [sqsqueue.NewWithConfig](#sqsqueue-newwithconfig) | | **Testing** | [AssertBatchCount](#queue-fakequeue-assertbatchcount) · [AssertBatched](#queue-fakequeue-assertbatched) · [AssertChained](#queue-fakequeue-assertchained) · [AssertCount](#queue-fakequeue-assertcount) · [AssertDispatched](#queue-fakequeue-assertdispatched) · [AssertDispatchedOn](#queue-fakequeue-assertdispatchedon) · [AssertDispatchedTimes](#queue-fakequeue-assertdispatchedtimes) · [AssertNotDispatched](#queue-fakequeue-assertnotdispatched) · [AssertNothingBatched](#queue-fakequeue-assertnothingbatched) · [AssertNothingDispatched](#queue-fakequeue-assertnothingdispatched) · [Batch](#queue-fakequeue-batch) · [BatchRecords](#queue-fakequeue-batchrecords) · [Chain](#queue-fakequeue-chain) · [ChainRecords](#queue-fakequeue-chainrecords) · [Dispatch](#queue-fakequeue-dispatch) · [Driver](#queue-fakequeue-driver) · [FindBatch](#queue-fakequeue-findbatch) · [FindChain](#queue-fakequeue-findchain) · [NewFake](#queue-newfake) · [Prune](#queue-fakequeue-prune) · [Ready](#queue-fakequeue-ready) · [Records](#queue-fakequeue-records) · [Register](#queue-fakequeue-register) · [Reset](#queue-fakequeue-reset) · [Shutdown](#queue-fakequeue-shutdown) · [StartWorkers](#queue-fakequeue-startworkers) · [WithContext](#queue-fakequeue-withcontext) · [Workers](#queue-fakequeue-workers) | @@ -963,6 +963,20 @@ jobMap := queue.NewJob("emails:send").Payload(map[string]any{ }) ``` +#### Job.PayloadAs + +PayloadAs unmarshals the job payload JSON and returns it as T. + +```go +type EmailPayload struct { + To string `json:"to"` +} +job := queue.NewJob("emails:send").Payload(EmailPayload{To: "user@example.com"}) +payload, err := job.PayloadAs[EmailPayload]() +fmt.Println(err == nil, payload.To) +// true user@example.com +``` + #### Job.PayloadBytes PayloadBytes returns a copy of job payload bytes. @@ -1670,6 +1684,20 @@ if queue.SupportsPause(q) { } ``` +#### Message.PayloadAs + +PayloadAs unmarshals the delivered payload and returns it as T. + +```go +type EmailPayload struct { + To string `json:"to"` +} +message := queue.NewMessage("emails:send", []byte(`{"to":"user@example.com"}`)) +payload, err := message.PayloadAs[EmailPayload]() +fmt.Println(err == nil, payload.To) +// true user@example.com +``` + #### Message.PayloadBytes PayloadBytes returns an isolated copy of the raw job payload. diff --git a/docs/go.mod b/docs/go.mod index 7950bae..5c425a0 100644 --- a/docs/go.mod +++ b/docs/go.mod @@ -1,3 +1,3 @@ module github.com/goforj/queue/docs -go 1.24.4 +go 1.27.0 diff --git a/docs/readme/testcounts/integration_count.json b/docs/readme/testcounts/integration_count.json index a3ae1f6..516f059 100644 --- a/docs/readme/testcounts/integration_count.json +++ b/docs/readme/testcounts/integration_count.json @@ -1,5 +1,5 @@ { "count": 618, - "source_hash": "sha256:f85ae1ec431ad62eda63a7ee43fb312f5860ed03f168307230c2c5ed6a56350c", + "source_hash": "sha256:f7ee227bc9d85ad4965b1db917f83ad94350a0ebf3ba9d2637f883b03c82af14", "backend_scope": "all" } diff --git a/driver/mysqlqueue/go.mod b/driver/mysqlqueue/go.mod index 3c8df97..8acecae 100644 --- a/driver/mysqlqueue/go.mod +++ b/driver/mysqlqueue/go.mod @@ -1,6 +1,6 @@ module github.com/goforj/queue/driver/mysqlqueue -go 1.24.4 +go 1.27.0 require ( github.com/go-sql-driver/mysql v1.9.3 diff --git a/driver/natsqueue/go.mod b/driver/natsqueue/go.mod index 52bd26a..bcf39fb 100644 --- a/driver/natsqueue/go.mod +++ b/driver/natsqueue/go.mod @@ -1,6 +1,6 @@ module github.com/goforj/queue/driver/natsqueue -go 1.24.4 +go 1.27.0 require ( github.com/goforj/queue v0.0.0 diff --git a/driver/postgresqueue/go.mod b/driver/postgresqueue/go.mod index 0272dc3..1d4a483 100644 --- a/driver/postgresqueue/go.mod +++ b/driver/postgresqueue/go.mod @@ -1,6 +1,6 @@ module github.com/goforj/queue/driver/postgresqueue -go 1.25.0 +go 1.27.0 require ( github.com/goforj/queue v0.0.0 diff --git a/driver/rabbitmqqueue/go.mod b/driver/rabbitmqqueue/go.mod index 9da9c83..4c3c224 100644 --- a/driver/rabbitmqqueue/go.mod +++ b/driver/rabbitmqqueue/go.mod @@ -1,6 +1,6 @@ module github.com/goforj/queue/driver/rabbitmqqueue -go 1.24.4 +go 1.27.0 require ( github.com/goforj/queue v0.0.0 diff --git a/driver/redisqueue/go.mod b/driver/redisqueue/go.mod index 7591c77..07d94c1 100644 --- a/driver/redisqueue/go.mod +++ b/driver/redisqueue/go.mod @@ -1,6 +1,6 @@ module github.com/goforj/queue/driver/redisqueue -go 1.24.4 +go 1.27.0 require ( github.com/goforj/queue v0.0.0 diff --git a/driver/sqlitequeue/go.mod b/driver/sqlitequeue/go.mod index c745df1..6869786 100644 --- a/driver/sqlitequeue/go.mod +++ b/driver/sqlitequeue/go.mod @@ -1,6 +1,6 @@ module github.com/goforj/queue/driver/sqlitequeue -go 1.24.4 +go 1.27.0 require ( github.com/goforj/queue v0.0.0 diff --git a/driver/sqlqueuecore/go.mod b/driver/sqlqueuecore/go.mod index 29f30d3..2fc1a28 100644 --- a/driver/sqlqueuecore/go.mod +++ b/driver/sqlqueuecore/go.mod @@ -1,6 +1,6 @@ module github.com/goforj/queue/driver/sqlqueuecore -go 1.24.4 +go 1.27.0 require github.com/goforj/queue v0.0.0 diff --git a/driver/sqsqueue/go.mod b/driver/sqsqueue/go.mod index e6f07eb..afd23a8 100644 --- a/driver/sqsqueue/go.mod +++ b/driver/sqsqueue/go.mod @@ -1,6 +1,6 @@ module github.com/goforj/queue/driver/sqsqueue -go 1.24.4 +go 1.27.0 require ( github.com/aws/aws-sdk-go-v2 v1.41.1 diff --git a/examples/go.mod b/examples/go.mod index 817143f..140ef48 100644 --- a/examples/go.mod +++ b/examples/go.mod @@ -1,6 +1,6 @@ module github.com/goforj/queue/examples -go 1.25.0 +go 1.27.0 require ( github.com/goforj/queue v0.0.0 diff --git a/examples/job-payloadas/main.go b/examples/job-payloadas/main.go new file mode 100644 index 0000000..eac3e4b --- /dev/null +++ b/examples/job-payloadas/main.go @@ -0,0 +1,24 @@ +//go:build ignore +// +build ignore + +// examplegen:generated + +package main + +import ( + "fmt" + "github.com/goforj/queue" +) + +func main() { + // PayloadAs unmarshals the job payload JSON and returns it as T. + + // Example: typed job payload + type EmailPayload struct { + To string `json:"to"` + } + job := queue.NewJob("emails:send").Payload(EmailPayload{To: "user@example.com"}) + payload, err := job.PayloadAs[EmailPayload]() + fmt.Println(err == nil, payload.To) + // true user@example.com +} diff --git a/examples/message-payloadas/main.go b/examples/message-payloadas/main.go new file mode 100644 index 0000000..7ab18bc --- /dev/null +++ b/examples/message-payloadas/main.go @@ -0,0 +1,24 @@ +//go:build ignore +// +build ignore + +// examplegen:generated + +package main + +import ( + "fmt" + "github.com/goforj/queue" +) + +func main() { + // PayloadAs unmarshals the delivered payload and returns it as T. + + // Example: typed message payload + type EmailPayload struct { + To string `json:"to"` + } + message := queue.NewMessage("emails:send", []byte(`{"to":"user@example.com"}`)) + payload, err := message.PayloadAs[EmailPayload]() + fmt.Println(err == nil, payload.To) + // true user@example.com +} diff --git a/go.mod b/go.mod index 29a35b3..81f9998 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/goforj/queue -go 1.24.4 +go 1.27.0 require modernc.org/sqlite v1.46.0 diff --git a/go.work b/go.work index 199d52e..d0a3809 100644 --- a/go.work +++ b/go.work @@ -1,4 +1,4 @@ -go 1.25.0 +go 1.27.0 use ( . diff --git a/integration/go.mod b/integration/go.mod index 36ddfd3..d2c3fbe 100644 --- a/integration/go.mod +++ b/integration/go.mod @@ -1,6 +1,6 @@ module github.com/goforj/queue/integration -go 1.25.0 +go 1.27.0 replace github.com/goforj/queue => .. diff --git a/job.go b/job.go index 5351346..c4122a3 100644 --- a/job.go +++ b/job.go @@ -271,6 +271,24 @@ func (t Job) Bind(dst any) error { return nil } +// PayloadAs unmarshals the job payload JSON and returns it as T. +// @group Job +// +// Example: typed job payload +// +// type EmailPayload struct { +// To string `json:"to"` +// } +// job := queue.NewJob("emails:send").Payload(EmailPayload{To: "user@example.com"}) +// payload, err := job.PayloadAs[EmailPayload]() +// fmt.Println(err == nil, payload.To) +// // true user@example.com +func (t Job) PayloadAs[T any]() (T, error) { + var out T + err := t.Bind(&out) + return out, err +} + func encodePayload(payload any) ([]byte, error) { if payload == nil { return nil, nil diff --git a/payload_as_test.go b/payload_as_test.go new file mode 100644 index 0000000..ed05c52 --- /dev/null +++ b/payload_as_test.go @@ -0,0 +1,96 @@ +package queue + +import ( + "testing" +) + +// payloadAsValue is the shared payload shape for result-method tests and benchmarks. +type payloadAsValue struct { + ID int `json:"id"` +} + +// payloadAsSink prevents decoded benchmark values from being optimized away. +var payloadAsSink payloadAsValue + +// TestPayloadAsMethods verifies jobs and delivered messages return equivalent typed payloads. +func TestPayloadAsMethods(t *testing.T) { + job := NewJob("reports:build").Payload(payloadAsValue{ID: 27}) + jobAs := (Job).PayloadAs[payloadAsValue] + fromJob, err := jobAs(job) + if err != nil || fromJob.ID != 27 { + t.Fatalf("Job.PayloadAs returned %+v, %v", fromJob, err) + } + + message := NewMessage("reports:build", job.PayloadBytes()) + messageAs := message.PayloadAs[payloadAsValue] + fromMessage, err := messageAs() + if err != nil || fromMessage.ID != 27 { + t.Fatalf("Message.PayloadAs returned %+v, %v", fromMessage, err) + } +} + +// TestPayloadAsErrors verifies result methods preserve each receiver's existing Bind errors. +func TestPayloadAsErrors(t *testing.T) { + job := NewJob("reports:build").Payload([]byte("not-json")) + var jobDestination payloadAsValue + bindErr := job.Bind(&jobDestination) + _, resultErr := job.PayloadAs[payloadAsValue]() + if bindErr == nil || resultErr == nil || bindErr.Error() != resultErr.Error() { + t.Fatalf("Job errors differ: Bind=%v PayloadAs=%v", bindErr, resultErr) + } + + message := NewMessage("reports:build", []byte("not-json")) + var messageDestination payloadAsValue + bindErr = message.Bind(&messageDestination) + _, resultErr = message.PayloadAs[payloadAsValue]() + if bindErr == nil || resultErr == nil || bindErr.Error() != resultErr.Error() { + t.Fatalf("Message errors differ: Bind=%v PayloadAs=%v", bindErr, resultErr) + } +} + +// BenchmarkPayloadDecode compares caller-owned binding and generic result methods. +func BenchmarkPayloadDecode(b *testing.B) { + job := NewJob("reports:build").Payload(payloadAsValue{ID: 27}) + message := NewMessage("reports:build", job.PayloadBytes()) + + b.Run("JobBind", func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + var out payloadAsValue + if err := job.Bind(&out); err != nil { + b.Fatalf("Bind: %v", err) + } + payloadAsSink = out + } + }) + b.Run("JobPayloadAs", func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + out, err := job.PayloadAs[payloadAsValue]() + if err != nil { + b.Fatalf("PayloadAs: %v", err) + } + payloadAsSink = out + } + }) + b.Run("MessageBind", func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + var out payloadAsValue + if err := message.Bind(&out); err != nil { + b.Fatalf("Bind: %v", err) + } + payloadAsSink = out + } + }) + b.Run("MessagePayloadAs", func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + out, err := message.PayloadAs[payloadAsValue]() + if err != nil { + b.Fatalf("PayloadAs: %v", err) + } + payloadAsSink = out + } + }) +} diff --git a/scripts/module-go-versions.tsv b/scripts/module-go-versions.tsv index 21a27c3..c261f04 100644 --- a/scripts/module-go-versions.tsv +++ b/scripts/module-go-versions.tsv @@ -1,13 +1,13 @@ # Exact module Go versions. The workspace uses the highest version in this file. -. 1.24.4 -docs 1.24.4 -driver/mysqlqueue 1.24.4 -driver/natsqueue 1.24.4 -driver/postgresqueue 1.25.0 -driver/rabbitmqqueue 1.24.4 -driver/redisqueue 1.24.4 -driver/sqlitequeue 1.24.4 -driver/sqlqueuecore 1.24.4 -driver/sqsqueue 1.24.4 -examples 1.25.0 -integration 1.25.0 +. 1.27.0 +docs 1.27.0 +driver/mysqlqueue 1.27.0 +driver/natsqueue 1.27.0 +driver/postgresqueue 1.27.0 +driver/rabbitmqqueue 1.27.0 +driver/redisqueue 1.27.0 +driver/sqlitequeue 1.27.0 +driver/sqlqueuecore 1.27.0 +driver/sqsqueue 1.27.0 +examples 1.27.0 +integration 1.27.0 diff --git a/workflow_models.go b/workflow_models.go index 6dce5bf..6baecdc 100644 --- a/workflow_models.go +++ b/workflow_models.go @@ -43,6 +43,24 @@ func (m Message) Bind(dst any) error { return json.Unmarshal(m.payload, dst) } +// PayloadAs unmarshals the delivered payload and returns it as T. +// @group Queue +// +// Example: typed message payload +// +// type EmailPayload struct { +// To string `json:"to"` +// } +// message := queue.NewMessage("emails:send", []byte(`{"to":"user@example.com"}`)) +// payload, err := message.PayloadAs[EmailPayload]() +// fmt.Println(err == nil, payload.To) +// // true user@example.com +func (m Message) PayloadAs[T any]() (T, error) { + var out T + err := m.Bind(&out) + return out, err +} + // DispatchResult identifies an accepted logical dispatch. // @group Queue type DispatchResult struct {