diff --git a/crates/client/src/grpc.rs b/crates/client/src/grpc.rs index 1c3d5b0a6d..ef9e212e41 100644 --- a/crates/client/src/grpc.rs +++ b/crates/client/src/grpc.rs @@ -1878,6 +1878,52 @@ proxier! { r.extensions_mut().insert(labels); } ); + ( + attach_stream_callback, + AttachStreamCallbackRequest, + AttachStreamCallbackResponse, + |r| { + let labels = namespaced_request!(r); + r.extensions_mut().insert(labels); + } + ); + ( + detach_stream_callback, + DetachStreamCallbackRequest, + DetachStreamCallbackResponse, + |r| { + let labels = namespaced_request!(r); + r.extensions_mut().insert(labels); + } + ); + ( + notify_stream, + NotifyStreamRequest, + NotifyStreamResponse, + |r| { + let labels = namespaced_request!(r); + r.extensions_mut().insert(labels); + } + ); + ( + describe_stream_notifier, + DescribeStreamNotifierRequest, + DescribeStreamNotifierResponse, + |r| { + let labels = namespaced_request!(r); + r.extensions_mut().insert(labels); + } + ); + ( + poll_stream_notifier, + PollStreamNotifierRequest, + PollStreamNotifierResponse, + |r| { + let labels = namespaced_request!(r); + r.extensions_mut().insert(labels); + r.extensions_mut().insert(IsUserLongPoll); + } + ); } proxier! { diff --git a/crates/common/build.rs b/crates/common/build.rs index b3170c97ef..b8a83a26ef 100644 --- a/crates/common/build.rs +++ b/crates/common/build.rs @@ -761,6 +761,8 @@ const BLOB_FIELDS: &[&str] = &[ "temporal.api.protocol.v1.Message.body", // whole Any body; see EXTRA_WHOLE_MESSAGE_LEAVES "temporal.api.query.v1.WorkflowQuery.query_args", "temporal.api.workflow.v1.NewWorkflowExecutionInfo.input", + "temporal.api.workflowservice.v1.NotifyStreamRequest.close_failure", + "temporal.api.workflowservice.v1.NotifyStreamRequest.close_result", "temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatByIdRequest.details", "temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatRequest.details", "temporal.api.workflowservice.v1.RespondActivityTaskCanceledByIdRequest.details", diff --git a/crates/protos/build.rs b/crates/protos/build.rs index 48c05389c0..55d4a24e39 100644 --- a/crates/protos/build.rs +++ b/crates/protos/build.rs @@ -41,6 +41,7 @@ const SERDE_DERIVE_PREFIXES: &[&str] = &[ ".temporal.api.rules", ".temporal.api.schedule", ".temporal.api.sdk", + ".temporal.api.stream", ".temporal.api.taskqueue", ".temporal.api.testservice", ".temporal.api.update", diff --git a/crates/protos/protos/api_upstream/openapi/openapiv2.json b/crates/protos/protos/api_upstream/openapi/openapiv2.json index 0b643ceb79..a677fc013c 100644 --- a/crates/protos/protos/api_upstream/openapi/openapiv2.json +++ b/crates/protos/protos/api_upstream/openapi/openapiv2.json @@ -20049,6 +20049,52 @@ } } }, + "v1StreamWake": { + "type": "object", + "properties": { + "subscriptionId": { + "type": "string", + "description": "The `subscription_id` of the `WorkflowWakeTarget` this wake is for." + }, + "position": { + "type": "string", + "description": "The position that came with `counter`, in the producer's terms." + }, + "counter": { + "type": "string", + "format": "int64", + "description": "The highest counter the notifier has seen. A reader that armed its read at a counter wakes\nwhen this value is above it. Zero when the notifier has seen no notification." + }, + "metadata": { + "type": "object", + "additionalProperties": { + "type": "string" + }, + "description": "The metadata that came with `counter`." + }, + "recheck": { + "type": "boolean", + "description": "Set when the notifier asks the reader to read again with no new counter. The notifier sends\nit after a quiet period, so a reader also finds records whose notification was lost." + }, + "closed": { + "type": "boolean", + "description": "Set when the stream closed. The reader reads to the end of the stream, then sees the\noutcome." + }, + "closeResult": { + "$ref": "#/definitions/v1Payload", + "description": "The result the stream closed with, when `closed` is set. Optional." + }, + "closeFailure": { + "$ref": "#/definitions/v1Failure", + "description": "The failure the stream closed with, when `closed` is set and it closed with one." + }, + "resubscribe": { + "type": "boolean", + "description": "Set when the notifier dropped this reader, for example at the notifier's idle timeout. The\nreader attaches again under a new request ID." + } + }, + "description": "A wake for one reader of a stream in a Workflow. The server records it on the next Workflow\nTask scheduled event, so it adds no event of its own.\n\nA wake says that the stream moved, not what it holds. The reader reads the stream itself. Wakes\nof one reader are folded, so the reader sees the latest one, with the highest `counter`." + }, "v1StructuredCalendarSpec": { "type": "object", "properties": { @@ -22620,6 +22666,14 @@ "type": "integer", "format": "int32", "title": "Starting at 1, how many attempts there have been to complete this task" + }, + "streamWakes": { + "type": "array", + "items": { + "type": "object", + "$ref": "#/definitions/v1StreamWake" + }, + "description": "Wakes for the Workflow's stream readers that arrived since the previous Workflow Task was\nscheduled, at most one entry per reader: the latest one. Recorded here rather than as\nevents of their own, so a burst of wakes costs one task and replay sees the same wakes the\nWorkflow saw. Bounded by `history.maximumStreamWakeSources` readers per execution, and it\ncounts toward the History size limit." } } }, diff --git a/crates/protos/protos/api_upstream/openapi/openapiv3.yaml b/crates/protos/protos/api_upstream/openapi/openapiv3.yaml index 2aa7c45bfb..92e4433895 100644 --- a/crates/protos/protos/api_upstream/openapi/openapiv3.yaml +++ b/crates/protos/protos/api_upstream/openapi/openapiv3.yaml @@ -17680,6 +17680,54 @@ components: type: type: string description: The type of the driver, required. + StreamWake: + type: object + properties: + subscriptionId: + type: string + description: The `subscription_id` of the `WorkflowWakeTarget` this wake is for. + position: + type: string + description: The position that came with `counter`, in the producer's terms. + counter: + type: string + description: |- + The highest counter the notifier has seen. A reader that armed its read at a counter wakes + when this value is above it. Zero when the notifier has seen no notification. + metadata: + type: object + additionalProperties: + type: string + description: The metadata that came with `counter`. + recheck: + type: boolean + description: |- + Set when the notifier asks the reader to read again with no new counter. The notifier sends + it after a quiet period, so a reader also finds records whose notification was lost. + closed: + type: boolean + description: |- + Set when the stream closed. The reader reads to the end of the stream, then sees the + outcome. + closeResult: + allOf: + - $ref: '#/components/schemas/Payload' + description: The result the stream closed with, when `closed` is set. Optional. + closeFailure: + allOf: + - $ref: '#/components/schemas/Failure' + description: The failure the stream closed with, when `closed` is set and it closed with one. + resubscribe: + type: boolean + description: |- + Set when the notifier dropped this reader, for example at the notifier's idle timeout. The + reader attaches again under a new request ID. + description: |- + A wake for one reader of a stream in a Workflow. The server records it on the next Workflow + Task scheduled event, so it adds no event of its own. + + A wake says that the stream moved, not what it holds. The reader reads the stream itself. Wakes + of one reader are folded, so the reader sees the latest one, with the highest `counter`. StructuredCalendarSpec: type: object properties: @@ -21255,6 +21303,16 @@ components: type: integer description: Starting at 1, how many attempts there have been to complete this task format: int32 + streamWakes: + type: array + items: + $ref: '#/components/schemas/StreamWake' + description: |- + Wakes for the Workflow's stream readers that arrived since the previous Workflow Task was + scheduled, at most one entry per reader: the latest one. Recorded here rather than as + events of their own, so a burst of wakes costs one task and replay sees the same wakes the + Workflow saw. Bounded by `history.maximumStreamWakeSources` readers per execution, and it + counts toward the History size limit. WorkflowTaskStartedEventAttributes: type: object properties: diff --git a/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto b/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto new file mode 100644 index 0000000000..4a079844ce --- /dev/null +++ b/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto @@ -0,0 +1,19 @@ +syntax = "proto3"; + +package temporal.api.enums.v1; + +option go_package = "go.temporal.io/api/enums/v1;enums"; +option java_package = "io.temporal.api.enums.v1"; +option java_multiple_files = true; +option java_outer_classname = "StreamProto"; +option ruby_package = "Temporalio::Api::Enums::V1"; +option csharp_namespace = "Temporalio.Api.Enums.V1"; + +// What owns a stream. The owner decides when the stream closes. +enum StreamOwnerKind { + STREAM_OWNER_KIND_UNSPECIFIED = 0; + // A Workflow owns the stream. The Workflow closes the stream itself, with a result or a + // failure. If the Workflow's run chain ends with no close and callbacks are attached, the + // stream notifier closes the stream with a failure within `streamnotifier.ownerCheckInterval`. + STREAM_OWNER_KIND_WORKFLOW = 1; +} diff --git a/crates/protos/protos/api_upstream/temporal/api/history/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/history/v1/message.proto index b40324d68c..bad6ae63f7 100644 --- a/crates/protos/protos/api_upstream/temporal/api/history/v1/message.proto +++ b/crates/protos/protos/api_upstream/temporal/api/history/v1/message.proto @@ -25,6 +25,7 @@ import "temporal/api/workflow/v1/message.proto"; import "temporal/api/sdk/v1/task_complete_metadata.proto"; import "temporal/api/sdk/v1/user_metadata.proto"; import "temporal/api/sdk/v1/event_group_marker.proto"; +import "temporal/api/stream/v1/message.proto"; // Always the first event in workflow history message WorkflowExecutionStartedEventAttributes { @@ -302,6 +303,12 @@ message WorkflowTaskScheduledEventAttributes { google.protobuf.Duration start_to_close_timeout = 2; // Starting at 1, how many attempts there have been to complete this task int32 attempt = 3; + // Wakes for the Workflow's stream readers that arrived since the previous Workflow Task was + // scheduled, at most one entry per reader: the latest one. Recorded here rather than as + // events of their own, so a burst of wakes costs one task and replay sees the same wakes the + // Workflow saw. Bounded by `history.maximumStreamWakeSources` readers per execution, and it + // counts toward the History size limit. + repeated temporal.api.stream.v1.StreamWake stream_wakes = 5; } message WorkflowTaskStartedEventAttributes { diff --git a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto new file mode 100644 index 0000000000..a179a71aad --- /dev/null +++ b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto @@ -0,0 +1,93 @@ +syntax = "proto3"; + +package temporal.api.stream.v1; + +option go_package = "go.temporal.io/api/stream/v1;stream"; +option java_package = "io.temporal.api.stream.v1"; +option java_multiple_files = true; +option java_outer_classname = "MessageProto"; +option ruby_package = "Temporalio::Api::Stream::V1"; +option csharp_namespace = "Temporalio.Api.Stream.V1"; + +import "google/protobuf/timestamp.proto"; +import "temporal/api/common/v1/message.proto"; +import "temporal/api/enums/v1/common.proto"; +import "temporal/api/enums/v1/stream.proto"; +import "temporal/api/failure/v1/message.proto"; + +// Names one stream within a namespace: its owner and its topic. +message StreamReference { + temporal.api.enums.v1.StreamOwnerKind owner_kind = 1; + // The owning Workflow, for `STREAM_OWNER_KIND_WORKFLOW`. + string workflow_id = 2; + // The id of the first run in the owning Workflow's run chain. The stream notifier is keyed by + // it, so one notifier serves the stream across the chain's runs, and a stream that reuses the + // Workflow id in a new chain gets a notifier of its own. An empty run id is refused. + string run_id = 3; + // The stream's topic within its owner. + string topic = 4; +} + +// A callback attached to a stream notifier, as DescribeStreamNotifier reports it. +message StreamCallbackInfo { + // The request ID the callback was attached with. + string request_id = 1; + // The callback. Its header keeps its keys, but every value is empty, since the values hold + // the caller's credentials. + temporal.api.common.v1.Callback callback = 2; + // The state of the callback's completion delivery. + temporal.api.enums.v1.CallbackState state = 3; + // The highest counter delivered to the callback as progress. + int64 delivered_counter = 4; + // Whether the receiver refused progress, so the callback only gets the completion. + bool progress_disabled = 5; + // The number of attempts made for the delivery in flight, starting at 1. + int32 attempt = 6; + // The time the last attempt completed. + google.protobuf.Timestamp last_attempt_complete_time = 7; + // The failure of the last attempt, if it failed. + temporal.api.failure.v1.Failure last_attempt_failure = 8; + // The Workflow the callback wakes, when it was attached with a `workflow_target` and not a + // Nexus callback. Then `callback` is unset. + WorkflowWakeTarget workflow_target = 9; +} + +// A Workflow that a stream notifier wakes when the stream moves or closes. The Workflow is in the +// namespace of the request that attached it. +message WorkflowWakeTarget { + string workflow_id = 1; + string run_id = 2; + // Names one reader of the stream in that run. The Workflow gets the wakes of each reader as + // their own `StreamWake`. At most 256 bytes. + string subscription_id = 3; +} + +// A wake for one reader of a stream in a Workflow. The server records it on the next Workflow +// Task scheduled event, so it adds no event of its own. +// +// A wake says that the stream moved, not what it holds. The reader reads the stream itself. Wakes +// of one reader are folded, so the reader sees the latest one, with the highest `counter`. +message StreamWake { + // The `subscription_id` of the `WorkflowWakeTarget` this wake is for. + string subscription_id = 1; + // The position that came with `counter`, in the producer's terms. + string position = 2; + // The highest counter the notifier has seen. A reader that armed its read at a counter wakes + // when this value is above it. Zero when the notifier has seen no notification. + int64 counter = 3; + // The metadata that came with `counter`. + map metadata = 4; + // Set when the notifier asks the reader to read again with no new counter. The notifier sends + // it after a quiet period, so a reader also finds records whose notification was lost. + bool recheck = 5; + // Set when the stream closed. The reader reads to the end of the stream, then sees the + // outcome. + bool closed = 6; + // The result the stream closed with, when `closed` is set. Optional. + temporal.api.common.v1.Payload close_result = 7; + // The failure the stream closed with, when `closed` is set and it closed with one. + temporal.api.failure.v1.Failure close_failure = 8; + // Set when the notifier dropped this reader, for example at the notifier's idle timeout. The + // reader attaches again under a new request ID. + bool resubscribe = 9; +} diff --git a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto index 1aae988d87..aaa5bca8bc 100644 --- a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto +++ b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto @@ -46,6 +46,7 @@ import "temporal/api/sdk/v1/task_complete_metadata.proto"; import "temporal/api/sdk/v1/user_metadata.proto"; import "temporal/api/nexus/v1/message.proto"; import "temporal/api/nexusoperation/v1/message.proto"; +import "temporal/api/stream/v1/message.proto"; import "temporal/api/worker/v1/message.proto"; import "google/protobuf/duration.proto"; @@ -3694,3 +3695,121 @@ message PollWorkflowExecutionTimeSkippingResponse { // The execution's current fast-forward, if any. temporal.api.common.v1.TimeSkippingFastForwardInfo fast_forward_info = 3; } + +message AttachStreamCallbackRequest { + string namespace = 1; + // The stream whose notifier holds the callback. + temporal.api.stream.v1.StreamReference stream_ref = 2; + // Identifies the attachment. Attaching again with the same request ID changes nothing, and + // DetachStreamCallback names the callback by it. A Nexus handler passes its start request ID. + string request_id = 3; + // Receives the stream's progress while it is open, and one completion when it closes. If the + // stream is already closed, the completion is delivered right away. A Nexus handler passes the + // callback URL and headers of its start request. + temporal.api.common.v1.Callback.Nexus callback = 4; + // The identity of the client attaching the callback. + string identity = 5; + // The operation token the handler returns from its start. The notifier sends it with each + // progress delivery and with the completion, so a completion that reaches the caller before + // the start response still names the operation. + string operation_token = 6; + // When the handler started the operation. Sent with the token, for the same reason. + google.protobuf.Timestamp start_time = 7; + // A Workflow in this namespace that the notifier wakes, in place of a Nexus callback. The + // Workflow gets each change of the stream as a `StreamWake` on its next Workflow Task. Exactly + // one of `callback` and `workflow_target` is set. + temporal.api.stream.v1.WorkflowWakeTarget workflow_target = 8; +} + +message AttachStreamCallbackResponse { +} + +message DetachStreamCallbackRequest { + string namespace = 1; + temporal.api.stream.v1.StreamReference stream_ref = 2; + // The request ID the callback was attached with. Detaching a callback that is not attached + // changes nothing. + string request_id = 3; + // The identity of the client detaching the callback. + string identity = 4; +} + +message DetachStreamCallbackResponse { +} + +message NotifyStreamRequest { + string namespace = 1; + temporal.api.stream.v1.StreamReference stream_ref = 2; + // Where the stream stands after the append this notification reports, in the producer's terms, + // such as a stream cursor. Passed on to the callbacks as the progress position. The server + // rejects a position longer than 1 KiB of UTF-8. + string position = 3; + // Orders the notifications of one stream. The notifier keeps the highest counter it has seen + // and ignores a lower one, so a producer may send them out of order. Must be positive. + int64 counter = 4; + // Small details passed on to the callbacks. The server rejects metadata whose keys and values + // exceed 2 KiB in total, counted as UTF-8 bytes. + map metadata = 5; + // Closes the stream: the notifier completes every attached callback with `close_result`, or + // with `close_failure`, and completes callbacks attached later right away. + bool close = 6; + // The completion result when `close` is set, such as a summary of the stream. Optional. + temporal.api.common.v1.Payload close_result = 7; + // The identity of the producer. + string identity = 8; + // Closes the stream with a failure instead of `close_result`. The notifier completes every + // callback as failed, or as canceled when the failure has canceled info. Needs `close`, and + // the server rejects a request that sets both `close_result` and `close_failure`. + temporal.api.failure.v1.Failure close_failure = 9; +} + +message NotifyStreamResponse { +} + +message DescribeStreamNotifierRequest { + string namespace = 1; + temporal.api.stream.v1.StreamReference stream_ref = 2; +} + +message DescribeStreamNotifierResponse { + // The highest counter the notifier has seen. + int64 counter = 1; + // The position that came with `counter`. + string position = 2; + // Whether the stream was closed. + bool closed = 3; + // The callbacks the notifier holds, in the order they were attached. + repeated temporal.api.stream.v1.StreamCallbackInfo callbacks = 4; + // The failure the stream closed with, when it closed with one. + temporal.api.failure.v1.Failure close_failure = 5; +} + +message PollStreamNotifierRequest { + string namespace = 1; + temporal.api.stream.v1.StreamReference stream_ref = 2; + // The poll returns when the notifier's counter is above this value, or when the stream is + // closed. A reader passes the counter of the position where it armed its read, so a + // notification that landed before the poll makes it return at once. + // + // (-- api-linter: core::0140::prepositions=disabled + // aip.dev/not-precedent: "after" names the bound the counter must pass. --) + int64 after_counter = 3; + // The identity of the client that polls. + string identity = 4; +} + +message PollStreamNotifierResponse { + // The highest counter the notifier has seen. At or below the request's `after_counter` when + // the poll reached its wait limit with no change. Then the client reads the stream again. + int64 counter = 1; + // The position that came with `counter`. + string position = 2; + // The metadata that came with `counter`. + map metadata = 3; + // Whether the stream was closed. + bool closed = 4; + // The result the stream closed with, when `closed` is set. Optional. + temporal.api.common.v1.Payload close_result = 5; + // The failure the stream closed with, when `closed` is set and it closed with one. + temporal.api.failure.v1.Failure close_failure = 6; +} diff --git a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto index 34f6a73f68..7aec05b140 100644 --- a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto +++ b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto @@ -2039,4 +2039,49 @@ service WorkflowService { value: "workflow:{workflow_execution.workflow_id}" }; } + + // AttachStreamCallback attaches a callback to the notifier of a stream. The callback receives the + // stream's progress while it is open and one completion when it closes. A Nexus handler whose + // operation returns a stream calls it with the caller's callback when the operation starts. + // + // (-- api-linter: core::0127::http-annotation=disabled + // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + rpc AttachStreamCallback (AttachStreamCallbackRequest) returns (AttachStreamCallbackResponse) { + } + + // DetachStreamCallback removes a callback from the notifier of a stream. The notifier completes + // the callback's operation as canceled and sends it no more progress. A Nexus handler calls it + // when the caller cancels the operation. + // + // (-- api-linter: core::0127::http-annotation=disabled + // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + rpc DetachStreamCallback (DetachStreamCallbackRequest) returns (DetachStreamCallbackResponse) { + } + + // NotifyStream tells the notifier of a stream that the stream moved, or that it closed. The + // notifier passes the latest notification on to every attached callback as progress, and on close + // completes them. Producers fold their notifications and keep at most one call in flight. + // + // (-- api-linter: core::0127::http-annotation=disabled + // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + rpc NotifyStream (NotifyStreamRequest) returns (NotifyStreamResponse) { + } + + // DescribeStreamNotifier returns the state of the notifier of a stream and its callbacks. It is + // for operators and tools. + // + // (-- api-linter: core::0127::http-annotation=disabled + // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + rpc DescribeStreamNotifier (DescribeStreamNotifierRequest) returns (DescribeStreamNotifierResponse) { + } + + // PollStreamNotifier waits until the notifier of a stream has a counter above `after_counter`, + // or the stream is closed. It returns with no change after a wait limit set by the server + // (`streamnotifier.pollMaxWait`). A client that reads a stream calls it after a read that + // found nothing and armed the topic, in place of another blocking read on the store. + // + // (-- api-linter: core::0127::http-annotation=disabled + // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + rpc PollStreamNotifier (PollStreamNotifierRequest) returns (PollStreamNotifierResponse) { + } } diff --git a/crates/protos/src/protos/mod.rs b/crates/protos/src/protos/mod.rs index a4c64848a1..bc07da0350 100644 --- a/crates/protos/src/protos/mod.rs +++ b/crates/protos/src/protos/mod.rs @@ -2634,6 +2634,11 @@ pub mod temporal { tonic::include_proto!("temporal.api.sdk.v1"); } } + pub mod stream { + pub mod v1 { + tonic::include_proto!("temporal.api.stream.v1"); + } + } pub mod taskqueue { pub mod v1 { tonic::include_proto!("temporal.api.taskqueue.v1"); diff --git a/crates/sdk-core-c-bridge/src/client.rs b/crates/sdk-core-c-bridge/src/client.rs index 580874fd36..516a04ad9c 100644 --- a/crates/sdk-core-c-bridge/src/client.rs +++ b/crates/sdk-core-c-bridge/src/client.rs @@ -600,6 +600,9 @@ async fn call_workflow_service( let rpc = call.rpc.to_str(); let mut client = client.clone(); match rpc { + "AttachStreamCallback" => { + rpc_call_on_trait!(client, call, WorkflowService, attach_stream_callback) + } "CountActivityExecutions" => { rpc_call_on_trait!(client, call, WorkflowService, count_activity_executions) } @@ -684,6 +687,9 @@ async fn call_workflow_service( ) } "DescribeSchedule" => rpc_call_on_trait!(client, call, WorkflowService, describe_schedule), + "DescribeStreamNotifier" => { + rpc_call_on_trait!(client, call, WorkflowService, describe_stream_notifier) + } "DescribeTaskQueue" => { rpc_call_on_trait!(client, call, WorkflowService, describe_task_queue) } @@ -707,6 +713,9 @@ async fn call_workflow_service( "DescribeWorkflowRule" => { rpc_call_on_trait!(client, call, WorkflowService, describe_workflow_rule) } + "DetachStreamCallback" => { + rpc_call_on_trait!(client, call, WorkflowService, detach_stream_callback) + } "ExecuteMultiOperation" => { rpc_call_on_trait!(client, call, WorkflowService, execute_multi_operation) } @@ -802,6 +811,9 @@ async fn call_workflow_service( "ListWorkflowRules" => { rpc_call_on_trait!(client, call, WorkflowService, list_workflow_rules) } + "NotifyStream" => { + rpc_call_on_trait!(client, call, WorkflowService, notify_stream) + } "PatchSchedule" => rpc_call_on_trait!(client, call, WorkflowService, patch_schedule), "PauseActivity" => rpc_call_on_trait!(client, call, WorkflowService, pause_activity), "PauseActivityExecution" => { @@ -835,6 +847,9 @@ async fn call_workflow_service( "PollNexusTaskQueue" => { rpc_call_on_trait!(client, call, WorkflowService, poll_nexus_task_queue) } + "PollStreamNotifier" => { + rpc_call_on_trait!(client, call, WorkflowService, poll_stream_notifier) + } "PollWorkflowExecutionUpdate" => rpc_call_on_trait!( client, call,