RUST-1399 Implement AsyncRead for GridFsDownloadStream (#764)

Author: isabelatkinson

src/cursor/common.rs

@@ -96,7 +96,7 @@ where
         self.state.as_mut().unwrap()
     }
 
-    fn state(&self) -> &CursorState {
+    pub(super) fn state(&self) -> &CursorState {
         self.state.as_ref().unwrap()
     }
 

src/cursor/mod.rs

@@ -141,10 +141,24 @@ impl<T> Cursor<T> {
             .and_then(|c| c.post_batch_resume_token())
     }
 
+    /// Whether this cursor has exhausted all of its getMore calls. The cursor may have more
+    /// items remaining in the buffer.
     pub(crate) fn is_exhausted(&self) -> bool {
         self.wrapped_cursor.as_ref().unwrap().is_exhausted()
     }
 
+    /// Whether this cursor has any additional items to return.
+    pub(crate) fn has_next(&self) -> bool {
+        !self.is_exhausted()
+            || !self
+                .wrapped_cursor
+                .as_ref()
+                .unwrap()
+                .state()
+                .buffer
+                .is_empty()
+    }
+
     pub(crate) fn client(&self) -> &Client {
         &self.client
     }

src/error.rs

@@ -818,7 +818,7 @@ pub enum GridFsErrorKind {
     /// The chunk was the incorrect size.
     #[non_exhaustive]
     WrongSizeChunk {
-        actual_size: u32,
+        actual_size: usize,
         expected_size: u32,
     },
 

src/gridfs.rs

@@ -12,14 +12,13 @@ use std::{
 };
 
 use serde::{Deserialize, Serialize};
-use tokio::io::ReadBuf;
+use serde_with::skip_serializing_none;
 
 use crate::{
     bson::{doc, oid::ObjectId, Bson, DateTime, Document, RawBinaryRef},
-    concern::{ReadConcern, WriteConcern},
     cursor::Cursor,
     error::{ErrorKind, GridFsErrorKind, GridFsFileIdentifier, Result},
-    options::{CollectionOptions, FindOptions, SelectionCriteria},
+    options::{CollectionOptions, FindOptions, ReadConcern, SelectionCriteria, WriteConcern},
     Collection,
     Database,
 };
@@ -42,8 +41,9 @@ struct Chunk<'a> {
 
 /// A collection in which information about stored files is stored. There will be one files
 /// collection document per stored file.
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
+#[skip_serializing_none]
 #[non_exhaustive]
 pub struct FilesCollectionDocument {
     #[serde(rename = "_id")]
@@ -52,12 +52,39 @@ pub struct FilesCollectionDocument {
     pub chunk_size: u32,
     pub upload_date: DateTime,
     pub filename: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
     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)
+    }
+
+    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);
+        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)
+    }
+
+    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 {
+            remainder as u32
+        } else {
+            chunk_size
+        }
+    }
+}
+
 #[derive(Debug)]
 struct GridFsBucketInner {
+    db: Database,
     options: GridFsBucketOptions,
     files: Collection<FilesCollectionDocument>,
     chunks: Collection<Chunk<'static>>,
@@ -134,27 +161,6 @@ impl futures_util::AsyncWrite for GridFsUploadStream {
     }
 }
 
-pub struct GridFsDownloadStream {
-    files_id: Bson,
-}
-
-impl GridFsDownloadStream {
-    /// Gets the file `id` for the stream.
-    pub fn files_id(&self) -> &Bson {
-        &self.files_id
-    }
-}
-
-impl tokio::io::AsyncRead for GridFsDownloadStream {
-    fn poll_read(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &mut ReadBuf<'_>,
-    ) -> Poll<tokio::io::Result<()>> {
-        todo!()
-    }
-}
-
 impl GridFsBucket {
     pub(crate) fn new(db: Database, mut options: GridFsBucketOptions) -> GridFsBucket {
         if options.read_concern.is_none() {
@@ -188,6 +194,7 @@ impl GridFsBucket {
 
         GridFsBucket {
             inner: Arc::new(GridFsBucketInner {
+                db: db.clone(),
                 options,
                 files,
                 chunks,
@@ -230,7 +237,7 @@ impl GridFsBucket {
     }
 
     /// Gets a handle to the chunks collection for the [`GridFsBucket`].
-    fn chunks(&self) -> &Collection<Chunk> {
+    fn chunks(&self) -> &Collection<Chunk<'static>> {
         &self.inner.chunks
     }
 
@@ -260,23 +267,6 @@ impl GridFsBucket {
             .await
     }
 
-    /// Opens and returns a [`GridFsDownloadStream`] from which the application can read
-    /// the contents of the stored file specified by `id`.
-    pub async fn open_download_stream(&self, id: Bson) -> Result<GridFsDownloadStream> {
-        todo!()
-    }
-
-    /// 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`.
-    pub async fn open_download_stream_by_name(
-        &self,
-        filename: String,
-        options: impl Into<Option<GridFsDownloadByNameOptions>>,
-    ) -> Result<GridFsDownloadStream> {
-        todo!()
-    }
-
     /// Deletes the [`FilesCollectionDocument`] with the given `id `and its associated chunks from
     /// this bucket.
     pub async fn delete(&self, id: Bson) -> Result<()> {