Skip to content

Commit 778d41d

Browse files
authored
Recursively visit payloads inside ScheduleNexusOperationCommandAttributes when using system nexus (#290)
* Visit payloads inside system Nexus envelopes * Fix a bug in defaultWellKnownAnyVisitor, which could return a result before all goroutines within it had finished
1 parent 0df6e05 commit 778d41d

4 files changed

Lines changed: 364 additions & 62 deletions

File tree

cmd/proxygenerator/interceptor.go

Lines changed: 47 additions & 30 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

proxy/interceptor.go

Lines changed: 47 additions & 32 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

proxy/system_nexus.go

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
package proxy
2+
3+
import (
4+
"fmt"
5+
6+
"go.temporal.io/api/command/v1"
7+
"google.golang.org/protobuf/proto"
8+
"google.golang.org/protobuf/reflect/protoreflect"
9+
"google.golang.org/protobuf/reflect/protoregistry"
10+
)
11+
12+
// visitSystemNexusEnvelope decodes the system Nexus envelope in attrs.Input,
13+
// visits the payloads inside the decoded request message, and re-encodes it.
14+
//
15+
// The envelope's proto message type is taken from the payload's "messageType"
16+
// metadata, so no operation registry is required. The envelope must be encoded
17+
// as binary/protobuf. The inner payloads (and only those) are passed to the
18+
// visitor, so external storage and codecs apply to them and not to the envelope
19+
// itself, which is never offloaded or codec-encoded.
20+
func visitSystemNexusEnvelope(
21+
ctx *VisitPayloadsContext,
22+
options *VisitPayloadsOptions,
23+
concState *payloadConcurrencyState,
24+
attrs *command.ScheduleNexusOperationCommandAttributes,
25+
) error {
26+
input := attrs.Input
27+
28+
if encoding := string(input.GetMetadata()["encoding"]); encoding != "binary/protobuf" {
29+
return fmt.Errorf(
30+
"system nexus envelope for operation %q must be encoded as binary/protobuf but got %q",
31+
attrs.GetOperation(), encoding,
32+
)
33+
}
34+
35+
messageType := string(input.GetMetadata()["messageType"])
36+
if messageType == "" {
37+
return fmt.Errorf(
38+
"system nexus envelope for operation %q is missing the messageType metadata",
39+
attrs.GetOperation(),
40+
)
41+
}
42+
43+
mt, err := protoregistry.GlobalTypes.FindMessageByName(protoreflect.FullName(messageType))
44+
if err != nil {
45+
return fmt.Errorf(
46+
"system nexus envelope for operation %q references unknown message type %q: %w",
47+
attrs.GetOperation(), messageType, err,
48+
)
49+
}
50+
msg := mt.New().Interface()
51+
52+
if err := proto.Unmarshal(input.GetData(), msg); err != nil {
53+
return fmt.Errorf(
54+
"failed to unmarshal system nexus envelope for operation %q: %w",
55+
attrs.GetOperation(), err,
56+
)
57+
}
58+
59+
if err := visitPayloadsAndWait(ctx, options, msg, concState, msg); err != nil {
60+
return err
61+
}
62+
63+
data, err := proto.Marshal(msg)
64+
if err != nil {
65+
return fmt.Errorf(
66+
"failed to marshal system nexus envelope for operation %q: %w",
67+
attrs.GetOperation(), err,
68+
)
69+
}
70+
input.Data = data
71+
return nil
72+
}

0 commit comments

Comments
 (0)