a3s-event 0.3.0

Pluggable event subscription, dispatch, and persistence for the A3S ecosystem
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
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

//! NATS JetStream subscription — implements the `Subscription` trait

use crate::error::{EventError, Result};
use crate::provider::{PendingEvent, Subscription};
use crate::types::{Event, ReceivedEvent};
use async_trait::async_trait;

/// NATS JetStream subscription backed by a pull consumer
pub struct NatsSubscription {
    messages: async_nats::jetstream::consumer::pull::Stream,
    stream_name: String,
}

impl NatsSubscription {
    /// Create a new subscription from a JetStream pull consumer message stream
    pub(crate) fn new(
        messages: async_nats::jetstream::consumer::pull::Stream,
        stream_name: String,
    ) -> Self {
        Self {
            messages,
            stream_name,
        }
    }
}

#[async_trait]
impl Subscription for NatsSubscription {
    async fn next(&mut self) -> Result<Option<ReceivedEvent>> {
        use futures_util::StreamExt;

        match self.messages.next().await {
            Some(Ok(msg)) => {
                let event: Event = serde_json::from_slice(&msg.payload)?;
                let info = msg.info().map_err(|e| {
                    EventError::JetStream(format!("Failed to get message info: {}", e))
                })?;

                let received = ReceivedEvent {
                    event,
                    sequence: info.stream_sequence,
                    num_delivered: info.delivered as u64,
                    stream: self.stream_name.clone(),
                };

                // Auto-ack on successful deserialization
                msg.ack().await.map_err(|e| {
                    EventError::Ack(format!(
                        "Failed to ack message seq {}: {}",
                        info.stream_sequence, e
                    ))
                })?;

                Ok(Some(received))
            }
            Some(Err(e)) => Err(EventError::JetStream(format!(
                "Error receiving message: {}",
                e
            ))),
            None => Ok(None),
        }
    }

    async fn next_manual_ack(&mut self) -> Result<Option<PendingEvent>> {
        use futures_util::StreamExt;

        match self.messages.next().await {
            Some(Ok(msg)) => {
                let event: Event = serde_json::from_slice(&msg.payload)?;
                let info = msg.info().map_err(|e| {
                    EventError::JetStream(format!("Failed to get message info: {}", e))
                })?;

                let received = ReceivedEvent {
                    event,
                    sequence: info.stream_sequence,
                    num_delivered: info.delivered as u64,
                    stream: self.stream_name.clone(),
                };

                let ack_msg = msg.clone();
                let nak_msg = msg;

                Ok(Some(PendingEvent::new(
                    received,
                    move || {
                        Box::pin(async move {
                            ack_msg.ack().await.map_err(|e| {
                                EventError::Ack(format!("Failed to ack: {}", e))
                            })
                        })
                    },
                    move || {
                        Box::pin(async move {
                            nak_msg
                                .ack_with(async_nats::jetstream::AckKind::Nak(None))
                                .await
                                .map_err(|e| {
                                    EventError::Ack(format!("Failed to nak: {}", e))
                                })
                        })
                    },
                )))
            }
            Some(Err(e)) => Err(EventError::JetStream(format!(
                "Error receiving message: {}",
                e
            ))),
            None => Ok(None),
        }
    }
}