mrcon 0.1.0

MongoDB to RabbitMQ connector library for streaming change events
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

use std::sync::Arc;

use super::error::Error;
use super::publish::Publish;
use crate::config::RabbitMq;
use lapin::{
    options::BasicPublishOptions, publisher_confirm::Confirmation, types::FieldTable,
    BasicProperties, Channel, Connection, ConnectionProperties,
};
use mongodb::{bson::Document, change_stream::event::ChangeStreamEvent};
use serde_json;
use tracing::trace;

/// RabbitMQ publisher for MongoDB change events.
///
/// The `Publisher` encapsulates a RabbitMQ channel and configuration, providing methods to declare queues
/// and publish MongoDB change stream events as JSON messages. It is used by the connector to forward
/// change events to RabbitMQ reliably.
pub struct Publisher {
    pub config: RabbitMq,
    channel: Channel,
    // DD: to hold a connection while the channel is alive
    _connection: Arc<Connection>,
}

impl Publisher {
    /// Create a new `Publisher` for the given RabbitMQ configuration and URI.
    ///
    /// This method establishes a connection to RabbitMQ, creates a channel, and declares the target queue.
    ///
    /// # Arguments
    /// * `config` - RabbitMQ configuration (queue/stream name, etc).
    /// * `rabbitmq_uri` - Connection string for RabbitMQ.
    ///
    /// # Errors
    /// Returns an error if the connection, channel, or queue declaration fails.
    pub async fn new(config: &RabbitMq, rabbitmq_uri: &str) -> Result<Self, Error> {
        let conn = Connection::connect(rabbitmq_uri, ConnectionProperties::default()).await?;
        let channel = conn.create_channel().await?;
        channel
            .queue_declare(
                &config.stream_name,
                Default::default(),
                FieldTable::default(),
            )
            .await?;
        Ok(Self {
            config: config.clone(),
            channel,
            _connection: Arc::new(conn),
        })
    }

    /// Create a new `Publisher` using an existing RabbitMQ connection.
    ///
    /// This is the preferred way to construct a `Publisher` when you plan to create
    /// multiple publishers that share the same RabbitMQ connection (for example,
    /// when you want to avoid opening a new TCP connection for each publisher).
    ///
    /// # Arguments
    /// * `config` - RabbitMQ configuration (queue/stream name, etc).
    /// * `connection` - An existing, shared RabbitMQ `Connection`.
    ///
    /// # Returns
    /// Returns a new `Publisher` instance with its own channel, but sharing the provided connection.
    ///
    /// # Errors
    /// Returns an error if the channel creation or queue declaration fails.
    pub async fn with_connection(
        config: RabbitMq,
        connection: Arc<Connection>,
    ) -> Result<Self, Error> {
        let channel = connection.create_channel().await?;
        channel
            .queue_declare(
                &config.stream_name,
                Default::default(),
                FieldTable::default(),
            )
            .await?;
        Ok(Self {
            config,
            channel,
            _connection: connection,
        })
    }

    /// Publish a MongoDB change event to RabbitMQ as a JSON message.
    ///
    /// # Arguments
    /// * `event` - The MongoDB change stream event to publish.
    ///
    /// # Errors
    /// Returns an error if serialization or publishing fails.
    pub async fn publish(&self, event: &ChangeStreamEvent<Document>) -> Result<(), Error> {
        let payload = serde_json::to_vec(event)?;
        let confirm: Confirmation = self
            .channel
            .basic_publish(
                "",
                &self.config.stream_name,
                BasicPublishOptions::default(),
                &payload,
                BasicProperties::default(),
            )
            .await?
            .await?;
        trace!(queue = %self.config.stream_name, "Published message to RabbitMQ, payload: {}, confirmation: {:?}", serde_json::to_string(event)?, confirm);
        Ok(())
    }
}

#[async_trait::async_trait]
impl Publish for Publisher {
    async fn publish(&self, event: &ChangeStreamEvent<Document>) -> Result<(), Error> {
        self.publish(event).await
    }
}