From 9afad0483cd2a10ca8e2cc6f4793f7fb994c254e Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 17 Oct 2026 00:21:22 -0700 Subject: [PATCH 1/2] Vendored the stream API from the api branch ja8. --- .../temporal/api/enums/v1/stream.proto | 32 ++++ .../temporal/api/stream/v1/message.proto | 122 ++++++++++++++ .../workflowservice/v1/request_response.proto | 158 ++++++++++++++++++ .../api/workflowservice/v1/service.proto | 78 +++++++++ 4 files changed, 390 insertions(+) create mode 100644 crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto create mode 100644 crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto 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..4f53e16d8b --- /dev/null +++ b/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto @@ -0,0 +1,32 @@ +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; + // A standalone Activity owns the stream. The Activity closes the stream itself, with a result + // or a failure. If the Activity run reaches a terminal status with no close and callbacks + // are attached, the stream notifier closes the stream within + // `streamnotifier.ownerCheckInterval`. A completed run gives no result. A failed, canceled, + // terminated or timed-out run gives the run's own failure. + STREAM_OWNER_KIND_ACTIVITY = 2; + // An Activity that a Workflow scheduled owns the stream. `workflow_id` names the Workflow, + // `run_id` the run that scheduled the Activity, and `activity_id` the Activity. The stream + // closes when the Workflow sees the Activity end. If the Activity is no longer pending in the + // run and no close came, the stream notifier closes the stream with a failure, after + // `streamnotifier.ownerStartGrace` from its start and `streamnotifier.ownerCloseGrace` from + // the first check that found the Activity gone. + STREAM_OWNER_KIND_WORKFLOW_ACTIVITY = 3; +} 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..debeabfd45 --- /dev/null +++ b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto @@ -0,0 +1,122 @@ +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`. The Workflow that scheduled the + // owning Activity, for `STREAM_OWNER_KIND_WORKFLOW_ACTIVITY`. Empty for other kinds. + string workflow_id = 2; + // For `STREAM_OWNER_KIND_WORKFLOW`, 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. For `STREAM_OWNER_KIND_ACTIVITY`, the id of the owning Activity run. For + // `STREAM_OWNER_KIND_WORKFLOW_ACTIVITY`, the id of the run that scheduled the Activity, not + // the chain's first run. An empty run id is refused. + string run_id = 3; + // The stream's topic within its owner. + string topic = 4; + // The owning standalone Activity, for `STREAM_OWNER_KIND_ACTIVITY`. The owning Activity's id + // in the scheduling run, for `STREAM_OWNER_KIND_WORKFLOW_ACTIVITY`. Empty for other kinds. + string activity_id = 5; +} + +// 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 as one of the +// `attachments` of 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 subscribed after 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; +} + +// One stream's entry in a NotifyStreams request: the same fields as a NotifyStream request that +// does not close the stream. +message StreamNotification { + StreamReference stream_ref = 1; + // Where the stream stands, in the producer's terms. At most 1 KiB of UTF-8. + string position = 2; + // The newest counter of the stream. Must be positive. The notifier ignores a counter at or + // below the one it holds. + int64 counter = 3; + // Small details passed on to the callbacks. At most 2 KiB in total, counted as UTF-8 bytes. + map metadata = 4; +} + +// The answer to one entry of a NotifyStreams request. +message StreamNotificationResult { + // The gRPC status code of the entry, as a NotifyStream call for it would answer. Zero (OK) + // when the server applied the entry, or dropped it since the stream has no notifier. + int32 code = 1; + // Why the server refused the entry. Empty when `code` is zero. + string message = 2; +} 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..f1fdb8e2ab 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,160 @@ 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; + // Attaches the callback idle. An idle callback gets no progress until WaitStreamCallback asks + // for one wake, and it goes back to idle after that delivery. It still gets the completion + // when the stream closes. A Workflow reader attaches its `workflow_target` idle, so a busy + // reader gets no wake. + bool idle = 9; +} + +message AttachStreamCallbackResponse { +} + +message WaitStreamCallbackRequest { + string namespace = 1; + temporal.api.stream.v1.StreamReference stream_ref = 2; + // The request ID the idle callback was attached with. + string request_id = 3; + // The notifier delivers progress to the callback once, when its counter is above this value. + // If the counter is above it already, the delivery goes out at once. A reader passes the + // counter of the position its last read reached. + // + // (-- api-linter: core::0140::prepositions=disabled + // aip.dev/not-precedent: "after" names the bound the counter must pass. --) + int64 after_counter = 4; + // The identity of the caller. + string identity = 5; +} + +message WaitStreamCallbackResponse { +} + +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 its last read reached, 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; +} + +message NotifyStreamsRequest { + string namespace = 1; + // One entry per stream. The server rejects a request with more than + // `streamnotifier.maxNotifyBatch` entries, or with two entries for one stream. + repeated temporal.api.stream.v1.StreamNotification notifications = 2; + // The identity of the caller, such as a stream driver. + string identity = 3; +} + +message NotifyStreamsResponse { + // One result per entry of the request, in the same order. Empty only from a server that + // fails the whole call on a bad entry, so an empty list on success means every entry applied. + repeated temporal.api.stream.v1.StreamNotificationResult results = 1; +} 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..ff94563af8 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,82 @@ 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. It + // starts the notifier when the stream has none. + // + // (-- 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) { + } + + // WaitStreamCallback asks the notifier of a stream to wake one idle callback once, when the + // notifier's counter passes `after_counter`. If it passed already, the wake goes out at once. + // After the delivery the callback is idle again. A reader calls it when a read finds nothing. + // The handler of a stream operation calls it for its caller, and a Workflow reader's Worker + // for its `workflow_target`. So a busy stream that the reader keeps up with sends the reader + // nothing. Waiting on a callback that is not idle changes nothing. + // + // (-- api-linter: core::0127::http-annotation=disabled + // aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --) + rpc WaitStreamCallback (WaitStreamCallbackRequest) returns (WaitStreamCallbackResponse) { + option (nexusannotations.v1.operation).tags = "exposed"; + } + + // 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. + // + // A notification that does not close the stream starts nothing. When the stream has no + // notifier, nobody waits on it, and the server drops the notification. A close starts a notifier + // when the stream has none, so a callback that attaches later still gets the close outcome. + // + // (-- 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, in place of another blocking read on the store. It starts the notifier when the + // stream has none. The client reads the stream once more after the poll starts, since a + // notification that came before the notifier existed was dropped. + // + // (-- api-linter: core::0127::http-annotation=disabled + // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + rpc PollStreamNotifier (PollStreamNotifierRequest) returns (PollStreamNotifierResponse) { + } + + // NotifyStreams tells the notifiers of many streams in one namespace that they moved. It acts as + // one NotifyStream without close per entry. A stream with no notifier drops its entry, as + // NotifyStream does. A driver that reads the store's notification log calls it with the newest + // counter of each stream in a batch. The server checks and applies each entry on its own and + // answers each in `results`, so a bad entry doesn't hold up the others. The call fails only for + // the request as a whole, such as an empty batch or one that is too large. Sending an entry + // again is safe, since a notifier ignores a counter at or below the one it holds. + // + // (-- api-linter: core::0127::http-annotation=disabled + // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + rpc NotifyStreams (NotifyStreamsRequest) returns (NotifyStreamsResponse) { + } } From c72c739a33878e0b5ab70a5dec6c217a54f517a9 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 17 Oct 2026 00:21:22 -0700 Subject: [PATCH 2/2] Added the client lines for the stream notifier RPCs. --- crates/client/src/grpc.rs | 64 ++++++++++++++++++++++++++ crates/common/build.rs | 2 + crates/protos/build.rs | 1 + crates/protos/src/protos/mod.rs | 5 ++ crates/sdk-core-c-bridge/src/client.rs | 21 +++++++++ 5 files changed, 93 insertions(+) diff --git a/crates/client/src/grpc.rs b/crates/client/src/grpc.rs index 1c3d5b0a6d..b6d8e61c7f 100644 --- a/crates/client/src/grpc.rs +++ b/crates/client/src/grpc.rs @@ -1878,6 +1878,70 @@ 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); + } + ); + ( + wait_stream_callback, + WaitStreamCallbackRequest, + WaitStreamCallbackResponse, + |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); + } + ); + ( + notify_streams, + NotifyStreamsRequest, + NotifyStreamsResponse, + |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 346d506c62..5c55b0a43a 100644 --- a/crates/protos/build.rs +++ b/crates/protos/build.rs @@ -40,6 +40,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/src/protos/mod.rs b/crates/protos/src/protos/mod.rs index 5ccf2f2d1c..2a4220b87e 100644 --- a/crates/protos/src/protos/mod.rs +++ b/crates/protos/src/protos/mod.rs @@ -2617,6 +2617,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..dbf51630d5 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,12 @@ 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) + } + "WaitStreamCallback" => { + rpc_call_on_trait!(client, call, WorkflowService, wait_stream_callback) + } "ExecuteMultiOperation" => { rpc_call_on_trait!(client, call, WorkflowService, execute_multi_operation) } @@ -802,6 +814,12 @@ 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) + } + "NotifyStreams" => { + rpc_call_on_trait!(client, call, WorkflowService, notify_streams) + } "PatchSchedule" => rpc_call_on_trait!(client, call, WorkflowService, patch_schedule), "PauseActivity" => rpc_call_on_trait!(client, call, WorkflowService, pause_activity), "PauseActivityExecution" => { @@ -835,6 +853,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,