stream-consumer-task 0.1.0

Start asynchronous task in background to consume stream
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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

use std::time::Duration;

use lapin::Connection;
use stream_consumer_task::StreamConsumerTask;
use tokio::{
    select,
    signal::unix::{signal, SignalKind},
    spawn,
    task::JoinSet,
    time::sleep,
};

// main

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let queue_name = "test";
    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", Default::default()).await?;
    let chan_csm = conn.create_channel().await?;
    chan_csm
        .queue_declare(queue_name, Default::default(), Default::default())
        .await?;
    let csm = chan_csm
        .basic_consume(queue_name, "", Default::default(), Default::default())
        .await?;
    let csm = StreamConsumerTask::start(csm, |delivery, mut stop_rx| async move {
        match delivery {
            Ok(delivery) => {
                let res: anyhow::Result<()> = async {
                    let payload = String::from_utf8(delivery.data.clone())?;
                    select! {
                        _ = sleep(Duration::from_secs(5)) => {
                            println!("{payload} - processed");
                            delivery.ack(Default::default()).await?;
                        }
                        _ = stop_rx.changed() => {
                            println!("{payload} - stopped");
                        }
                    }
                    Ok(())
                }
                .await;
                if let Err(err) = res {
                    eprintln!("failed to handle delivery: {err}");
                }
            }
            Err(err) => {
                eprintln!("failed to handle delivery: {err}");
            }
        }
    });
    let chan_prd = conn.create_channel().await?;
    let mut idx = 0;
    let _task = spawn(async move {
        loop {
            let payload = format!("msg_{idx:0>3}");
            let res = chan_prd
                .basic_publish(
                    "",
                    queue_name,
                    Default::default(),
                    payload.as_bytes(),
                    Default::default(),
                )
                .await;
            match res {
                Ok(_) => println!("{payload} - sent"),
                Err(err) => eprintln!("failed to send payload: {err}"),
            }
            sleep(Duration::from_secs(1)).await;
            idx += 1;
        }
    });
    let mut sig_int = signal(SignalKind::interrupt())?;
    let mut sig_term = signal(SignalKind::terminate())?;
    let mut sigs = JoinSet::new();
    sigs.spawn(async move {
        sig_int.recv().await;
    });
    sigs.spawn(async move {
        sig_term.recv().await;
    });
    sigs.join_next().await;
    csm.stop().await;
    Ok(())
}