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
113
114
115
116
117
118
119
120

//! NATS JetStream event provider
//!
//! Implements `EventProvider` using NATS JetStream for persistent,
//! distributed event pub/sub with at-least-once delivery.

mod client;
mod config;
mod subscriber;

pub use client::NatsClient;
pub use config::{NatsConfig, StorageType};
pub use subscriber::NatsSubscription;

use crate::error::Result;
use crate::provider::{EventProvider, ProviderInfo, Subscription};
use crate::types::{Event, PublishOptions, SubscribeOptions};
use async_trait::async_trait;

/// NATS JetStream event provider
///
/// Wraps `NatsClient` and implements the `EventProvider` trait.
pub struct NatsProvider {
    client: NatsClient,
}

impl NatsProvider {
    /// Connect to NATS and initialize the JetStream stream
    pub async fn connect(config: NatsConfig) -> Result<Self> {
        let client = NatsClient::connect(config).await?;
        Ok(Self { client })
    }

    /// Get the underlying NATS client for advanced usage
    pub fn client(&self) -> &NatsClient {
        &self.client
    }
}

#[async_trait]
impl EventProvider for NatsProvider {
    async fn publish(&self, event: &Event) -> Result<u64> {
        self.client.publish(event).await
    }

    async fn subscribe_durable(
        &self,
        consumer_name: &str,
        filter_subject: &str,
    ) -> Result<Box<dyn Subscription>> {
        let sub = self.client.subscribe_durable(consumer_name, filter_subject).await?;
        Ok(Box::new(sub))
    }

    async fn subscribe(
        &self,
        filter_subject: &str,
    ) -> Result<Box<dyn Subscription>> {
        let sub = self.client.subscribe(filter_subject).await?;
        Ok(Box::new(sub))
    }

    async fn history(
        &self,
        filter_subject: Option<&str>,
        limit: usize,
    ) -> Result<Vec<Event>> {
        self.client.history(filter_subject, limit).await
    }

    async fn unsubscribe(&self, consumer_name: &str) -> Result<()> {
        self.client.unsubscribe(consumer_name).await
    }

    async fn info(&self) -> Result<ProviderInfo> {
        let info = self.client.stream_info().await?;
        Ok(ProviderInfo {
            provider: "nats".to_string(),
            messages: info.messages,
            bytes: info.bytes,
            consumers: info.consumer_count,
        })
    }

    fn subject_prefix(&self) -> &str {
        &self.client.config().subject_prefix
    }

    fn name(&self) -> &str {
        "nats"
    }

    async fn publish_with_options(&self, event: &Event, opts: &PublishOptions) -> Result<u64> {
        self.client.publish_with_options(event, opts).await
    }

    async fn subscribe_durable_with_options(
        &self,
        consumer_name: &str,
        filter_subject: &str,
        opts: &SubscribeOptions,
    ) -> Result<Box<dyn Subscription>> {
        let sub = self
            .client
            .subscribe_durable_with_options(consumer_name, filter_subject, opts)
            .await?;
        Ok(Box::new(sub))
    }

    async fn subscribe_with_options(
        &self,
        filter_subject: &str,
        opts: &SubscribeOptions,
    ) -> Result<Box<dyn Subscription>> {
        let sub = self
            .client
            .subscribe_with_options(filter_subject, opts)
            .await?;
        Ok(Box::new(sub))
    }
}