RUST-1104 sync wrapper for the change stream API (#566)

Author: abr-egn

src/change_stream/mod.rs

@@ -34,8 +34,8 @@ use crate::{
 };
 
 /// A `ChangeStream` streams the ongoing changes of its associated collection, database or
-/// deployment. `ChangeStream` instances should be created with method `watch` or
-/// `watch_with_pipeline` against the relevant target.
+/// deployment. `ChangeStream` instances should be created with method `watch` against the relevant
+/// target.
 ///
 /// `ChangeStream`s are "resumable", meaning that they can be restarted at a given place in the
 /// stream of events. This is done automatically when the `ChangeStream` encounters certain

src/change_stream/session.rs

@@ -26,28 +26,20 @@ use super::{
 };
 
 /// A [`SessionChangeStream`] is a change stream that was created with a [`ClientSession`] that must
-/// be iterated using one. To iterate, use [`SessionChangeStream::next`] or retrieve a
-/// [`SessionCursorStream`] using [`SessionChangeStream::stream`]:
+/// be iterated using one. To iterate, use [`SessionChangeStream::next`]:
 ///
 /// ```ignore
-/// # use mongodb::{bson::Document, Client, error::Result, ClientSession, SessionCursor};
+/// # use mongodb::{bson::Document, Client, error::Result};
 /// #
 /// # async fn do_stuff() -> Result<()> {
 /// # let client = Client::with_uri_str("mongodb://example.com").await?;
 /// # let mut session = client.start_session(None).await?;
 /// # let coll = client.database("foo").collection::<Document>("bar");
 /// #
-/// // iterate using next()
 /// let mut cs = coll.watch_with_session(None, None, &mut session).await?;
-/// while let Some(event) = cs.next(&mut session).await.transpose()? {
+/// while let Some(event) = cs.next(&mut session).await? {
 ///     println!("{:?}", event)
 /// }
-///
-/// // iterate using `Stream`:
-/// use futures::stream::TryStreamExt;
-///
-/// let mut cs = coll.watch_with_session(None, None, &mut session).await?;
-/// let results: Vec<_> = cs.values(&mut session).try_collect().await?;
 /// #
 /// # Ok(())
 /// # }

src/sync/change_stream.rs

@@ -0,0 +1,242 @@
+use futures_util::stream::StreamExt;
+use serde::de::DeserializeOwned;
+
+use crate::{
+    change_stream::{
+        event::ResumeToken,
+        session::SessionChangeStream as AsyncSessionChangeStream,
+        ChangeStream as AsyncChangeStream,
+    },
+    error::Result,
+    RUNTIME,
+};
+
+use super::ClientSession;
+
+/// A `ChangeStream` streams the ongoing changes of its associated collection, database or
+/// deployment. `ChangeStream` instances should be created with method `watch` against the relevant
+/// target.
+///
+/// `ChangeStream`s are "resumable", meaning that they can be restarted at a given place in the
+/// stream of events. This is done automatically when the `ChangeStream` encounters certain
+/// ["resumable"](https://github.com/mongodb/specifications/blob/master/source/change-streams/change-streams.rst#resumable-error)
+/// errors, such as transient network failures. It can also be done manually by passing
+/// a [`ResumeToken`] retrieved from a past event into either the
+/// [`resume_after`](ChangeStreamOptions::resume_after) or
+/// [`start_after`](ChangeStreamOptions::start_after) (4.2+) options used to create the
+/// `ChangeStream`. Issuing a raw change stream aggregation is discouraged unless users wish to
+/// explicitly opt out of resumability.
+///
+/// A `ChangeStream` can be iterated like any other [`Iterator`]:
+///
+/// ```ignore
+/// # use mongodb::{sync::Client, error::Result, bson::doc,
+/// # change_stream::event::ChangeStreamEvent};
+/// #
+/// # fn func() -> Result<()> {
+/// # let client = Client::with_uri_str("mongodb://example.com")?;
+/// # let coll = client.database("foo").collection("bar");
+/// let mut change_stream = coll.watch(None, None)?;
+/// coll.insert_one(doc! { "x": 1 }, None)?;
+/// for event in change_stream {
+///     let event = event?;
+///     println!("operation performed: {:?}, document: {:?}", event.operation_type, event.full_document);
+///     // operation performed: Insert, document: Some(Document({"x": Int32(1)}))
+/// }
+/// #
+/// # Ok(())
+/// # }
+/// ```
+///
+/// See the documentation [here](https://docs.mongodb.com/manual/changeStreams) for more
+/// details. Also see the documentation on [usage recommendations](https://docs.mongodb.com/manual/administration/change-streams-production-recommendations/).
+pub struct ChangeStream<T>
+where
+    T: DeserializeOwned + Unpin + Send + Sync,
+{
+    async_stream: AsyncChangeStream<T>,
+}
+
+impl<T> ChangeStream<T>
+where
+    T: DeserializeOwned + Unpin + Send + Sync,
+{
+    pub(crate) fn new(async_stream: AsyncChangeStream<T>) -> Self {
+        Self { async_stream }
+    }
+
+    /// Returns the cached resume token that can be used to resume after the most recently returned
+    /// change.
+    ///
+    /// See the documentation
+    /// [here](https://docs.mongodb.com/manual/changeStreams/#change-stream-resume-token) for more
+    /// information on change stream resume tokens.
+    pub fn resume_token(&self) -> Option<ResumeToken> {
+        self.async_stream.resume_token()
+    }
+
+    /// Update the type streamed values will be parsed as.
+    pub fn with_type<D: DeserializeOwned + Unpin + Send + Sync>(self) -> ChangeStream<D> {
+        ChangeStream {
+            async_stream: self.async_stream.with_type(),
+        }
+    }
+
+    /// Returns whether the change stream will continue to receive events.
+    pub fn is_alive(&self) -> bool {
+        self.async_stream.is_alive()
+    }
+
+    /// Retrieves the next result from the change stream, if any.
+    ///
+    /// Where calling `Iterator::next` will internally loop until a change document is received,
+    /// this will make at most one request and return `None` if the returned document batch is
+    /// empty.  This method should be used when storing the resume token in order to ensure the
+    /// most up to date token is received, e.g.
+    ///
+    /// ```ignore
+    /// # use mongodb::{sync::Client, error::Result};
+    /// # fn func() -> Result<()> {
+    /// # let client = Client::with_uri_str("mongodb://example.com")?;
+    /// # let coll = client.database("foo").collection("bar");
+    /// let mut change_stream = coll.watch(None, None)?;
+    /// let mut resume_token = None;
+    /// while change_stream.is_alive() {
+    ///     if let Some(event) = change_stream.next_if_any() {
+    ///         // process event
+    ///     }
+    ///     resume_token = change_stream.resume_token().cloned();
+    /// }
+    /// #
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn next_if_any(&mut self) -> Result<Option<T>> {
+        RUNTIME.block_on(self.async_stream.next_if_any())
+    }
+}
+
+impl<T> Iterator for ChangeStream<T>
+where
+    T: DeserializeOwned + Unpin + Send + Sync,
+{
+    type Item = Result<T>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        RUNTIME.block_on(self.async_stream.next())
+    }
+}
+
+/// A [`SessionChangeStream`] is a change stream that was created with a [`ClientSession`] that must
+/// be iterated using one. To iterate, use [`SessionChangeStream::next`]:
+///
+/// ```ignore
+/// # use mongodb::{bson::Document, sync::Client, error::Result};
+/// #
+/// # async fn do_stuff() -> Result<()> {
+/// # let client = Client::with_uri_str("mongodb://example.com")?;
+/// # let mut session = client.start_session(None)?;
+/// # let coll = client.database("foo").collection::<Document>("bar");
+/// #
+/// let mut cs = coll.watch_with_session(None, None, &mut session)?;
+/// while let Some(event) = cs.next(&mut session)? {
+///     println!("{:?}", event)
+/// }
+/// #
+/// # Ok(())
+/// # }
+/// ```
+pub struct SessionChangeStream<T>
+where
+    T: DeserializeOwned + Unpin,
+{
+    async_stream: AsyncSessionChangeStream<T>,
+}
+
+impl<T> SessionChangeStream<T>
+where
+    T: DeserializeOwned + Unpin + Send + Sync,
+{
+    pub(crate) fn new(async_stream: AsyncSessionChangeStream<T>) -> Self {
+        Self { async_stream }
+    }
+
+    /// Returns the cached resume token that can be used to resume after the most recently returned
+    /// change.
+    ///
+    /// See the documentation
+    /// [here](https://docs.mongodb.com/manual/changeStreams/#change-stream-resume-token) for more
+    /// information on change stream resume tokens.
+    pub fn resume_token(&self) -> Option<ResumeToken> {
+        self.async_stream.resume_token()
+    }
+
+    /// Update the type streamed values will be parsed as.
+    pub fn with_type<D: DeserializeOwned + Unpin + Send + Sync>(self) -> SessionChangeStream<D> {
+        SessionChangeStream {
+            async_stream: self.async_stream.with_type(),
+        }
+    }
+
+    /// Retrieve the next result from the change stream.
+    /// The session provided must be the same session used to create the change stream.
+    ///
+    /// ```ignore
+    /// # use bson::{doc, Document};
+    /// # use mongodb::sync::Client;
+    /// # fn main() {
+    /// # async {
+    /// # let client = Client::with_uri_str("foo")?;
+    /// # let coll = client.database("foo").collection::<Document>("bar");
+    /// # let other_coll = coll.clone();
+    /// # let mut session = client.start_session(None)?;
+    /// let mut cs = coll.watch_with_session(None, None, &mut session)?;
+    /// while let Some(event) = cs.next(&mut session)? {
+    ///     let id = bson::to_bson(&event.id)?;
+    ///     other_coll.insert_one_with_session(doc! { "id": id }, None, &mut session)?;
+    /// }
+    /// # Ok::<(), mongodb::error::Error>(())
+    /// # };
+    /// # }
+    /// ```
+    pub fn next(&mut self, session: &mut ClientSession) -> Result<Option<T>> {
+        RUNTIME.block_on(self.async_stream.next(&mut session.async_client_session))
+    }
+
+    /// Returns whether the change stream will continue to receive events.
+    pub fn is_alive(&self) -> bool {
+        self.async_stream.is_alive()
+    }
+
+    /// Retrieve the next result from the change stream, if any.
+    ///
+    /// Where calling `next` will internally loop until a change document is received,
+    /// this will make at most one request and return `None` if the returned document batch is
+    /// empty.  This method should be used when storing the resume token in order to ensure the
+    /// most up to date token is received, e.g.
+    ///
+    /// ```ignore
+    /// # use mongodb::{sync::Client, error::Result};
+    /// # async fn func() -> Result<()> {
+    /// # let client = Client::with_uri_str("mongodb://example.com")?;
+    /// # let coll = client.database("foo").collection("bar");
+    /// # let mut session = client.start_session(None)?;
+    /// let mut change_stream = coll.watch_with_session(None, None, &mut session)?;
+    /// let mut resume_token = None;
+    /// while change_stream.is_alive() {
+    ///     if let Some(event) = change_stream.next_if_any(&mut session) {
+    ///         // process event
+    ///     }
+    ///     resume_token = change_stream.resume_token();
+    /// }
+    /// #
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn next_if_any(&mut self, session: &mut ClientSession) -> Result<Option<T>> {
+        RUNTIME.block_on(
+            self.async_stream
+                .next_if_any(&mut session.async_client_session),
+        )
+    }
+}

