dex_sdk/
stream.rs

1// Copyright (c) 2026 Super Durable, Inc.
2//
3// Licensed under the Super Durable Source License 1.0.
4// You may not use this file except in compliance with the License.
5// See the LICENSE file in the repository root.
6//
7// SPDX-License-Identifier: LicenseRef-Super-Durable-1.0
8
9use std::marker::PhantomData;
10use std::sync::Arc;
11use std::time::SystemTime;
12
13use crate::{Context, HandlerResult, Value};
14
15#[derive(Debug)]
16struct StreamDefinition {
17    name: String,
18    max_estimated_bytes: i64,
19}
20
21/// Defines a typed best-effort resumable message Stream owned by one Flow type.
22///
23/// Register the same definition in exactly one [`crate::PersistenceSchema`]. Clones share
24/// identity and may be used at Client and Step call sites.
25#[derive(Clone, Debug)]
26pub struct Stream<T> {
27    definition: Arc<StreamDefinition>,
28    marker: PhantomData<fn() -> T>,
29}
30
31impl<T> Stream<T> {
32    /// Creates a Stream with a positive approximate budget shared by all instances of its Flow.
33    ///
34    /// # Panics
35    ///
36    /// Panics when `name` is empty or `max_estimated_bytes` is not positive.
37    pub fn new(name: impl Into<String>, max_estimated_bytes: i64) -> Self {
38        let name = name.into();
39        assert!(!name.is_empty(), "Stream name must not be empty");
40        assert!(
41            max_estimated_bytes > 0,
42            "Stream max_estimated_bytes must be positive"
43        );
44        Self {
45            definition: Arc::new(StreamDefinition {
46                name,
47                max_estimated_bytes,
48            }),
49            marker: PhantomData,
50        }
51    }
52
53    /// Returns the stable logical Stream name.
54    pub fn name(&self) -> &str {
55        &self.definition.name
56    }
57
58    /// Returns the approximate shared byte budget.
59    pub fn max_estimated_bytes(&self) -> i64 {
60        self.definition.max_estimated_bytes
61    }
62
63    /// Appends one message immediately from the current Step execution.
64    ///
65    /// # Errors
66    ///
67    /// Returns [`crate::HandlerError`] for RPC Contexts, duplicate writes, unregistered Streams,
68    /// encoding failures, or a failed FlowService write.
69    pub fn write(&self, context: &mut Context, value: T) -> HandlerResult<()>
70    where
71        T: Value,
72    {
73        context.write_stream(self, value)
74    }
75
76    pub(crate) fn identity(&self) -> usize {
77        Arc::as_ptr(&self.definition) as usize
78    }
79}
80
81/// Describes one retained Stream message returned by [`crate::Client::read_stream`].
82#[derive(Clone, Debug)]
83pub struct StreamMessage<T> {
84    /// Decoded application message.
85    pub value: T,
86    /// Opaque token to pass unchanged to the next read.
87    pub resume_token: String,
88    /// Server-assigned creation time.
89    pub created_time: SystemTime,
90    /// Client key or Step-generated runID#stepExecutionID key.
91    pub idempotency_key: String,
92}
dex_sdk/stream.rs

dex_sdk/
stream.rs
