1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
use crate::types::RedisBytes;
use crate::types::{RsmqMessage, RsmqQueueAttributes};
use crate::RsmqResult;
use core::convert::TryFrom;
use std::future::Future;
use std::time::Duration;
pub trait RsmqConnection {
/// Change the hidden time of a already sent message.
///
/// `hidden` has a max time of 9_999_999 for compatibility reasons to this library JS version counterpart
fn change_message_visibility(
&mut self,
qname: &str,
message_id: &str,
hidden: Duration,
) -> impl Future<Output = RsmqResult<()>> + Send;
/// Creates a new queue. Attributes can be later modified with "set_queue_attributes" method
///
/// hidden: Time the messages will be hidden when they are received with the "receive_message" method. It
/// has a max time of 9_999_999 for compatibility reasons to this library JS version counterpart
///
/// delay: Time the messages will be delayed before being delivered
///
/// maxsize: Maximum size in bytes of each message in the queue. Needs to be between 1024 or 65536 or -1 (unlimited
/// size)
fn create_queue(
&mut self,
qname: &str,
hidden: Option<Duration>,
delay: Option<Duration>,
maxsize: Option<i64>,
) -> impl Future<Output = RsmqResult<()>> + Send;
/// Deletes a message from the queue.
///
/// Important to use when you are using receive_message.
fn delete_message(
&mut self,
qname: &str,
id: &str,
) -> impl Future<Output = RsmqResult<bool>> + Send;
/// Deletes the queue and all the messages on it
fn delete_queue(&mut self, qname: &str) -> impl Future<Output = RsmqResult<()>> + Send;
/// Returns the queue attributes and statistics
fn get_queue_attributes(
&mut self,
qname: &str,
) -> impl Future<Output = RsmqResult<RsmqQueueAttributes>> + Send;
/// Returns a list of queues in the namespace
fn list_queues(&mut self) -> impl Future<Output = RsmqResult<Vec<String>>> + Send;
/// Deletes and returns a message. Be aware that using this you may end with deleted & unprocessed messages.
fn pop_message<E: TryFrom<RedisBytes, Error = Vec<u8>>>(
&mut self,
qname: &str,
) -> impl Future<Output = RsmqResult<Option<RsmqMessage<E>>>> + Send;
/// Returns a message. The message stays hidden for some time (defined by "hidden" argument or the queue
/// settings). After that time, the message will be redelivered. In order to avoid the redelivery, you need to use
/// the "delete_message" after this function.
///
/// `hidden` has a max time of 9_999_999 for compatibility reasons to this library JS version counterpart.
fn receive_message<E: TryFrom<RedisBytes, Error = Vec<u8>>>(
&mut self,
qname: &str,
hidden: Option<Duration>,
) -> impl Future<Output = RsmqResult<Option<RsmqMessage<E>>>> + Send;
/// Sends a message to the queue. The message will be delayed some time (controlled by the "delayed" argument or
/// the queue settings) before being delivered to a client.
fn send_message<E: Into<RedisBytes> + Send>(
&mut self,
qname: &str,
message: E,
delay: Option<Duration>,
) -> impl Future<Output = RsmqResult<String>> + Send;
/// Sends a batch of messages to the queue. Returns the assigned message IDs in input order.
///
/// The default impl loops `send_message`, which means a mid-batch error leaves whatever
/// succeeded before that point in the queue. Implementations like [`crate::Rsmq`] and
/// [`crate::PooledRsmq`] override this with a single Lua script for full atomicity (either
/// all messages land or none do) and at most one realtime PUBLISH per call.
fn send_message_batch<E: Into<RedisBytes> + Send>(
&mut self,
qname: &str,
messages: Vec<E>,
delay: Option<Duration>,
) -> impl Future<Output = RsmqResult<Vec<String>>> + Send
where
Self: Send,
{
async move {
if messages.is_empty() {
return Ok(Vec::new());
}
let mut ids = Vec::with_capacity(messages.len());
for m in messages {
ids.push(self.send_message(qname, m, delay).await?);
}
Ok(ids)
}
}
/// Receives up to `max_count` messages, sharing the same hidden duration. Returns at most
/// `max_count` messages but may return fewer (or zero) if not enough are visible.
///
/// The default impl loops `receive_message`. Implementations like [`crate::Rsmq`] and
/// [`crate::PooledRsmq`] override this with a single Lua script that picks all messages
/// in one ZRANGE BYSCORE pass.
fn receive_message_batch<E: TryFrom<RedisBytes, Error = Vec<u8>> + Send>(
&mut self,
qname: &str,
hidden: Option<Duration>,
max_count: u32,
) -> impl Future<Output = RsmqResult<Vec<RsmqMessage<E>>>> + Send
where
Self: Send,
{
async move {
if max_count == 0 {
return Ok(Vec::new());
}
let mut out = Vec::with_capacity(max_count as usize);
for _ in 0..max_count {
match self.receive_message(qname, hidden).await? {
Some(m) => out.push(m),
None => break,
}
}
Ok(out)
}
}
/// Modify the queue attributes. Keep in mind that "hidden" and "delay" can be overwritten when the message
/// is sent. "hidden" can be changed by the method "change_message_visibility"
///
/// hidden: Time the messages will be hidden when they are received with the "receive_message" method. It
/// has a max time of 9_999_999 for compatibility reasons to this library JS version counterpart
///
/// delay: Time the messages will be delayed before being delivered
///
/// maxsize: Maximum size in bytes of each message in the queue. Needs to be between 1024 or 65536 or -1 (unlimited
/// size)
fn set_queue_attributes(
&mut self,
qname: &str,
hidden: Option<Duration>,
delay: Option<Duration>,
maxsize: Option<i64>,
) -> impl Future<Output = RsmqResult<RsmqQueueAttributes>> + Send;
/// Atomically moves a message from `src` to `dst`, preserving its body and the
/// `:rc` / `:fr` metadata. The message becomes visible in `dst` immediately
/// (score = current time). Returns `true` if the message was found in `src`,
/// `false` if it didn't exist there. Rejects `src == dst`.
///
/// `dst` must be a queue that already exists (created via `create_queue`); this
/// method does not initialize the destination queue's configuration, only inserts
/// the message and bumps `totalsent`.
fn move_message(
&mut self,
src: &str,
msg_id: &str,
dst: &str,
) -> impl Future<Output = RsmqResult<bool>> + Send;
/// Receives the next visible message, but transparently routes any message whose
/// post-increment `rc` exceeds `max_receives` to `dlq` (preserving `:rc` and `:fr`)
/// and tries the next visible message. Returns the first message whose `rc` is
/// still within budget, or `None` if no eligible message was found within the
/// script's per-call iteration cap. Rejects `qname == dlq`.
///
/// `max_receives = 0` ⇒ never deliver, always route to DLQ.
/// `max_receives = N` ⇒ deliver at most N times before routing to DLQ.
fn receive_message_or_dlq<E: TryFrom<RedisBytes, Error = Vec<u8>>>(
&mut self,
qname: &str,
hidden: Option<Duration>,
dlq: &str,
max_receives: u64,
) -> impl Future<Output = RsmqResult<Option<RsmqMessage<E>>>> + Send;
}
#[cfg(feature = "sync")]
pub trait RsmqConnectionSync {
/// Change the hidden time of a already sent message.
///
/// # Arguments
/// * `qname` - Name of the queue
/// * `message_id` - ID of the message to modify
/// * `hidden` - New hidden duration. Has a max time of 9_999_999 for compatibility reasons with the JS version
fn change_message_visibility(
&mut self,
qname: &str,
message_id: &str,
hidden: Duration,
) -> RsmqResult<()>;
/// Creates a new queue. Attributes can be later modified with "set_queue_attributes" method
///
/// # Arguments
/// * `qname` - Name of the queue to create
/// * `hidden` - Time the messages will be hidden when received. Max 9_999_999
/// * `delay` - Time messages will be delayed before delivery
/// * `maxsize` - Maximum message size in bytes (1024-65536 or -1 for unlimited)
fn create_queue(
&mut self,
qname: &str,
hidden: Option<Duration>,
delay: Option<Duration>,
maxsize: Option<i64>,
) -> RsmqResult<()>;
/// Deletes a message from the queue.
///
/// Important to use when you are using receive_message.
///
/// # Arguments
/// * `qname` - Name of the queue
/// * `id` - ID of the message to delete
fn delete_message(&mut self, qname: &str, id: &str) -> RsmqResult<bool>;
/// Deletes the queue and all messages in it
///
/// # Arguments
/// * `qname` - Name of the queue to delete
fn delete_queue(&mut self, qname: &str) -> RsmqResult<()>;
/// Returns the queue attributes and statistics
///
/// # Arguments
/// * `qname` - Name of the queue
fn get_queue_attributes(&mut self, qname: &str) -> RsmqResult<RsmqQueueAttributes>;
/// Returns a list of queues in the namespace
fn list_queues(&mut self) -> RsmqResult<Vec<String>>;
/// Deletes and returns a message. Be aware that using this you may end with deleted & unprocessed messages.
///
/// # Arguments
/// * `qname` - Name of the queue
fn pop_message<E: TryFrom<RedisBytes, Error = Vec<u8>>>(
&mut self,
qname: &str,
) -> RsmqResult<Option<RsmqMessage<E>>>;
/// Returns a message. The message stays hidden for some time (defined by "hidden" argument or the queue
/// settings). After that time, the message will be redelivered. To avoid redelivery, use "delete_message"
/// after this function.
///
/// # Arguments
/// * `qname` - Name of the queue
/// * `hidden` - Optional custom hidden duration. Max 9_999_999
fn receive_message<E: TryFrom<RedisBytes, Error = Vec<u8>>>(
&mut self,
qname: &str,
hidden: Option<Duration>,
) -> RsmqResult<Option<RsmqMessage<E>>>;
/// Sends a message to the queue. The message will be delayed some time (controlled by the "delayed" argument or
/// the queue settings) before being delivered to a client.
///
/// # Arguments
/// * `qname` - Name of the queue
/// * `message` - Message content to send
/// * `delay` - Optional custom delay duration
fn send_message<E: Into<RedisBytes> + Send>(
&mut self,
qname: &str,
message: E,
delay: Option<Duration>,
) -> RsmqResult<String>;
/// Sends a batch of messages. Returns assigned IDs in input order. The default impl loops
/// `send_message`; specialized implementations atomically insert via a single Lua script.
fn send_message_batch<E: Into<RedisBytes> + Send>(
&mut self,
qname: &str,
messages: Vec<E>,
delay: Option<Duration>,
) -> RsmqResult<Vec<String>> {
if messages.is_empty() {
return Ok(Vec::new());
}
let mut ids = Vec::with_capacity(messages.len());
for m in messages {
ids.push(self.send_message(qname, m, delay)?);
}
Ok(ids)
}
/// Receives up to `max_count` messages. Returns at most `max_count` but may return fewer.
/// The default impl loops `receive_message`; specialized implementations use a single
/// Lua script.
fn receive_message_batch<E: TryFrom<RedisBytes, Error = Vec<u8>>>(
&mut self,
qname: &str,
hidden: Option<Duration>,
max_count: u32,
) -> RsmqResult<Vec<RsmqMessage<E>>> {
if max_count == 0 {
return Ok(Vec::new());
}
let mut out = Vec::with_capacity(max_count as usize);
for _ in 0..max_count {
match self.receive_message(qname, hidden)? {
Some(m) => out.push(m),
None => break,
}
}
Ok(out)
}
/// Modify the queue attributes. Note that "hidden" and "delay" can be overwritten when sending messages.
/// "hidden" can be changed by the method "change_message_visibility"
///
/// # Arguments
/// * `qname` - Name of the queue
/// * `hidden` - Time messages will be hidden when received. Max 9_999_999
/// * `delay` - Time messages will be delayed before delivery
/// * `maxsize` - Maximum message size in bytes (1024-65536 or -1 for unlimited)
fn set_queue_attributes(
&mut self,
qname: &str,
hidden: Option<Duration>,
delay: Option<Duration>,
maxsize: Option<i64>,
) -> RsmqResult<RsmqQueueAttributes>;
/// Atomically moves a message from `src` to `dst`. See the async counterpart
/// [`RsmqConnection::move_message`] for the full contract.
fn move_message(&mut self, src: &str, msg_id: &str, dst: &str) -> RsmqResult<bool>;
/// Atomically receives a message, routing over-budget messages to `dlq`.
/// See the async counterpart [`RsmqConnection::receive_message_or_dlq`].
fn receive_message_or_dlq<E: TryFrom<RedisBytes, Error = Vec<u8>>>(
&mut self,
qname: &str,
hidden: Option<Duration>,
dlq: &str,
max_receives: u64,
) -> RsmqResult<Option<RsmqMessage<E>>>;
}