[Brook] Added stream store calls to the C bridge.

crates/sdk-core-c-bridge/Cargo.toml

@@ -47,7 +47,12 @@ features = ["experimental"]
 [dependencies.temporalio-sdk-core]
 path = "../sdk-core"
 version = "=0.10.0"
-features = ["ephemeral-server", "otel"]
+features = ["ephemeral-server", "otel", "streams"]
+
+[dependencies.temporalio-streams]
+path = "../streams"
+version = "0.1"
+features = ["redis"]
 
 [dependencies.temporalio-common]
 path = "../common"

crates/sdk-core-c-bridge/include/temporal-sdk-core-c-bridge.h

@@ -108,6 +108,12 @@ typedef struct TemporalCoreRuntime TemporalCoreRuntime;
 
 typedef struct TemporalCoreSlotReserveCompletionCtx TemporalCoreSlotReserveCompletionCtx;
 
+/**
+ * One process's stream store, serving the stream calls lang makes. Workers take it in their
+ * options, so a Workflow's output and outside producers share it.
+ */
+typedef struct TemporalCoreStreamStore TemporalCoreStreamStore;
+
 typedef struct TemporalCoreWorker TemporalCoreWorker;
 
 typedef struct TemporalCoreWorkerReplayPusher TemporalCoreWorkerReplayPusher;
@@ -547,6 +553,22 @@ typedef struct TemporalCoreRuntimeOptions {
   bool disable_environment_info;
 } TemporalCoreRuntimeOptions;
 
