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    stream_capacity_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 `stream_capacity_bytes` is not positive.
37    pub fn new(name: impl Into<String>, stream_capacity_bytes: i64) -> Self {
38        let name = name.into();
39        assert!(!name.is_empty(), "Stream name must not be empty");
40        assert!(
41            stream_capacity_bytes > 0,
42            "Stream stream_capacity_bytes must be positive"
43        );
44        Self {
45            definition: Arc::new(StreamDefinition {
46                name,
47                stream_capacity_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 stream_capacity_bytes(&self) -> i64 {
60        self.definition.stream_capacity_bytes
61    }
62
63    /// Appends one message immediately from the current Step execution.
64    ///
65    /// # Errors
66    ///
67    /// Returns [`crate::HandlerError`] for RPC or Flow timeout Contexts, unregistered Streams,
68    /// encoding failures, cancellation, or a closed Worker output stream. Dex storage failures are
69    /// not acknowledged.
70    pub fn write(&self, context: &mut Context, value: T) -> HandlerResult<()>
71    where
72        T: Value,
73    {
74        context.write_stream(self, value)
75    }
76
77    pub(crate) fn identity(&self) -> usize {
78        Arc::as_ptr(&self.definition) as usize
79    }
80}
81
82/// Describes one retained Stream message returned by [`crate::Client::read_stream`].
83#[derive(Clone, Debug)]
84pub struct StreamMessage<T> {
85    /// Decoded application message.
86    pub value: T,
87    /// Opaque token to pass unchanged to the next read.
88    pub resume_token: String,
89    /// Server-assigned creation time.
90    pub created_time: SystemTime,
91    /// Client-provided source or Step-generated `#stepExecutionID` source.
92    pub source: String,
93}
dex_sdk/stream.rs

dex_sdk/
stream.rs
