RUST-1395 / RUST-1402 GridFS sync API and documentation (#780)

Author: isabelatkinson

src/db/mod.rs

@@ -584,7 +584,7 @@ impl Database {
             .await
     }
 
-    /// Creates a new GridFsBucket in the database with the given options.
+    /// Creates a new [`GridFsBucket`] in the database with the given options.
     pub fn gridfs_bucket(&self, options: impl Into<Option<GridFsBucketOptions>>) -> GridFsBucket {
         GridFsBucket::new(self.clone(), options.into().unwrap_or_default())
     }

src/error.rs

@@ -815,7 +815,8 @@ pub enum GridFsErrorKind {
     #[non_exhaustive]
     MissingChunk { n: u32 },
 
-    /// An operation was attempted on a `GridFsUploadStream` that has already been shut down.
+    /// An operation was attempted on a [`GridFsUploadStream`](crate::gridfs::GridFsUploadStream)
+    /// that has already been shut down.
     UploadStreamClosed,
 
     /// The chunk was the incorrect size.
@@ -843,7 +844,8 @@ pub enum GridFsErrorKind {
         delete_error: Error,
     },
 
-    /// A close operation was attempted on a [`GridFsUploadStream`] while a write was still in
+    /// A close operation was attempted on a
+    /// [`GridFsUploadStream`](crate::gridfs::GridFsUploadStream) while a write was still in
     /// progress.
     WriteInProgress,
 }

src/gridfs.rs

@@ -1,8 +1,7 @@
-// TODO(RUST-1395) Remove these allows.
-#![allow(dead_code, unused_variables, missing_docs)]
+//! Contains the functionality for GridFS operations.
 
 mod download;
-pub mod options;
+pub(crate) mod options;
 mod upload;
 
 use std::sync::{atomic::AtomicBool, Arc};
@@ -20,13 +19,13 @@ use crate::{
 };
 
 pub use download::GridFsDownloadStream;
-pub use options::*;
+pub(crate) use options::*;
 pub use upload::GridFsUploadStream;
 
-pub const DEFAULT_BUCKET_NAME: &str = "fs";
-pub const DEFAULT_CHUNK_SIZE_BYTES: u32 = 255 * 1024;
+const DEFAULT_BUCKET_NAME: &str = "fs";
+const DEFAULT_CHUNK_SIZE_BYTES: u32 = 255 * 1024;
 
-// Contained in a "chunks" collection for each user file
+/// A model for the documents stored in the chunks collection.
 #[derive(Debug, Deserialize, Serialize)]
 pub(crate) struct Chunk<'a> {
     #[serde(rename = "_id")]
@@ -37,45 +36,57 @@ pub(crate) struct Chunk<'a> {
     data: RawBinaryRef<'a>,
 }
 