+/**
+ * If success or fail are not null, they must be manually freed when done.
+ */
+typedef void (*TemporalCoreStreamStoreNewCallback)(void *user_data,
+                                                   struct TemporalCoreStreamStore *success,
+                                                   const struct TemporalCoreByteArray *fail);
+
+/**
+ * If success or failure are not null, they must be manually freed when done. Exactly one is
+ * set. Success is the serialized response, and failure a serialized
+ * `coresdk.streams.StreamFailure`.
+ */
+typedef void (*TemporalCoreStreamStoreCallCallback)(void *user_data,
+                                                    const struct TemporalCoreByteArray *success,
+                                                    const struct TemporalCoreByteArray *failure);
+
 typedef struct TemporalCoreTestServerOptions {
   /**
    * Empty means default behavior
@@ -891,6 +913,12 @@ typedef struct TemporalCoreWorkerOptions {
    * a workflow task. Zero disables eager activity execution.
    */
   uint32_t max_eager_activity_reservations_per_workflow_task;
+  /**
+   * The store this worker stages and promotes workflow stream output in, or null for none. It
+   * must outlive the worker.
+   * NOTE: Experimental
+   */
+  const struct TemporalCoreStreamStore *stream_store;
 } TemporalCoreWorkerOptions;
 
 /**
@@ -1073,6 +1101,30 @@ uint64_t temporal_core_forwarded_log_timestamp_millis(const struct TemporalCoreF
 
 struct TemporalCoreByteArrayRef temporal_core_forwarded_log_fields_json(const struct TemporalCoreForwardedLog *log);
 
+/**
+ * Connects to the store `config` names, a serialized `coresdk.streams.StreamStoreConfig`. The
+ * store asks `client`'s server about streams' owners. Client must live as long as the store, and
+ * config and user data must live through the callback.
+ */
+void temporal_core_stream_store_new(struct TemporalCoreConnection *client,
+                                    struct TemporalCoreByteArrayRef config,
+                                    void *user_data,
+                                    TemporalCoreStreamStoreNewCallback callback);
+
+void temporal_core_stream_store_free(struct TemporalCoreStreamStore *store);
+
+/**
+ * Makes one stream call. `rpc` names a `coresdk.streams.StreamService` method and `request` is
+ * its serialized request. Cancelling the token drops the call, which a read uses to stop
+ * waiting. Store, request and user data must live through the callback.
+ */
+void temporal_core_stream_store_call(struct TemporalCoreStreamStore *store,
+                                     struct TemporalCoreByteArrayRef rpc,
+                                     struct TemporalCoreByteArrayRef request,
+                                     const struct TemporalCoreCancellationToken *cancellation_token,
+                                     void *user_data,
+                                     TemporalCoreStreamStoreCallCallback callback);
+
 /**
  * Runtime must live as long as server. Options and user data must live through
  * callback.

crates/sdk-core-c-bridge/src/lib.rs

@@ -13,6 +13,7 @@ pub mod envconfig;
 pub mod metric;
 pub mod random;
 pub mod runtime;
+pub mod streams;
 pub mod testing;
 pub mod worker;
 

crates/sdk-core-c-bridge/src/streams.rs

@@ -0,0 +1,133 @@
+use crate::{
+    ByteArray, ByteArrayRef, CancellationToken, UserDataHandle, client::Connection,
+    runtime::Runtime,
+};
+use prost::Message;
+use std::sync::Arc;
+use temporalio_sdk_core::streams::{StreamService, connect_stream_service, proto};
+
+/// One process's stream store, serving the stream calls lang makes. Workers take it in their
+/// options, so a Workflow's output and outside producers share it.
+pub struct StreamStore {
+    pub(crate) runtime: Runtime,
+    pub(crate) service: Arc<StreamService>,
+}
+
+/// If success or fail are not null, they must be manually freed when done.
+pub type StreamStoreNewCallback = unsafe extern "C" fn(
+    user_data: *mut libc::c_void,
+    success: *mut StreamStore,
+    fail: *const ByteArray,
+);
+
+/// Connects to the store `config` names, a serialized `coresdk.streams.StreamStoreConfig`. The
+/// store asks `client`'s server about streams' owners. Client must live as long as the store, and
+/// config and user data must live through the callback.
+#[unsafe(no_mangle)]
+pub extern "C" fn temporal_core_stream_store_new(
+    client: *mut Connection,
+    config: ByteArrayRef,
+    user_data: *mut libc::c_void,
+    callback: StreamStoreNewCallback,
+) {
+    let client = unsafe { &*client };
+    let mut runtime = client.runtime.clone();
+    let config = match proto::StreamStoreConfig::decode(config.to_slice()) {
+        Ok(config) => config,
+        Err(err) => {
+            let fail = runtime.alloc_utf8(&format!("Invalid stream store config: {err}"));
+            unsafe { callback(user_data, std::ptr::null_mut(), fail.into_raw()) };
+            return;
+        }
+    };
+    let connection = client.core.clone();
+    let user_data = UserDataHandle(user_data);
+    client.runtime.core.tokio_handle().spawn(async move {
+        match connect_stream_service(config, connection).await {
+            Ok(service) => {
+                let store = Box::into_raw(Box::new(StreamStore {
+                    runtime: runtime.clone(),
+                    service,
+                }));
+                unsafe { callback(user_data.into(), store, std::ptr::null()) };
+            }
+            Err(err) => {
+                let fail =
+                    ByteArray::from_utf8(format!("Could not connect the stream store: {err}"));
+                unsafe { callback(user_data.into(), std::ptr::null_mut(), fail.into_raw()) };
+            }
+        }
+    });
+}
+
+#[unsafe(no_mangle)]
+pub extern "C" fn temporal_core_stream_store_free(store: *mut StreamStore) {
+    unsafe {
+        let _ = Box::from_raw(store);
+    }
+}
+
+/// The failure a call answers with when its caller cancelled it. A write may or may not have
+/// landed, but a read only stopped, so it gets a message that says so.
+fn cancelled(rpc: &str) -> proto::StreamFailure {
+    let message = match rpc {
+        "Read" | "Latest" => "Cancelled before the read answered",
+        _ => "Cancelled",
+    };
+    proto::StreamFailure {
+        kind: proto::StreamFailureKind::OutcomeUnknown as i32,
+        message: message.to_string(),
+        cursor: String::new(),
+    }
+}
+
+/// If success or failure are not null, they must be manually freed when done. Exactly one is
+/// set. Success is the serialized response, and failure a serialized
+/// `coresdk.streams.StreamFailure`.
+pub type StreamStoreCallCallback = unsafe extern "C" fn(
+    user_data: *mut libc::c_void,
+    success: *const ByteArray,
+    failure: *const ByteArray,
+);
+
+/// Makes one stream call. `rpc` names a `coresdk.streams.StreamService` method and `request` is
+/// its serialized request. Cancelling the token drops the call, which a read uses to stop
+/// waiting. Store, request and user data must live through the callback.
+#[unsafe(no_mangle)]
+pub extern "C" fn temporal_core_stream_store_call(
+    store: *mut StreamStore,
+    rpc: ByteArrayRef,
+    request: ByteArrayRef,
+    cancellation_token: *const CancellationToken,
+    user_data: *mut libc::c_void,
+    callback: StreamStoreCallCallback,
+) {
+    let store = unsafe { &*store };
+    let rpc = rpc.to_string();
+    let request = request.to_vec();
+    let cancel_token = unsafe { cancellation_token.as_ref() }.map(|v| v.token.clone());
+    let service = store.service.clone();
+    let user_data = UserDataHandle(user_data);
+    store.runtime.core.tokio_handle().spawn(async move {
+        let call = service.call(&rpc, &request);
+        let result = if let Some(cancel_token) = cancel_token {
+            tokio::select! {
+                _ = cancel_token.cancelled() => Err(cancelled(&rpc)),
+                result = call => result,
+            }
+        } else {
+            call.await
+        };
+        let (success, failure) = match result {
+            Ok(response) => (
+                ByteArray::from_vec(response).into_raw(),
+                std::ptr::null_mut(),
+            ),
+            Err(failure) => (
+                std::ptr::null_mut(),
+                ByteArray::from_vec(failure.encode_to_vec()).into_raw(),
+            ),
+        };
+        unsafe { callback(user_data.into(), success, failure) };
+    });
+}

crates/sdk-core-c-bridge/src/tests/context.rs

@@ -8,6 +8,10 @@ use crate::{
         Runtime, RuntimeInfoArray, RuntimeOptions, RuntimeOrFail, temporal_core_byte_array_free,
         temporal_core_runtime_free, temporal_core_runtime_new,
     },
+    streams::{
+        StreamStore, temporal_core_stream_store_call, temporal_core_stream_store_free,
+        temporal_core_stream_store_new,
+    },
     testing::{
         DevServerOptions, EphemeralServer, TestServerOptions, temporal_core_ephemeral_server_free,
         temporal_core_ephemeral_server_shutdown, temporal_core_ephemeral_server_start_dev_server,
@@ -449,6 +453,69 @@ impl Context {
         self.wait_for_operation_result().map(|r| *r)
     }
 
+    /// Builds a stream store from the serialized `config` on this context's client.
+    pub fn stream_store_new(self: &Arc<Self>, config: Vec<u8>) -> anyhow::Result<StreamStorePtr> {
+        let (runtime, client) = {
+            let mut guard = self.wait_for_available()?;
+            let client = guard.client;
+            if client.is_null() {
+                return Err(anyhow!("Client is null"));
+            }
+            guard.operation_state = ContextOperationState::InProgress;
+            (guard.runtime, client)
+        };
+        let config_ref: ByteArrayRef = config.as_slice().into();
+        let user_data = Box::into_raw(Box::new(CallbackUserData {
+            data: runtime,
+            context: Arc::downgrade(self),
+            _allocations: Box::new(config),
+        })) as *mut libc::c_void;
+        temporal_core_stream_store_new(client, config_ref, user_data, stream_store_new_callback);
+        self.wait_for_operation_result().map(|r| *r)
+    }
+
+    /// Makes one stream call, answering with the response or the serialized failure.
+    pub fn stream_store_call(
+        self: &Arc<Self>,
+        store: &StreamStorePtr,
+        rpc: &str,
+        request: Vec<u8>,
+    ) -> anyhow::Result<Result<Vec<u8>, Vec<u8>>> {
+        self.stream_store_call_with_token(store, rpc, request, std::ptr::null())
+    }
+
+    /// Makes a stream call that `cancellation_token`, when not null, can cancel.
+    pub fn stream_store_call_with_token(
+        self: &Arc<Self>,
+        store: &StreamStorePtr,
+        rpc: &str,
+        request: Vec<u8>,
+        cancellation_token: *const crate::CancellationToken,
+    ) -> anyhow::Result<Result<Vec<u8>, Vec<u8>>> {
+        let runtime = {
+            let mut guard = self.wait_for_available()?;
+            guard.operation_state = ContextOperationState::InProgress;
+            guard.runtime
+        };
+        let rpc = rpc.to_string();
+        let rpc_ref: ByteArrayRef = rpc.as_str().into();
+        let request_ref: ByteArrayRef = request.as_slice().into();
+        let user_data = Box::into_raw(Box::new(CallbackUserData {
+            data: runtime,
+            context: Arc::downgrade(self),
+            _allocations: Box::new((rpc, request)),
+        })) as *mut libc::c_void;
+        temporal_core_stream_store_call(
+            store.0,
+            rpc_ref,
+            request_ref,
+            cancellation_token,
+            user_data,
+            stream_store_call_callback,
+        );
+        self.wait_for_operation_result().map(|r| *r)
+    }
+
     fn wait_while(
         &self,
         condition: impl FnMut(&mut InnerContext) -> bool,
@@ -710,3 +777,70 @@ impl From<&HashMap<String, Vec<u8>>> for GrpcMetadataHolder {
         }
     }
 }
+
+/// A stream store the test owns, freed when dropped.
+#[derive(Debug)]
+pub struct StreamStorePtr(*mut StreamStore);
+
+unsafe impl Send for StreamStorePtr {}
+
+impl Drop for StreamStorePtr {
+    fn drop(&mut self) {
+        temporal_core_stream_store_free(self.0);
+    }
+}
+
+extern "C" fn stream_store_new_callback(
+    user_data: *mut libc::c_void,
+    mut store: *mut StreamStore,
+    mut fail: *const ByteArray,
+) {
+    let user_data = unsafe { Box::from_raw(user_data as *mut CallbackUserData<(), Context>) };
+    if let Some(context) = user_data.context.upgrade() {
+        let _ = context.complete_operation_catch_unwind(|guard| {
+            if let Some(fail) = byte_array_to_string(guard.runtime, std::mem::take(&mut fail)) {
+                ContextOperationState::CallbackError(anyhow!("Stream store failed: {fail}"))
+            } else {
+                let store = StreamStorePtr(std::mem::take(&mut store));
+                ContextOperationState::CallbackOk(Some(Box::new(store) as _))
+            }
+        });
+    }
+    if !fail.is_null() {
+        temporal_core_byte_array_free(std::ptr::null_mut(), fail);
+    }
+    if !store.is_null() {
+        temporal_core_stream_store_free(store);
+    }
+}
+
+extern "C" fn stream_store_call_callback(
+    user_data: *mut libc::c_void,
+    mut success: *const ByteArray,
+    mut failure: *const ByteArray,
+) {
+    let user_data = unsafe { Box::from_raw(user_data as *mut CallbackUserData<(), Context>) };
+    if let Some(context) = user_data.context.upgrade() {
+        let _ = context.complete_operation_catch_unwind(|guard| {
+            let success = byte_array_to_vec(guard.runtime, std::mem::take(&mut success));
+            let failure = byte_array_to_vec(guard.runtime, std::mem::take(&mut failure));
+            let result: Result<Vec<u8>, Vec<u8>> = match (success, failure) {
+                (Some(success), None) => Ok(success),
+                (None, Some(failure)) => Err(failure),
+                other => {
+                    return ContextOperationState::CallbackError(anyhow!(
+                        "A stream call must answer with exactly one of success and failure, got \
+                         {other:?}"
+                    ));
+                }
+            };
+            ContextOperationState::CallbackOk(Some(Box::new(result) as _))
+        });
+    }
+    if !success.is_null() {
+        temporal_core_byte_array_free(std::ptr::null_mut(), success);
+    }
+    if !failure.is_null() {
+        temporal_core_byte_array_free(std::ptr::null_mut(), failure);
+    }
+}