diff --git a/modules/mesh-build-go/buildstep/pull_bus.go b/modules/mesh-build-go/buildstep/pull_bus.go index 8633ed77b..20c0528be 100644 --- a/modules/mesh-build-go/buildstep/pull_bus.go +++ b/modules/mesh-build-go/buildstep/pull_bus.go @@ -93,7 +93,8 @@ func DeclareAskWorkers(ctx context.Context, js jetstream.JetStream, language str for _, kind := range sdkstep.AskKinds { worker, _ := sdkstep.AskWorker(language, kind) subject, _ := sdkstep.AskSubject(language, kind) - if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: worker, FilterSubject: subject, + // Both forms of the accept subject: main's and the asker's, `.` (design 47 G §600). + if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: worker, FilterSubjects: []string{subject, subject + ".*"}, AckPolicy: jetstream.AckExplicitPolicy, AckWait: ackWait, MaxDeliver: 5}); err != nil { return fmt.Errorf("the worker %s: %w", worker, err) } diff --git a/modules/mesh-build-go/buildstep/pull_integration_test.go b/modules/mesh-build-go/buildstep/pull_integration_test.go index 63bfc2df2..d6f043a6f 100644 --- a/modules/mesh-build-go/buildstep/pull_integration_test.go +++ b/modules/mesh-build-go/buildstep/pull_integration_test.go @@ -196,3 +196,32 @@ func TestAnAskThatMetTheNetworkIsGivenAgainAndADroppedOneIsEnded(t *testing.T) { } func contains(s, sub string) bool { return strings.Contains(s, sub) } + +// Design 47 G §600: the seat's worker takes an ask put on either form of the accept subject, main's and the asker's. +func TestTheWorkerTakesBothFormsOfTheAcceptSubject(t *testing.T) { + js := askBus(t, 30*time.Second) + stream, _ := sdkstep.AskStream(sdkstep.AskGo) + worker, _ := sdkstep.AskWorker(sdkstep.AskGo, sdkstep.AskDelivery) + main, _ := sdkstep.AskSubject(sdkstep.AskGo, sdkstep.AskDelivery) + for _, subject := range []string{main, main + ".builder"} { + if _, err := js.Publish(context.Background(), subject, []byte("{}")); err != nil { + t.Fatal(err) + } + } + c, err := js.Consumer(context.Background(), stream, worker) + if err != nil { + t.Fatal(err) + } + batch, err := c.Fetch(2, jetstream.FetchMaxWait(3*time.Second)) + if err != nil { + t.Fatal(err) + } + var got []string + for m := range batch.Messages() { + got = append(got, m.Subject()) + _ = m.Ack() + } + if len(got) != 2 { + t.Errorf("the worker took %v, want both forms of %s", got, main) + } +}