Skip to content

Commit 84a3122

Browse files
committed
refactor(queue): adopt WithContext dispatch pattern
1 parent 7a66260 commit 84a3122

44 files changed

Lines changed: 519 additions & 329 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

README.md

Lines changed: 29 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
<a href="https://goreportcard.com/report/github.com/goforj/queue"><img src="https://goreportcard.com/badge/github.com/goforj/queue" alt="Go Report Card"></a>
1616
<a href="https://codecov.io/gh/goforj/queue"><img src="https://codecov.io/gh/goforj/queue/graph/badge.svg?token=40Z5UQATME"/></a>
1717
<!-- test-count:embed:start -->
18-
<img src="https://img.shields.io/badge/unit_tests-262-brightgreen" alt="Unit tests (executed count)">
18+
<img src="https://img.shields.io/badge/unit_tests-273-brightgreen" alt="Unit tests (executed count)">
1919
<img src="https://img.shields.io/badge/integration_tests-541-blue" alt="Integration tests (executed count)">
2020
<!-- test-count:embed:end -->
2121
</p>
@@ -550,7 +550,8 @@ The API section below is autogenerated; do not edit between the markers.
550550
| **Constructors** | [New](#queue-new) [NewNull](#queue-newnull) [NewStatsCollector](#queue-newstatscollector) [NewSync](#queue-newsync) [NewWorkerpool](#queue-newworkerpool) |
551551
| **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) |
552552
| **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) |
553-
| **Queue** | [Batch](#queue-queue-batch) [Chain](#queue-queue-chain) [Dispatch](#queue-queue-dispatch) [DispatchCtx](#queue-queue-dispatchctx) [Driver](#queue-queue-driver) [FindBatch](#queue-queue-findbatch) [FindChain](#queue-queue-findchain) [Pause](#queue-queue-pause) [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) [WithMiddleware](#queue-withmiddleware) [WithObserver](#queue-withobserver) [WithStore](#queue-withstore) [WithWorkers](#queue-withworkers) [Queue.WithWorkers](#queue-queue-withworkers) |
553+
| **Other** | [ResolveObservedJobType](#queue-resolveobservedjobtype) |
554+
| **Queue** | [Batch](#queue-queue-batch) [Chain](#queue-queue-chain) [Dispatch](#queue-queue-dispatch) [Driver](#queue-queue-driver) [FindBatch](#queue-queue-findbatch) [FindChain](#queue-queue-findchain) [Pause](#queue-queue-pause) [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) [WithMiddleware](#queue-withmiddleware) [WithObserver](#queue-withobserver) [WithStore](#queue-withstore) [WithWorkers](#queue-withworkers) [Queue.WithWorkers](#queue-queue-withworkers) |
554555
| **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) |
555556
| **Testing** | [AssertBatchCount](#queuefake-fake-assertbatchcount) [AssertBatched](#queuefake-fake-assertbatched) [AssertChained](#queuefake-fake-assertchained) [AssertCount](#queuefake-fake-assertcount) [AssertDispatched](#queuefake-fake-assertdispatched) [AssertDispatchedOn](#queuefake-fake-assertdispatchedon) [AssertDispatchedTimes](#queuefake-fake-assertdispatchedtimes) [AssertNotDispatched](#queuefake-fake-assertnotdispatched) [AssertNothingBatched](#queuefake-fake-assertnothingbatched) [AssertNothingDispatched](#queuefake-fake-assertnothingdispatched) [AssertNothingWorkflowDispatched](#queuefake-fake-assertnothingworkflowdispatched) [AssertWorkflowDispatched](#queuefake-fake-assertworkflowdispatched) [AssertWorkflowDispatchedOn](#queuefake-fake-assertworkflowdispatchedon) [AssertWorkflowDispatchedTimes](#queuefake-fake-assertworkflowdispatchedtimes) [AssertWorkflowNotDispatched](#queuefake-fake-assertworkflownotdispatched) [Count](#queuefake-fake-count) [CountJob](#queuefake-fake-countjob) [CountOn](#queuefake-fake-counton) [New](#queuefake-new) [Queue](#queuefake-fake-queue) [Records](#queuefake-fake-records) [Reset](#queuefake-fake-reset) [Workflow](#queuefake-fake-workflow) |
556557

@@ -1047,9 +1048,9 @@ MultiObserver fans out events to multiple observers.
10471048
events := make(chan queue.Event, 2)
10481049
observer := queue.MultiObserver(
10491050
queue.ChannelObserver{Events: events},
1050-
queue.ObserverFunc(func(queue.Event) {}),
1051+
queue.ObserverFunc(func(context.Context, queue.Event) {}),
10511052
)
1052-
observer.Observe(queue.Event{Kind: queue.EventEnqueueAccepted})
1053+
observer.Observe(context.Background(), queue.Event{Kind: queue.EventEnqueueAccepted})
10531054
fmt.Println(len(events))
10541055
// Output: 1
10551056
```
@@ -1061,7 +1062,7 @@ Observe forwards an event to the configured channel.
10611062
```go
10621063
ch := make(chan queue.Event, 1)
10631064
observer := queue.ChannelObserver{Events: ch}
1064-
observer.Observe(queue.Event{Kind: queue.EventProcessStarted, Queue: "default"})
1065+
observer.Observe(context.Background(), queue.Event{Kind: queue.EventProcessStarted, Queue: "default"})
10651066
event := <-ch
10661067
```
10671068

@@ -1071,7 +1072,7 @@ Observe handles a queue runtime event.
10711072

10721073
```go
10731074
var observer queue.Observer
1074-
observer.Observe(queue.Event{
1075+
observer.Observe(context.Background(), queue.Event{
10751076
Kind: queue.EventEnqueueAccepted,
10761077
Driver: queue.DriverSync,
10771078
Queue: "default",
@@ -1084,7 +1085,7 @@ Observe calls the wrapped function.
10841085

10851086
```go
10861087
logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
1087-
observer := queue.ObserverFunc(func(event queue.Event) {
1088+
observer := queue.ObserverFunc(func(ctx context.Context, event queue.Event) {
10881089
logger.Info("queue event",
10891090
"kind", event.Kind,
10901091
"driver", event.Driver,
@@ -1096,7 +1097,7 @@ observer := queue.ObserverFunc(func(event queue.Event) {
10961097
"err", event.Err,
10971098
)
10981099
})
1099-
observer.Observe(queue.Event{
1100+
observer.Observe(context.Background(), queue.Event{
11001101
Kind: queue.EventProcessSucceeded,
11011102
Driver: queue.DriverSync,
11021103
Queue: "default",
@@ -1110,7 +1111,7 @@ Observe records an event and updates normalized counters.
11101111

11111112
```go
11121113
collector := queue.NewStatsCollector()
1113-
collector.Observe(queue.Event{
1114+
collector.Observe(context.Background(), queue.Event{
11141115
Kind: queue.EventEnqueueAccepted,
11151116
Driver: queue.DriverSync,
11161117
Queue: "default",
@@ -1357,6 +1358,16 @@ fmt.Printf("ok=%v hour=%+v day=%+v week=%+v\n", ok, throughput.Hour, throughput.
13571358
// Output: ok=true hour={Processed:1 Failed:0} day={Processed:1 Failed:0} week={Processed:1 Failed:0}
13581359
```
13591360

1361+
#### Other
1362+
1363+
#### <a id="queue-resolveobservedjobtype"></a>ResolveObservedJobType
1364+
1365+
ResolveObservedJobType returns the effective application job type that should
1366+
be emitted to observers. External workers may process internal bus wrapper
1367+
jobs (for example, "bus:job") whose payload embeds the real application job
1368+
type. When possible, this helper unwraps that payload so dashboards and
1369+
metrics reflect the user-facing job type instead of the transport wrapper.
1370+
13601371
#### Queue
13611372

13621373
#### <a id="queue-queue-batch"></a>Batch
@@ -1394,7 +1405,7 @@ _, _ = q.Chain(
13941405

13951406
#### <a id="queue-queue-dispatch"></a>Queue.Dispatch
13961407

1397-
Dispatch enqueues a high-level job using context.Background.
1408+
Dispatch enqueues a high-level job using the queue's bound context.
13981409

13991410
```go
14001411
q, err := queue.NewSync()
@@ -1406,10 +1417,6 @@ job := queue.NewJob("emails:send").Payload(map[string]any{"id": 1}).OnQueue("def
14061417
_, _ = q.Dispatch(job)
14071418
```
14081419

1409-
#### <a id="queue-queue-dispatchctx"></a>Queue.DispatchCtx
1410-
1411-
DispatchCtx enqueues a high-level job using the provided context.
1412-
14131420
#### <a id="queue-queue-driver"></a>Queue.Driver
14141421

14151422
Driver reports the configured backend driver for the underlying queue runtime.
@@ -1606,6 +1613,10 @@ if err != nil {
16061613
}
16071614
```
16081615

1616+
#### <a id="queue-queue-withcontext"></a>Queue.WithContext
1617+
1618+
WithContext returns a derived queue handle bound to ctx.
1619+
16091620
#### <a id="queue-withmiddleware"></a>WithMiddleware
16101621

16111622
WithMiddleware appends queue workflow middleware.
@@ -1746,18 +1757,6 @@ fake := queue.NewFake()
17461757
err := fake.Dispatch(queue.NewJob("emails:send").OnQueue("default"))
17471758
```
17481759

1749-
#### <a id="queue-fakequeue-dispatchctx"></a>FakeQueue.DispatchCtx
1750-
1751-
DispatchCtx submits a typed job payload using the provided context.
1752-
1753-
```go
1754-
fake := queue.NewFake()
1755-
ctx := context.Background()
1756-
err := fake.DispatchCtx(ctx, queue.NewJob("emails:send").OnQueue("default"))
1757-
fmt.Println(err == nil)
1758-
// Output: true
1759-
```
1760-
17611760
#### <a id="queue-fakequeue-driver"></a>FakeQueue.Driver
17621761

17631762
Driver returns the active queue driver.
@@ -1847,6 +1846,10 @@ fake := queue.NewFake()
18471846
err := fake.StartWorkers(context.Background())
18481847
```
18491848

1849+
#### <a id="queue-fakequeue-withcontext"></a>FakeQueue.WithContext
1850+
1851+
WithContext returns a derived fake queue handle bound to ctx.
1852+
18501853
#### <a id="queue-fakequeue-workers"></a>FakeQueue.Workers
18511854

18521855
Workers sets desired worker concurrency before StartWorkers.

bus/driver/temporal/temporal.go

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -74,7 +74,7 @@ func (a *Adapter) Register(jobType string, handler bus.Handler) {
7474
a.mu.Unlock()
7575
}
7676

77-
func (a *Adapter) DispatchCtx(ctx context.Context, job bus.Job) (bus.DispatchResult, error) {
77+
func (a *Adapter) Dispatch(ctx context.Context, job bus.Job) (bus.DispatchResult, error) {
7878
if a.cfg.Engine == nil {
7979
return bus.DispatchResult{}, ErrEngineRequired
8080
}
@@ -271,10 +271,6 @@ func (b *temporalBatchBuilder) Finally(fn func(context.Context, bus.BatchState)
271271
return b
272272
}
273273

274-
func (a *Adapter) Dispatch(ctx context.Context, job bus.Job) (bus.DispatchResult, error) {
275-
return a.DispatchCtx(ctx, job)
276-
}
277-
278274
func (a *Adapter) StartWorkers(context.Context) error { return nil }
279275

280276
func (a *Adapter) FindChain(_ context.Context, chainID string) (bus.ChainState, error) {

contract_test.go

Lines changed: 18 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -123,7 +123,7 @@ func runQueueContractSuite(t *testing.T, factory contractFactory) {
123123
if factory.beforeEach != nil {
124124
factory.beforeEach(t)
125125
}
126-
err := d.DispatchCtx(context.Background(), NewJob("job:contract:immediate").Payload([]byte("ok")).OnQueue("default"))
126+
err := d.Dispatch(NewJob("job:contract:immediate").Payload([]byte("ok")).OnQueue("default"))
127127
if err != nil {
128128
t.Fatalf("immediate dispatch failed: %v", err)
129129
}
@@ -140,7 +140,7 @@ func runQueueContractSuite(t *testing.T, factory contractFactory) {
140140
if factory.beforeEach != nil {
141141
factory.beforeEach(t)
142142
}
143-
err := d.DispatchCtx(context.Background(), NewJob("job:contract:delay").Payload([]byte("delayed")).OnQueue("default").Delay(20*time.Millisecond))
143+
err := d.Dispatch(NewJob("job:contract:delay").Payload([]byte("delayed")).OnQueue("default").Delay(20 * time.Millisecond))
144144
if err != nil {
145145
t.Fatalf("delayed dispatch failed: %v", err)
146146
}
@@ -157,7 +157,7 @@ func runQueueContractSuite(t *testing.T, factory contractFactory) {
157157
if factory.beforeEach != nil {
158158
factory.beforeEach(t)
159159
}
160-
err := d.DispatchCtx(context.Background(), NewJob("job:contract:queue").Payload([]byte("queue")).OnQueue("contract"))
160+
err := d.Dispatch(NewJob("job:contract:queue").Payload([]byte("queue")).OnQueue("contract"))
161161
if err != nil {
162162
t.Fatalf("dispatch with queue failed: %v", err)
163163
}
@@ -174,7 +174,7 @@ func runQueueContractSuite(t *testing.T, factory contractFactory) {
174174
if factory.beforeEach != nil {
175175
factory.beforeEach(t)
176176
}
177-
err := d.DispatchCtx(context.Background(), NewJob("job:contract:noqueue").Payload([]byte("no-queue")))
177+
err := d.Dispatch(NewJob("job:contract:noqueue").Payload([]byte("no-queue")))
178178
if factory.requiresQueueName {
179179
if err == nil {
180180
t.Fatal("expected missing queue error")
@@ -200,7 +200,7 @@ func runQueueContractSuite(t *testing.T, factory contractFactory) {
200200
if factory.beforeEach != nil {
201201
factory.beforeEach(t)
202202
}
203-
err := d.DispatchCtx(nil, NewJob("job:contract:nilctx").Payload([]byte("ok")).OnQueue("default"))
203+
err := d.WithContext(nil).Dispatch(NewJob("job:contract:nilctx").Payload([]byte("ok")).OnQueue("default"))
204204
if err != nil {
205205
t.Fatalf("dispatch with nil context failed: %v", err)
206206
}
@@ -225,7 +225,7 @@ func runQueueContractSuite(t *testing.T, factory contractFactory) {
225225
if factory.beforeEach != nil {
226226
factory.beforeEach(t)
227227
}
228-
err := d.DispatchCtx(context.Background(), NewJob("job:contract:timeout").Payload([]byte("timeout")).OnQueue("default").Timeout(80*time.Millisecond))
228+
err := d.Dispatch(NewJob("job:contract:timeout").Payload([]byte("timeout")).OnQueue("default").Timeout(80 * time.Millisecond))
229229
if err != nil {
230230
t.Fatalf("dispatch with timeout failed: %v", err)
231231
}
@@ -253,7 +253,7 @@ func runQueueContractSuite(t *testing.T, factory contractFactory) {
253253
factory.beforeEach(t)
254254
}
255255
job := NewJob("job:contract:invalid-payload").Payload(func() {})
256-
err := d.DispatchCtx(context.Background(), job.OnQueue("default"))
256+
err := d.Dispatch(job.OnQueue("default"))
257257
if err == nil {
258258
t.Fatal("expected payload build error")
259259
}
@@ -280,7 +280,7 @@ func runQueueContractSuite(t *testing.T, factory contractFactory) {
280280
Backoff(-1 * time.Second).
281281
Delay(-1 * time.Second).
282282
UniqueFor(-1 * time.Second)
283-
err := d.DispatchCtx(context.Background(), job)
283+
err := d.Dispatch(job)
284284
if err == nil {
285285
t.Fatal("expected invalid option value error")
286286
}
@@ -314,7 +314,7 @@ func runQueueContractSuite(t *testing.T, factory contractFactory) {
314314
factory.beforeEach(t)
315315
}
316316
want := payload{ID: 42, Name: "bind"}
317-
if err := d.DispatchCtx(context.Background(), NewJob("job:contract:bind").Payload(want).OnQueue("default")); err != nil {
317+
if err := d.Dispatch(NewJob("job:contract:bind").Payload(want).OnQueue("default")); err != nil {
318318
t.Fatalf("dispatch bind job failed: %v", err)
319319
}
320320
select {
@@ -346,7 +346,7 @@ func runQueueContractSuite(t *testing.T, factory contractFactory) {
346346
if factory.beforeEach != nil {
347347
factory.beforeEach(t)
348348
}
349-
err := d.DispatchCtx(context.Background(), NewJob("job:contract:maxretry").Payload([]byte("retry")).OnQueue("default").Retry(2))
349+
err := d.Dispatch(NewJob("job:contract:maxretry").Payload([]byte("retry")).OnQueue("default").Retry(2))
350350
if err != nil {
351351
t.Fatalf("dispatch with max retry failed: %v", err)
352352
}
@@ -382,7 +382,7 @@ func runQueueContractSuite(t *testing.T, factory contractFactory) {
382382
if factory.beforeEach != nil {
383383
factory.beforeEach(t)
384384
}
385-
err := d.DispatchCtx(context.Background(), NewJob("job:contract:backoff").Payload([]byte("backoff")).OnQueue("default").Retry(1).Backoff(10*time.Millisecond))
385+
err := d.Dispatch(NewJob("job:contract:backoff").Payload([]byte("backoff")).OnQueue("default").Retry(1).Backoff(10 * time.Millisecond))
386386
if factory.backoffUnsupported {
387387
if !errors.Is(err, ErrBackoffUnsupported) {
388388
t.Fatalf("expected ErrBackoffUnsupported, got %v", err)
@@ -428,16 +428,16 @@ func runQueueContractSuite(t *testing.T, factory contractFactory) {
428428
}
429429
jobType := "job:contract:unique"
430430
payload := []byte("same")
431-
err := d.DispatchCtx(context.Background(), NewJob(jobType).Payload(payload).OnQueue("default").UniqueFor(ttl))
431+
err := d.Dispatch(NewJob(jobType).Payload(payload).OnQueue("default").UniqueFor(ttl))
432432
if err != nil {
433433
t.Fatalf("first unique dispatch failed: %v", err)
434434
}
435-
err = d.DispatchCtx(context.Background(), NewJob(jobType).Payload(payload).OnQueue("default").UniqueFor(ttl))
435+
err = d.Dispatch(NewJob(jobType).Payload(payload).OnQueue("default").UniqueFor(ttl))
436436
if !errors.Is(err, ErrDuplicate) {
437437
t.Fatalf("expected ErrDuplicate, got %v", err)
438438
}
439439
time.Sleep(expiryWait)
440-
err = d.DispatchCtx(context.Background(), NewJob(jobType).Payload(payload).OnQueue("default").UniqueFor(ttl))
440+
err = d.Dispatch(NewJob(jobType).Payload(payload).OnQueue("default").UniqueFor(ttl))
441441
if err != nil {
442442
t.Fatalf("dispatch after unique ttl failed: %v", err)
443443
}
@@ -461,19 +461,19 @@ func runQueueContractSuite(t *testing.T, factory contractFactory) {
461461
ttl = 400 * time.Millisecond
462462
}
463463
first := NewJob(jobType).Payload(payload).OnQueue("queue-a").UniqueFor(ttl)
464-
if err := d.DispatchCtx(context.Background(), first); err != nil {
464+
if err := d.Dispatch(first); err != nil {
465465
t.Fatalf("first dispatch failed: %v", err)
466466
}
467467
second := NewJob(jobType).Payload(payload).OnQueue("queue-b").UniqueFor(ttl)
468-
if err := d.DispatchCtx(context.Background(), second); err != nil {
468+
if err := d.Dispatch(second); err != nil {
469469
t.Fatalf("expected unique lock to be queue scoped, got %v", err)
470470
}
471471
})
472472

473473
t.Run("missing_job_type", func(t *testing.T) {
474474
d := factory.newQueue(t)
475475
t.Cleanup(func() { _ = d.Shutdown(context.Background()) })
476-
err := d.DispatchCtx(context.Background(), NewJob("").OnQueue("default"))
476+
err := d.Dispatch(NewJob("").OnQueue("default"))
477477
if err == nil {
478478
t.Fatal("expected missing job type error")
479479
}
@@ -491,7 +491,7 @@ func runQueueContractSuite(t *testing.T, factory contractFactory) {
491491
if factory.beforeEach != nil {
492492
factory.beforeEach(t)
493493
}
494-
err := d.DispatchCtx(context.Background(), NewJob("job:contract:missing-handler").OnQueue("default"))
494+
err := d.Dispatch(NewJob("job:contract:missing-handler").OnQueue("default"))
495495
if factory.assertMissingHandlerErr {
496496
if err == nil {
497497
t.Fatal("expected missing handler error")

docs/bus-design.md

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -172,7 +172,7 @@ Execution flow:
172172

173173
1. User dispatches job/chain/batch via Bus.
174174
2. Bus stores orchestration metadata (when required).
175-
3. Bus enqueues internal envelope job(s) via `q.DispatchCtx(...)`.
175+
3. Bus enqueues internal envelope job(s) via `q.WithContext(ctx).Dispatch(...)`.
176176
4. Queue workers execute bus internal handlers.
177177
5. Bus executes registered user handler and updates orchestration state.
178178
6. Bus enqueues next node(s)/callback jobs as needed.

docs/design.md

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,7 @@ Primary objectives:
4242

4343
- `StartWorkers(ctx context.Context) error`
4444
- `Dispatch(job any) error`
45-
- `DispatchCtx(ctx context.Context, job any) error`
45+
- `WithContext(ctx).Dispatch(job)` for context-aware dispatch
4646
- `Register(jobType string, handler Handler)`
4747
- `Shutdown(ctx context.Context) error`
4848

@@ -71,7 +71,7 @@ Worker execution settings are configured from `Queue` via `Workers(n).StartWorke
7171

7272
## Job API (fluent value object)
7373

74-
`Job` is immutable-ish via value receivers and explicit execution boundary (`Queue.DispatchCtx`):
74+
`Job` is immutable-ish via value receivers and explicit execution boundary (`Queue.WithContext(ctx).Dispatch(...)`):
7575

7676
- `NewJob(jobType string) Job`
7777
- `Payload(any) Job`

0 commit comments

Comments
 (0)