rsmq_async 18.0.0

Async RSMQ port to rust. RSMQ is a simple redis queue system that works in any redis v2.4+. It contains the same methods as the original one in https://github.com/smrchy/rsmq
Documentation 
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

//! Sending and receiving a custom message type.
//!
//! Implement `Into<RedisBytes>` to send and `TryFrom<RedisBytes, Error = Vec<u8>>`
//! to receive. The wire format here is plain JSON encoded by hand, so the
//! example stays dependency-free; in real code you'd typically use serde_json.
//!
//!     cargo run --example custom_type

use rsmq_async::{RedisBytes, Rsmq, RsmqConnection, RsmqError};

#[derive(Debug)]
struct Job {
    id: u64,
    payload: String,
}

impl From<Job> for RedisBytes {
    fn from(job: Job) -> RedisBytes {
        let payload = job.payload.replace('"', "\\\"");
        format!(r#"{{"id":{},"payload":"{}"}}"#, job.id, payload).into()
    }
}

impl TryFrom<RedisBytes> for Job {
    type Error = Vec<u8>;

    fn try_from(bytes: RedisBytes) -> Result<Self, Self::Error> {
        let raw = bytes.into_bytes();
        let s = std::str::from_utf8(&raw).map_err(|_| raw.clone())?;
        let id_start = s.find(r#""id":"#).ok_or_else(|| raw.clone())? + 5;
        let id_end = s[id_start..].find(',').ok_or_else(|| raw.clone())? + id_start;
        let id: u64 = s[id_start..id_end].parse().map_err(|_| raw.clone())?;
        let payload_start = s.find(r#""payload":""#).ok_or_else(|| raw.clone())? + 11;
        let payload_end = s.rfind('"').ok_or_else(|| raw.clone())?;
        Ok(Job {
            id,
            payload: s[payload_start..payload_end].replace("\\\"", "\""),
        })
    }
}

#[tokio::main]
async fn main() -> Result<(), RsmqError> {
    let mut rsmq = Rsmq::new(Default::default()).await?;
    let qname = "example_custom_type";

    let _ = rsmq.delete_queue(qname).await;
    rsmq.create_queue(qname, None, None, None).await?;

    rsmq.send_message(
        qname,
        Job {
            id: 42,
            payload: "hello".into(),
        },
        None,
    )
    .await?;

    if let Some(msg) = rsmq.receive_message::<Job>(qname, None).await? {
        println!("received: {:?}", msg.message);
        rsmq.delete_message(qname, &msg.id).await?;
    }

    rsmq.delete_queue(qname).await?;
    Ok(())
}