-/// A collection in which information about stored files is stored. There will be one files
-/// collection document per stored file.
+/// A model for the documents stored in a GridFS bucket's files
+/// collection.
 #[derive(Clone, Debug, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 #[skip_serializing_none]
 #[non_exhaustive]
 pub struct FilesCollectionDocument {
+    /// The file's unique identifier.
     #[serde(rename = "_id")]
     pub id: Bson,
+
+    /// The length of the file in bytes.
     pub length: u64,
-    pub chunk_size: u32,
+
+    /// The size of the file's chunks in bytes.
+    #[serde(rename = "chunkSize")]
+    pub chunk_size_bytes: u32,
+
+    /// The time at which the file was uploaded.
     pub upload_date: DateTime,
+
+    /// The name of the file.
     pub filename: Option<String>,
+
+    /// User-provided metadata associated with the file.
     pub metadata: Option<Document>,
 }
 
 impl FilesCollectionDocument {
     /// Returns the total number of chunks expected to be in the file.
     fn n(&self) -> u32 {
-        Self::n_from_vals(self.length, self.chunk_size)
+        Self::n_from_vals(self.length, self.chunk_size_bytes)
     }
 
-    fn n_from_vals(length: u64, chunk_size: u32) -> u32 {
-        let chunk_size = chunk_size as u64;
-        let n = length / chunk_size + u64::from(length % chunk_size != 0);
+    fn n_from_vals(length: u64, chunk_size_bytes: u32) -> u32 {
+        let chunk_size_bytes = chunk_size_bytes as u64;
+        let n = length / chunk_size_bytes + u64::from(length % chunk_size_bytes != 0);
         n as u32
     }
 
     /// Returns the expected length of a chunk given its index.
     fn expected_chunk_length(&self, n: u32) -> u32 {
-        Self::expected_chunk_length_from_vals(self.length, self.chunk_size, n)
+        Self::expected_chunk_length_from_vals(self.length, self.chunk_size_bytes, n)
     }
 
-    fn expected_chunk_length_from_vals(length: u64, chunk_size: u32, n: u32) -> u32 {
-        let remainder = length % (chunk_size as u64);
-        if n == Self::n_from_vals(length, chunk_size) - 1 && remainder != 0 {
+    fn expected_chunk_length_from_vals(length: u64, chunk_size_bytes: u32, n: u32) -> u32 {
+        let remainder = length % (chunk_size_bytes as u64);
+        if n == Self::n_from_vals(length, chunk_size_bytes) - 1 && remainder != 0 {
             remainder as u32
         } else {
-            chunk_size
+            chunk_size_bytes
         }
     }
 }
@@ -89,7 +100,15 @@ struct GridFsBucketInner {
     created_indexes: AtomicBool,
 }
 
-/// Struct for storing GridFS managed files within a [`Database`].
+/// A `GridFsBucket` provides the functionality for storing and retrieving binary BSON data that
+/// exceeds the 16 MiB size limit of a MongoDB document. Users may upload and download large amounts
+/// of data, called files, to the bucket. When a file is uploaded, its contents are divided into
+/// chunks and stored in a chunks collection. A corresponding [`FilesCollectionDocument`] is also
+/// stored in a files collection. When a user downloads a file, the bucket finds and returns the
+/// data stored in its chunks.
+///
+/// `GridFsBucket` uses [`std::sync::Arc`] internally, so it can be shared safely across threads or
+/// async tasks.
 #[derive(Debug, Clone)]
 pub struct GridFsBucket {
     inner: Arc<GridFsBucketInner>,
@@ -142,41 +161,42 @@ impl GridFsBucket {
         self.inner.files.client()
     }
 
-    /// Gets the read concern of the [`GridFsBucket`].
+    /// Gets the read concern of the bucket.
     pub fn read_concern(&self) -> Option<&ReadConcern> {
         self.inner.options.read_concern.as_ref()
     }
 
-    /// Gets the write concern of the [`GridFsBucket`].
+    /// Gets the write concern of the bucket.
     pub fn write_concern(&self) -> Option<&WriteConcern> {
         self.inner.options.write_concern.as_ref()
     }
 
-    /// Gets the selection criteria of the [`GridFsBucket`].
+    /// Gets the selection criteria of the bucket.
     pub fn selection_criteria(&self) -> Option<&SelectionCriteria> {
         self.inner.options.selection_criteria.as_ref()
     }
 
-    /// Gets the chunk size in bytes for the [`GridFsBucket`].
+    /// Gets the chunk size in bytes for the bucket.
     fn chunk_size_bytes(&self) -> u32 {
         self.inner
             .options
             .chunk_size_bytes
             .unwrap_or(DEFAULT_CHUNK_SIZE_BYTES)
     }
 
-    /// Gets a handle to the files collection for the [`GridFsBucket`].
+    /// Gets a handle to the files collection for the bucket.
     pub(crate) fn files(&self) -> &Collection<FilesCollectionDocument> {
         &self.inner.files
     }
 
-    /// Gets a handle to the chunks collection for the [`GridFsBucket`].
+    /// Gets a handle to the chunks collection for the bucket.
     pub(crate) fn chunks(&self) -> &Collection<Chunk<'static>> {
         &self.inner.chunks
     }
 
-    /// Deletes the [`FilesCollectionDocument`] with the given `id `and its associated chunks from
-    /// this bucket.
+    /// Deletes the [`FilesCollectionDocument`] with the given `id` and its associated chunks from
+    /// this bucket. This method returns an error if the `id` does not match any files in the
+    /// bucket.
     pub async fn delete(&self, id: Bson) -> Result<()> {
         let delete_result = self
             .files()
@@ -209,7 +229,8 @@ impl GridFsBucket {
         self.files().find(filter, find_options).await
     }
 
-    /// Renames the file with the given 'id' to the provided `new_filename`.
+    /// Renames the file with the given 'id' to the provided `new_filename`. This method returns an
+    /// error if the `id` does not match any files in the bucket.
     pub async fn rename(&self, id: Bson, new_filename: impl AsRef<str>) -> Result<()> {
         self.files()
             .update_one(
@@ -222,7 +243,7 @@ impl GridFsBucket {
         Ok(())
     }
 
-    /// Drops all of the files and their associated chunks in this bucket.
+    /// Removes all of the files and their associated chunks from this bucket.
     pub async fn drop(&self) -> Result<()> {
         self.files().drop(None).await?;
         self.chunks().drop(None).await?;

src/gridfs/download.rs

@@ -74,8 +74,29 @@ impl GridFsBucket {
 
 // User functions for downloading to writers.
 impl GridFsBucket {
-    /// Downloads the contents of the stored file specified by `id` and writes
-    /// the contents to the `destination`.
+    /// Downloads the contents of the stored file specified by `id` and writes the contents to the
+    /// `destination`, which may be any type that implements the [`futures_io::AsyncWrite`] trait.
+    ///
+    /// To download to a type that implements [`tokio::io::AsyncWrite`], use the
+    /// [`tokio_util::compat`] module to convert between types.
+    ///
+    /// ```rust
+    /// # use mongodb::{bson::Bson, error::Result, gridfs::GridFsBucket};
+    /// # async fn compat_example(
+    /// #     bucket: GridFsBucket,
+    /// #     tokio_writer: impl tokio::io::AsyncWrite + Unpin,
+    /// #     id: Bson,
+    /// # ) -> Result<()> {
+    /// use tokio_util::compat::TokioAsyncWriteCompatExt;
+    ///
+    /// let futures_writer = tokio_writer.compat_write();
+    /// bucket.download_to_futures_0_3_writer(id, futures_writer).await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
+    /// Note that once an `AsyncWrite` trait is stabilized in the standard library, this method will
+    /// be deprecated in favor of one that accepts a `std::io::AsyncWrite` source.
     pub async fn download_to_futures_0_3_writer<T>(&self, id: Bson, destination: T) -> Result<()>
     where
         T: AsyncWrite + Unpin,
@@ -85,9 +106,34 @@ impl GridFsBucket {
     }
 
     /// Downloads the contents of the stored file specified by `filename` and writes the contents to
-    /// the `destination`. If there are multiple files with the same filename, the `revision` in the
-    /// options provided is used to determine which one to download. If no `revision` is specified,
-    /// the most recent file with the given filename is chosen.
+    /// the `destination`, which may be any type that implements the [`futures_io::AsyncWrite`]
+    /// trait.
+    ///
+    /// If there are multiple files in the bucket with the given filename, the `revision` in the
+    /// options provided is used to determine which one to download. See the documentation for
+    /// [`GridFsDownloadByNameOptions`] for details on how to specify a revision. If no revision is
+    /// provided, the file with `filename` most recently uploaded will be downloaded.
+    ///
+    /// To download to a type that implements [`tokio::io::AsyncWrite`], use the
+    /// [`tokio_util::compat`] module to convert between types.
+    ///
+    /// ```rust
+    /// # use mongodb::{bson::Bson, error::Result, gridfs::GridFsBucket};
+    /// # async fn compat_example(
+    /// #     bucket: GridFsBucket,
+    /// #     tokio_writer: impl tokio::io::AsyncWrite + Unpin,
+    /// #     id: Bson,
+    /// # ) -> Result<()> {
+    /// use tokio_util::compat::TokioAsyncWriteCompatExt;
+    ///
+    /// let futures_writer = tokio_writer.compat_write();
+    /// bucket.download_to_futures_0_3_writer_by_name("example_file", futures_writer, None).await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
+    /// Note that once an `AsyncWrite` trait is stabilized in the standard library, this method will
+    /// be deprecated in favor of one that accepts a `std::io::AsyncWrite` source.
     pub async fn download_to_futures_0_3_writer_by_name<T>(
         &self,
         filename: impl AsRef<str>,
@@ -154,6 +200,38 @@ impl GridFsBucket {
     }
 }
 
+/// A stream from which a file stored in a GridFS bucket can be downloaded.
+///
+/// # Downloading from the Stream
+/// The `GridFsDownloadStream` type implements [`futures_io::AsyncRead`]. It is recommended that
+/// users call the utility methods in [`AsyncReadExt`](futures_util::io::AsyncReadExt) to interact
+/// with the stream.
+///
+/// ```rust
+/// # use mongodb::{bson::Bson, error::Result, gridfs::{GridFsBucket, GridFsDownloadStream}};
+/// # async fn download_example(bucket: GridFsBucket, id: Bson) -> Result<()> {
+/// use futures_util::io::AsyncReadExt;
+///
+/// let mut buf = Vec::new();
+/// let mut download_stream = bucket.open_download_stream(id).await?;
+/// download_stream.read_to_end(&mut buf).await?;
+/// # Ok(())
+/// # }
+/// ```
+///
+/// # Using [`tokio::io::AsyncRead`]
+/// Users who prefer to use tokio's `AsyncRead` trait can use the [`tokio_util::compat`] module.
+///
+/// ```rust
+/// # use mongodb::{bson::Bson, error::Result, gridfs::{GridFsBucket, GridFsUploadStream}};
+/// # async fn compat_example(bucket: GridFsBucket, id: Bson) -> Result<()> {
+/// use tokio_util::compat::FuturesAsyncReadCompatExt;
+///
+/// let futures_upload_stream = bucket.open_download_stream(id).await?;
+/// let tokio_upload_stream = futures_upload_stream.compat();
+/// # Ok(())
+/// # }
+/// ```
 pub struct GridFsDownloadStream {
     state: State,
     current_n: u32,
@@ -206,11 +284,6 @@ impl GridFsDownloadStream {
             file,
         })
     }
-
-    /// Gets the file `id` for the stream.
-    pub fn files_id(&self) -> &Bson {
-        &self.file.id
-    }
 }
 
 impl AsyncRead for GridFsDownloadStream {
@@ -230,7 +303,7 @@ impl AsyncRead for GridFsDownloadStream {
                 } else {
                     let chunks_in_buf = FilesCollectionDocument::n_from_vals(
                         buf.len() as u64,
-                        stream.file.chunk_size,
+                        stream.file.chunk_size_bytes,
                     );
                     // We should read from current_n to chunks_in_buf + current_n, or, if that would
                     // exceed the total number of chunks in the file, to the last chunk in the file.
@@ -244,7 +317,7 @@ impl AsyncRead for GridFsDownloadStream {
                             cursor,
                             buffer,
                             n_range,
-                            stream.file.chunk_size,
+                            stream.file.chunk_size_bytes,
                             stream.file.length,
                         )
                         .boxed(),
@@ -289,7 +362,7 @@ async fn get_bytes(
     mut cursor: Box<Cursor<Chunk<'static>>>,
     mut buffer: Vec<u8>,
     n_range: Range<u32>,
-    chunk_size: u32,
+    chunk_size_bytes: u32,
     file_len: u64,
 ) -> Result<(Vec<u8>, Box<Cursor<Chunk<'static>>>)> {
     for n in n_range {
@@ -305,7 +378,7 @@ async fn get_bytes(
         }
 
         let expected_len =
-            FilesCollectionDocument::expected_chunk_length_from_vals(file_len, chunk_size, n);
+            FilesCollectionDocument::expected_chunk_length_from_vals(file_len, chunk_size_bytes, n);
         if chunk_bytes.len() != (expected_len as usize) {
             return Err(ErrorKind::GridFs(GridFsErrorKind::WrongSizeChunk {
                 actual_size: chunk_bytes.len(),
@@ -330,8 +403,12 @@ impl GridFsBucket {
     }
 
     /// Opens and returns a [`GridFsDownloadStream`] from which the application can read
-    /// the contents of the stored file specified by `filename` and the revision
-    /// in `options`.
+    /// the contents of the stored file specified by `filename`.
+    ///
+    /// If there are multiple files in the bucket with the given filename, the `revision` in the
+    /// options provided is used to determine which one to download. See the documentation for
+    /// [`GridFsDownloadByNameOptions`] for details on how to specify a revision. If no revision is
+    /// provided, the file with `filename` most recently uploaded will be downloaded.
     pub async fn open_download_stream_by_name(
         &self,
         filename: impl AsRef<str>,