qstash_rs/client/
messages.rs

1use super::Client;
2use crate::error::Result;
3use reqwest::Method;
4use serde::{Deserialize, Serialize};
5use std::collections::HashMap;
6
7/// A queued or in-flight QStash message.
8#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
9#[serde(rename_all = "camelCase")]
10pub struct Message {
11    /// Unique identifier of the message.
12    pub message_id: String,
13    /// Destination URL.
14    pub url: String,
15    /// URL Group name when the message targets one.
16    pub topic_name: Option<String>,
17    /// Endpoint label inside the URL Group.
18    pub endpoint_name: Option<String>,
19    /// API name when the message targets a provider integration.
20    pub api: Option<String>,
21    /// Custom message key if one was set by the service.
22    pub key: Option<String>,
23    /// Delivery method.
24    pub method: Option<String>,
25    /// Forwarded headers.
26    pub header: Option<HashMap<String, Vec<String>>>,
27    /// UTF-8 body when available.
28    pub body: Option<String>,
29    /// Base64-encoded body for non-UTF-8 payloads.
30    pub body_base64: Option<String>,
31    /// Maximum retry count.
32    pub max_retries: Option<u32>,
33    /// Retry delay expression when configured.
34    pub retry_delay_expression: Option<String>,
35    /// Absolute not-before timestamp in milliseconds.
36    pub not_before: Option<u64>,
37    /// Creation timestamp in milliseconds.
38    pub created_at: u64,
39    /// Callback URL.
40    pub callback: Option<String>,
41    /// Failure callback URL.
42    pub failure_callback: Option<String>,
43    /// Queue name when the message was enqueued.
44    pub queue_name: Option<String>,
45    /// Schedule identifier when created by a schedule.
46    pub schedule_id: Option<String>,
47    /// Caller IP recorded by QStash.
48    pub caller_ip: Option<String>,
49    /// User-defined label.
50    pub label: Option<String>,
51    /// Flow control key used for rate limiting.
52    pub flow_control_key: Option<String>,
53    /// Flow control rate.
54    pub rate: Option<u32>,
55    /// Flow control period in seconds.
56    pub period: Option<u32>,
57    /// Flow control parallelism.
58    pub parallelism: Option<u32>,
59}
60
61/// Result returned by bulk cancellation endpoints.
62#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
63pub struct CancelResult {
64    /// Number of messages cancelled by the operation.
65    pub cancelled: u64,
66}
67
68/// Filters supported by bulk cancellation.
69#[derive(Debug, Clone, Default, PartialEq, Eq)]
70pub struct MessageFilter {
71    /// Filter by URL Group name.
72    pub url_group_name: Option<String>,
73    /// Filter by queue name.
74    pub queue_name: Option<String>,
75    /// Filter by destination URL.
76    pub url: Option<String>,
77    /// Filter by schedule identifier.
78    pub schedule_id: Option<String>,
79    /// Filter by caller IP.
80    pub caller_ip: Option<String>,
81    /// Filter by label.
82    pub label: Option<String>,
83    /// Filter by flow control key.
84    pub flow_control_key: Option<String>,
85    /// Filter from this inclusive timestamp in milliseconds.
86    pub from_date: Option<u64>,
87    /// Filter to this inclusive timestamp in milliseconds.
88    pub to_date: Option<u64>,
89    /// Maximum number of messages to cancel.
90    pub count: Option<u32>,
91}
92
93/// Message operations.
94pub struct MessagesApi<'a> {
95    pub(crate) client: &'a Client,
96}
97
98impl MessagesApi<'_> {
99    /// Retrieves a message by identifier.
100    pub async fn get(&self, message_id: &str) -> Result<Message> {
101        self.client
102            .http
103            .send_json(
104                Method::GET,
105                &format!("v2/messages/{message_id}"),
106                &[],
107                None,
108                None,
109            )
110            .await
111    }
112
113    /// Cancels a single pending message.
114    pub async fn cancel(&self, message_id: &str) -> Result<()> {
115        self.client
116            .http
117            .send_empty(
118                Method::DELETE,
119                &format!("v2/messages/{message_id}"),
120                &[],
121                None,
122                None,
123            )
124            .await
125    }
126
127    /// Cancels a set of message identifiers.
128    pub async fn cancel_many<I, S>(&self, message_ids: I) -> Result<CancelResult>
129    where
130        I: IntoIterator<Item = S>,
131        S: AsRef<str>,
132    {
133        let mut query = Vec::new();
134        for message_id in message_ids {
135            query.push((String::from("messageIds"), message_id.as_ref().to_owned()));
136        }
137
138        self.client
139            .http
140            .send_json(Method::DELETE, "v2/messages", &query, None, None)
141            .await
142    }
143
144    /// Cancels messages matching the provided filter.
145    pub async fn cancel_matching(&self, filter: &MessageFilter) -> Result<CancelResult> {
146        self.client
147            .http
148            .send_json(
149                Method::DELETE,
150                "v2/messages",
151                &message_filter_query(filter),
152                None,
153                None,
154            )
155            .await
156    }
157
158    /// Cancels all pending messages, optionally in bounded batches.
159    pub async fn cancel_all(&self, count: Option<u32>) -> Result<CancelResult> {
160        let query = count
161            .map(|count| vec![(String::from("count"), count.to_string())])
162            .unwrap_or_default();
163
164        self.client
165            .http
166            .send_json(Method::DELETE, "v2/messages", &query, None, None)
167            .await
168    }
169}
170
171fn message_filter_query(filter: &MessageFilter) -> Vec<(String, String)> {
172    let mut query = Vec::new();
173
174    if let Some(url_group_name) = &filter.url_group_name {
175        query.push((String::from("topicName"), url_group_name.clone()));
176    }
177    if let Some(queue_name) = &filter.queue_name {
178        query.push((String::from("queueName"), queue_name.clone()));
179    }
180    if let Some(url) = &filter.url {
181        query.push((String::from("url"), url.clone()));
182    }
183    if let Some(schedule_id) = &filter.schedule_id {
184        query.push((String::from("scheduleId"), schedule_id.clone()));
185    }
186    if let Some(caller_ip) = &filter.caller_ip {
187        query.push((String::from("callerIp"), caller_ip.clone()));
188    }
189    if let Some(label) = &filter.label {
190        query.push((String::from("label"), label.clone()));
191    }
192    if let Some(flow_control_key) = &filter.flow_control_key {
193        query.push((String::from("flowControlKey"), flow_control_key.clone()));
194    }
195    if let Some(from_date) = filter.from_date {
196        query.push((String::from("fromDate"), from_date.to_string()));
197    }
198    if let Some(to_date) = filter.to_date {
199        query.push((String::from("toDate"), to_date.to_string()));
200    }
201    if let Some(count) = filter.count {
202        query.push((String::from("count"), count.to_string()));
203    }
204
205    query
206}
qstash_rs/client/messages.rs

qstash_rs/client/
messages.rs
