RUST-1106 Make the change streams API visible (#571)

Author: abr-egn

src/change_stream/event.rs

@@ -1,14 +1,12 @@
 //! Contains the types related to a `ChangeStream` event.
+#[cfg(test)]
 use std::convert::TryInto;
 
-use crate::{
-    coll::Namespace,
-    cursor::CursorSpecification,
-    error::Result,
-    options::ChangeStreamOptions,
-};
+use crate::{coll::Namespace, cursor::CursorSpecification, options::ChangeStreamOptions};
 
-use bson::{Bson, Document, RawBson, RawDocument, RawDocumentBuf, Timestamp};
+#[cfg(test)]
+use bson::Bson;
+use bson::{Document, RawBson, RawDocumentBuf, Timestamp};
 use serde::{Deserialize, Serialize};
 
 /// An opaque token used for resuming an interrupted

src/change_stream/mod.rs

@@ -5,15 +5,15 @@ pub mod session;
 
 use std::{
     future::Future,
-    marker::PhantomData,
     pin::Pin,
     task::{Context, Poll},
 };
 
 use bson::{Document, Timestamp};
 use derivative::Derivative;
 use futures_core::{future::BoxFuture, Stream};
-use serde::{de::DeserializeOwned, Deserialize};
+use serde::de::DeserializeOwned;
+#[cfg(test)]
 use tokio::sync::oneshot;
 
 use crate::{
@@ -22,15 +22,10 @@ use crate::{
         options::ChangeStreamOptions,
     },
     cursor::{stream_poll_next, BatchValue, CursorStream, NextInBatchFuture},
-    error::{Error, ErrorKind, Result},
+    error::{ErrorKind, Result},
     operation::AggregateTarget,
-    options::AggregateOptions,
-    selection_criteria::{ReadPreference, SelectionCriteria},
-    Client,
     ClientSession,
-    Collection,
     Cursor,
-    Database,
 };
 
 /// A `ChangeStream` streams the ongoing changes of its associated collection, database or
@@ -49,7 +44,7 @@ use crate::{
 ///
 /// A `ChangeStream` can be iterated like any other [`Stream`]:
 ///
-/// ```ignore
+/// ```
 /// # #[cfg(not(feature = "sync"))]
 /// # use futures::stream::StreamExt;
 /// # use mongodb::{Client, error::Result, bson::doc,
@@ -144,18 +139,18 @@ where
     /// 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::{Client, error::Result};
+    /// ```
+    /// # use mongodb::{Client, Collection, bson::Document, error::Result};
     /// # async fn func() -> Result<()> {
     /// # let client = Client::with_uri_str("mongodb://example.com").await?;
-    /// # let coll = client.database("foo").collection("bar");
+    /// # let coll: Collection<Document> = client.database("foo").collection("bar");
     /// let mut change_stream = coll.watch(None, None).await?;
     /// let mut resume_token = None;
     /// while change_stream.is_alive() {
-    ///     if let Some(event) = change_stream.next_if_any() {
+    ///     if let Some(event) = change_stream.next_if_any().await? {
     ///         // process event
     ///     }
-    ///     resume_token = change_stream.resume_token().cloned();
+    ///     resume_token = change_stream.resume_token();
     /// }
     /// #
     /// # Ok(())
@@ -170,8 +165,6 @@ where
 
     #[cfg(test)]
     pub(crate) fn set_kill_watcher(&mut self, tx: oneshot::Sender<()>) {
-        use tokio::sync::oneshot;
-
         self.cursor.set_kill_watcher(tx);
     }
 }
@@ -307,7 +300,7 @@ where
 {
     type Item = Result<T>;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         stream_poll_next(Pin::into_inner(self), cx)
     }
 }

src/change_stream/session.rs

@@ -1,21 +1,11 @@
 //! Types for change streams using sessions.
