temporalio_client/
activity.rs

1mod activity_execution_info;
2mod activity_handle;
3
4use crate::errors::ClientError;
5pub use activity_execution_info::{
6    ActivityExecutionDescription, ActivityExecutionInfo, ActivityExecutionInfoLike,
7    ActivityExecutionStatus, PendingActivityState,
8};
9pub use activity_handle::ActivityHandle;
10use futures_util::{Stream, StreamExt};
11use std::{
12    collections::VecDeque,
13    pin::Pin,
14    task::{Context, Poll},
15};
16use temporalio_common::{
17    protos::temporal::api::{
18        activity::v1::ActivityExecutionListInfo,
19        workflowservice::v1::{
20            CountActivityExecutionsResponse, count_activity_executions_response,
21        },
22    },
23    search_attributes::{SearchAttributeError, SearchAttributeValue},
24};
25
26/// A stream of activity executions from a list query.
27/// Internally paginates through results from the server.
28pub struct ListActivitiesStream {
29    inner: Pin<Box<dyn Stream<Item = Result<Vec<ActivityExecutionListInfo>, ClientError>> + Send>>,
30    buffer: VecDeque<ActivityExecutionListInfo>,
31}
32
33impl ListActivitiesStream {
34    pub(crate) fn new(
35        stream: impl Stream<Item = Result<Vec<ActivityExecutionListInfo>, ClientError>> + Send + 'static,
36    ) -> Self {
37        Self {
38            inner: Box::pin(stream),
39            buffer: VecDeque::new(),
40        }
41    }
42}
43
44impl Stream for ListActivitiesStream {
45    type Item = Result<ActivityExecutionInfo, ClientError>;
46
47    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
48        loop {
49            if let Some(info) = self.buffer.pop_front() {
50                return Poll::Ready(Some(Ok(info.into())));
51            }
52            match self.inner.poll_next_unpin(cx) {
53                Poll::Ready(Some(Ok(items))) => {
54                    self.buffer = items.into();
55                }
56                Poll::Ready(Some(Err(e))) => {
57                    return Poll::Ready(Some(Err(e)));
58                }
59                Poll::Ready(None) => {
60                    return Poll::Ready(None);
61                }
62                Poll::Pending => {
63                    return Poll::Pending;
64                }
65            }
66        }
67    }
68}
69
70/// Result of an activity count operation.
71///
72/// If the query includes a group-by clause, `groups` will contain the aggregated
73/// counts and `count` will be the sum of all group counts.
74#[derive(Debug, Clone)]
75pub struct ActivityExecutionCount {
76    count: usize,
77    groups: Vec<ActivityExecutionCountAggregationGroup>,
78}
79
80impl ActivityExecutionCount {
81    pub(crate) fn from_response(resp: CountActivityExecutionsResponse) -> Self {
82        Self {
83            count: resp.count as usize,
84            groups: resp
85                .groups
86                .into_iter()
87                .map(ActivityExecutionCountAggregationGroup::from_proto)
88                .collect(),
89        }
90    }
91
92    /// The approximate number of activities matching the query.
93    /// If grouping was applied, this is the sum of all group counts.
94    pub fn count(&self) -> usize {
95        self.count
96    }
97
98    /// The groups if the query had a group-by clause, or empty if not.
99    pub fn groups(&self) -> &[ActivityExecutionCountAggregationGroup] {
100        &self.groups
101    }
102}
103
104/// Aggregation group from an activity count query with a group-by clause.
105#[derive(Debug, Clone)]
106pub struct ActivityExecutionCountAggregationGroup {
107    raw: count_activity_executions_response::AggregationGroup,
108}
109
110impl ActivityExecutionCountAggregationGroup {
111    fn from_proto(proto: count_activity_executions_response::AggregationGroup) -> Self {
112        Self { raw: proto }
113    }
114
115    /// Retrieve a typed group value at `index`.
116    ///
117    ///  Returns `None` if the index is out of bounds or deserialization fails.
118    ///  Use [`Self::try_get`] for explicit error handling.
119    pub fn get<T: SearchAttributeValue>(&self, index: usize) -> Option<T> {
120        self.try_get(index).ok().flatten()
121    }
122
123    /// Retrieve a typed group value at `index`, preserving deserialization
124    /// errors.
125    ///
126    /// Returns `Ok(None)` if the index is out of bounds and `Err` if the
127    /// payload cannot be deserialized.
128    pub fn try_get<T: SearchAttributeValue>(
129        &self,
130        index: usize,
131    ) -> Result<Option<T>, SearchAttributeError> {
132        match self.raw.group_values.get(index) {
133            Some(payload) => T::from_search_attribute_payload(payload).map(Some),
134            None => Ok(None),
135        }
136    }
137
138    /// The approximate number of workflows matching for this group.
139    pub fn count(&self) -> usize {
140        self.raw.count as usize
141    }
142}
temporalio_client/activity.rs

temporalio_client/
activity.rs
