qstash_rs/client/
queues.rs

1use bytes::Bytes;
2use reqwest::{header::CONTENT_TYPE, Method};
3use serde::{Deserialize, Serialize};
4
5use super::{
6    http::decode_json,
7    request::{build_publish_headers, PublishRequest, PublishResponse, PublishedMessage},
8    Client,
9};
10use crate::{error::Result, Error};
11
12/// Queue metadata returned by QStash.
13#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
14#[serde(rename_all = "camelCase")]
15pub struct Queue {
16    /// Queue name.
17    pub name: String,
18    /// Creation timestamp in milliseconds.
19    pub created_at: u64,
20    /// Last update timestamp in milliseconds.
21    pub updated_at: u64,
22    /// Configured parallel consumers.
23    pub parallelism: u32,
24    /// Number of queued messages waiting to be delivered.
25    pub lag: u64,
26    /// Whether the queue is paused.
27    pub paused: bool,
28}
29
30/// Queue upsert payload.
31#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
32#[serde(rename_all = "camelCase")]
33pub struct QueueUpsertRequest {
34    /// Queue name.
35    pub queue_name: String,
36    /// Queue parallelism.
37    pub parallelism: u32,
38}
39
40/// Queue operations.
41pub struct QueuesApi<'a> {
42    pub(crate) client: &'a Client,
43}
44
45impl QueuesApi<'_> {
46    /// Lists all queues.
47    pub async fn list(&self) -> Result<Vec<Queue>> {
48        self.client
49            .http
50            .send_json(Method::GET, "v2/queues", &[], None, None)
51            .await
52    }
53
54    /// Retrieves a single queue.
55    pub async fn get(&self, queue_name: &str) -> Result<Queue> {
56        self.client
57            .http
58            .send_json(
59                Method::GET,
60                &format!("v2/queues/{queue_name}"),
61                &[],
62                None,
63                None,
64            )
65            .await
66    }
67
68    /// Creates or updates a queue.
69    pub async fn upsert(&self, request: QueueUpsertRequest) -> Result<()> {
70        let body = Bytes::from(serde_json::to_vec(&request).map_err(Error::Serialize)?);
71        let mut headers = reqwest::header::HeaderMap::new();
72        headers.insert(
73            CONTENT_TYPE,
74            reqwest::header::HeaderValue::from_static("application/json"),
75        );
76
77        self.client
78            .http
79            .send_empty(Method::POST, "v2/queues", &[], Some(headers), Some(body))
80            .await
81    }
82
83    /// Deletes a queue.
84    pub async fn delete(&self, queue_name: &str) -> Result<()> {
85        self.client
86            .http
87            .send_empty(
88                Method::DELETE,
89                &format!("v2/queues/{queue_name}"),
90                &[],
91                None,
92                None,
93            )
94            .await
95    }
96
97    /// Pauses a queue.
98    pub async fn pause(&self, queue_name: &str) -> Result<()> {
99        self.client
100            .http
101            .send_empty(
102                Method::POST,
103                &format!("v2/queues/{queue_name}/pause"),
104                &[],
105                None,
106                None,
107            )
108            .await
109    }
110
111    /// Resumes a queue.
112    pub async fn resume(&self, queue_name: &str) -> Result<()> {
113        self.client
114            .http
115            .send_empty(
116                Method::POST,
117                &format!("v2/queues/{queue_name}/resume"),
118                &[],
119                None,
120                None,
121            )
122            .await
123    }
124
125    /// Enqueues a message into a named queue.
126    pub async fn enqueue(
127        &self,
128        queue_name: &str,
129        request: PublishRequest,
130    ) -> Result<PublishResponse> {
131        let headers = build_publish_headers(&request)?;
132        let response = self
133            .client
134            .http
135            .send_bytes(
136                Method::POST,
137                &format!(
138                    "v2/enqueue/{queue_name}/{}",
139                    request.destination.path_value()
140                ),
141                &[],
142                Some(headers),
143                request.body,
144            )
145            .await?;
146
147        normalize_publish_response(response.status, response.body)
148    }
149
150    /// Convenience wrapper for enqueueing JSON.
151    pub async fn enqueue_json<T>(
152        &self,
153        queue_name: &str,
154        destination: super::request::Destination,
155        body: &T,
156    ) -> Result<PublishResponse>
157    where
158        T: Serialize,
159    {
160        let request = PublishRequest::builder(destination)
161            .json_body(body)?
162            .build();
163        self.enqueue(queue_name, request).await
164    }
165}
166
167fn normalize_publish_response(status: reqwest::StatusCode, body: Bytes) -> Result<PublishResponse> {
168    if let Ok(single) = decode_json::<PublishedMessage>(status, body.clone()) {
169        return Ok(vec![single]);
170    }
171
172    decode_json::<Vec<PublishedMessage>>(status, body)
173}
qstash_rs/client/queues.rs

qstash_rs/client/
queues.rs