-use std::{
-    future::Future,
-    marker::PhantomData,
-    pin::Pin,
-    sync::{Arc, Mutex},
-    task::{Context, Poll},
-};
-
-use bson::Document;
 use serde::de::DeserializeOwned;
 
 use crate::{
-    cursor::{BatchValue, CursorStream, NextInBatchFuture},
+    cursor::{BatchValue, NextInBatchFuture},
     error::Result,
     ClientSession,
     SessionCursor,
-    SessionCursorStream,
 };
 
 use super::{
@@ -25,10 +15,10 @@ use super::{
     WatchArgs,
 };
 
-/// A [`SessionChangeStream`] is a change stream that was created with a [`ClientSession`] that must
+/// 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, Client, error::Result};
 /// #
 /// # async fn do_stuff() -> Result<()> {
@@ -79,7 +69,7 @@ where
     /// 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::Client;
     /// # fn main() {
@@ -120,16 +110,16 @@ where
     /// 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::{Client, error::Result};
+    /// ```
+    /// # use mongodb::{Client, Collection, bson::Document, error::Result};
     /// # async fn func() -> Result<()> {
     /// # let client = Client::with_uri_str("mongodb://example.com").await?;
-    /// # let coll = client.database("foo").collection("bar");
+    /// # let coll: Collection<Document> = client.database("foo").collection("bar");
     /// # let mut session = client.start_session(None).await?;
     /// let mut change_stream = coll.watch_with_session(None, None, &mut session).await?;
     /// let mut resume_token = None;
     /// while change_stream.is_alive() {
-    ///     if let Some(event) = change_stream.next_if_any(&mut session) {
+    ///     if let Some(event) = change_stream.next_if_any(&mut session).await? {
     ///         // process event
     ///     }
     ///     resume_token = change_stream.resume_token();

src/client/mod.rs

@@ -282,8 +282,7 @@ impl Client {
     ///
     /// 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) async fn watch(
+    pub async fn watch(
         &self,
         pipeline: impl IntoIterator<Item = Document>,
         options: impl Into<Option<ChangeStreamOptions>>,
@@ -299,8 +298,7 @@ impl Client {
 
     /// 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) async fn watch_with_session(
+    pub async fn watch_with_session(
         &self,
         pipeline: impl IntoIterator<Item = Document>,
         options: impl Into<Option<ChangeStreamOptions>>,

src/coll/mod.rs

@@ -814,8 +814,7 @@ impl<T> Collection<T> {
     ///
     /// 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) async fn watch(
+    pub async fn watch(
         &self,
         pipeline: impl IntoIterator<Item = Document>,
         options: impl Into<Option<ChangeStreamOptions>>,
@@ -833,8 +832,7 @@ impl<T> Collection<T> {
 
     /// Starts a new [`SessionChangeStream`] that receives events for all changes in this collection
     /// using the provided [`ClientSession`].  See [`Client::watch`] for more information.
-    #[allow(unused)]
-    pub(crate) async fn watch_with_session(
+    pub async fn watch_with_session(
         &self,
         pipeline: impl IntoIterator<Item = Document>,
         options: impl Into<Option<ChangeStreamOptions>>,

src/db/mod.rs

@@ -463,8 +463,7 @@ impl Database {
     ///
     /// 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) async fn watch(
+    pub async fn watch(
         &self,
         pipeline: impl IntoIterator<Item = Document>,
         options: impl Into<Option<ChangeStreamOptions>>,
@@ -479,8 +478,7 @@ impl Database {
 
     /// Starts a new [`SessionChangeStream`] that receives events for all changes in this database
     /// using the provided [`ClientSession`].  See [`Database::watch`] for more information.
-    #[allow(unused)]
-    pub(crate) async fn watch_with_session(
+    pub async fn watch_with_session(
         &self,
         pipeline: impl IntoIterator<Item = Document>,
         options: impl Into<Option<ChangeStreamOptions>>,

src/lib.rs

@@ -310,8 +310,7 @@ pub mod options;
 pub use ::bson;
 
 mod bson_util;
-#[allow(unused)]
-pub(crate) mod change_stream;
+pub mod change_stream;
 mod client;
 mod cmap;
 mod coll;

src/options.rs

@@ -16,6 +16,7 @@
 //! ```
 
 pub use crate::{
+    change_stream::options::*,
     client::{auth::*, options::*},
     coll::options::*,
     collation::*,
@@ -26,9 +27,6 @@ pub use crate::{
     selection_criteria::*,
 };
 
-#[allow(unused)]
-pub(crate) use crate::change_stream::options::*;
-
 /// Updates an options struct with the read preference/read concern/write concern of a
 /// client/database/collection.
 macro_rules! resolve_options {

src/sync/change_stream.rs

@@ -22,14 +22,14 @@ use super::ClientSession;
 /// ["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
+/// [`resume_after`](crate::options::ChangeStreamOptions::resume_after) or
+/// [`start_after`](crate::options::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};
 /// #
@@ -94,18 +94,18 @@ where
     /// 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};
+    /// ```
+    /// # use mongodb::{bson::Document, sync::{Client, Collection}, error::Result};
     /// # fn func() -> Result<()> {
     /// # let client = Client::with_uri_str("mongodb://example.com")?;
-    /// # let coll = client.database("foo").collection("bar");
+    /// # let coll: Collection<Document> = 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() {
+    ///     if let Some(event) = change_stream.next_if_any()? {
     ///         // process event
     ///     }
-    ///     resume_token = change_stream.resume_token().cloned();
+    ///     resume_token = change_stream.resume_token();
     /// }
     /// #
     /// # Ok(())
@@ -130,7 +130,7 @@ where
 /// 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<()> {
@@ -181,7 +181,7 @@ where
     /// 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() {
@@ -215,16 +215,16 @@ where
     /// 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};
+    /// ```
+    /// # use mongodb::{bson::Document, sync::{Client, Collection}, error::Result};
     /// # async fn func() -> Result<()> {
     /// # let client = Client::with_uri_str("mongodb://example.com")?;
-    /// # let coll = client.database("foo").collection("bar");
+    /// # let coll: Collection<Document> = 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) {
+    ///     if let Some(event) = change_stream.next_if_any(&mut session)? {
     ///         // process event
     ///     }
     ///     resume_token = change_stream.resume_token();

src/sync/client/mod.rs

@@ -177,8 +177,7 @@ impl Client {
     ///
     /// 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(
+    pub fn watch(
         &self,
         pipeline: impl IntoIterator<Item = Document>,
         options: impl Into<Option<ChangeStreamOptions>>,
@@ -190,8 +189,7 @@ impl Client {
 
     /// 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(
+    pub fn watch_with_session(
         &self,
         pipeline: impl IntoIterator<Item = Document>,
         options: impl Into<Option<ChangeStreamOptions>>,