Struct MongoStream 

pub struct MongoStream { /* private fields */ }

The main wrapper around MongoDB change streams.

MongoStream provides an interface for subscribing to MongoDB change events with custom callbacks for different event types per collection.

Implementations

impl MongoStream

pub fn new(db: Database) -> Self

Creates a new MongoStream instance for the given database.

Arguments

• db - The MongoDB database to monitor.

Returns

A new MongoStream instance configured to work with the provided database.

pub fn add_callback<F>( &mut self, collection_name: impl Into<String>, event: Event, callback: F, )

where F: Fn(&Document) -> Pin<Box<dyn Future<Output = ()> + Send>> + 'static + Send + Sync,

Registers a callback for a specific event type on a collection.

Arguments

• collection_name - The name of the MongoDB collection to monitor.
• event - The event type to subscribe to.
• callback - The async function to call when the event occurs.

Example

use MongoStream::Event;

// load the mongodb database correctly
let db = Database{};
let mut mongo_stream = MongoStream::new(db: Database);
mongo_stream.add_callback("users", Event::Insert, |event| {
    Box::pin(async move {
        println!("New user inserted: {:?}", event);
    })
});

pub async fn start_stream( &mut self, collection_name: &str, ) -> Result<(), MongoStreamError>

Starts monitoring a collection for changes.

This method will block and continuously process events from the MongoDB change stream until an error occurs or the stream is closed.

Arguments

• collection_name - The name of the collection to monitor.

Returns

A Result that is Ok if the stream was closed normally, or an error if something went wrong.

Example

async fn monitor() -> Result<(), MongoStreamError> {
    let mongo_stream = MongoStream::new(db);
    // ... register callbacks ...
    mongo_stream.start_stream("users").await
}

Starts monitoring a collection for changes.

This method will spawn a new task to monitor the collection and return immediately. To stop the stream, call close_stream with the same collection name.

Arguments

• collection_name - The name of the collection to monitor.

Returns

A Result that is Ok if the stream was started successfully, or an error if something went wrong.

Example

async fn start_monitoring() -> Result<(), MongoStreamError> {
    let mongo_stream = MongoStream::new(db);
    // ... register callbacks ...
    mongo_stream.start_stream("users").await?;
    // The stream is now running in the background
     
    // Later, when you want to stop it:
    mongo_stream.close_stream("users").await;
    Ok(())
}

pub async fn close_stream(&mut self, collection_name: &str) -> bool

Closes a specific stream for a collection.

This method sends a signal to stop the monitoring task for a specific collection.

Arguments

• collection_name - The name of the collection whose stream should be closed.

Returns

A boolean indicating whether a stream was found and closed. Returns false if there was no active stream for the given collection.

Example

use rs_mongo_stream::MongoStream;

async fn example() {
    let mongo_stream = MongoStream::new(db);
    // ... start streams ...
     
    // When done with a specific collection
    let closed = mongo_stream.close_stream("users").await;
    println!("Stream closed: {}", closed);
}

pub async fn close_all_streams(&mut self) -> usize

Closes all active streams.

This method sends signals to stop all monitoring tasks and clears the active streams list.

Returns

The number of streams that were closed.

Example

async fn shutdown() {
    let mongo_stream = MongoStream::new(db);
    // ... start streams ...
     
    // When shutting down the application
    let closed_count = mongo_stream.close_all_streams().await;
    println!("Closed {} streams", closed_count);
}

Trait Implementations

impl Stream for MongoStream

Implementation of Stream trait for MongoStream to make it compatible with tokio_stream.

Note: This is a placeholder implementation and would need to be expanded in a real-world scenario to connect to actual MongoDB change streams.

type Item = Result<Event, MongoStreamError>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, _cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.