src/sync/client/mod.rs

@@ -1,8 +1,9 @@
 pub mod session;
 
-use super::{ClientSession, Database};
+use super::{ChangeStream, ClientSession, Database, SessionChangeStream};
 use crate::{
     bson::Document,
+    change_stream::{event::ChangeStreamEvent, options::ChangeStreamOptions},
     concern::{ReadConcern, WriteConcern},
     error::Result,
     options::{
@@ -157,4 +158,51 @@ impl Client {
             .block_on(self.async_client.start_session(options))
             .map(Into::into)
     }
+
+    /// Starts a new [`ChangeStream`] that receives events for all changes in the cluster. The
+    /// stream does not observe changes from system collections or the "config", "local" or
+    /// "admin" databases. Note that this method (`watch` on a cluster) is only supported in
+    /// MongoDB 4.0 or greater.
+    ///
+    /// See the documentation [here](https://docs.mongodb.com/manual/changeStreams/) on change
+    /// streams.
+    ///
+    /// Change streams require either a "majority" read concern or no read
+    /// concern. Anything else will cause a server error.
+    ///
+    /// Note that using a `$project` stage to remove any of the `_id` `operationType` or `ns` fields
+    /// will cause an error. The driver requires these fields to support resumability. For
+    /// more information on resumability, see the documentation for
+    /// [`ChangeStream`](change_stream/struct.ChangeStream.html)
+    ///
+    /// If the pipeline alters the structure of the returned events, the parsed type will need to be
+    /// changed via [`ChangeStream::with_type`].
+    #[allow(unused)]
+    pub(crate) fn watch(
+        &self,
+        pipeline: impl IntoIterator<Item = Document>,
+        options: impl Into<Option<ChangeStreamOptions>>,
+    ) -> Result<ChangeStream<ChangeStreamEvent<Document>>> {
+        RUNTIME
+            .block_on(self.async_client.watch(pipeline, options))
+            .map(ChangeStream::new)
+    }
+
+    /// Starts a new [`SessionChangeStream`] that receives events for all changes in the cluster
+    /// using the provided [`ClientSession`].  See [`Client::watch`] for more information.
+    #[allow(unused)]
+    pub(crate) fn watch_with_session(
+        &self,
+        pipeline: impl IntoIterator<Item = Document>,
+        options: impl Into<Option<ChangeStreamOptions>>,
+        session: &mut ClientSession,
+    ) -> Result<SessionChangeStream<ChangeStreamEvent<Document>>> {
+        RUNTIME
+            .block_on(self.async_client.watch_with_session(
+                pipeline,
+                options,
+                &mut session.async_client_session,
+            ))
+            .map(SessionChangeStream::new)
+    }
 }