RUST-1507 Implement AsyncWrite for GridFsUploadStream (#774)

Author: isabelatkinson

src/error.rs

@@ -815,6 +815,9 @@ pub enum GridFsErrorKind {
     #[non_exhaustive]
     MissingChunk { n: u32 },
 
+    /// An operation was attempted on a `GridFsUploadStream` that has already been shut down.
+    UploadStreamClosed,
+
     /// The chunk was the incorrect size.
     #[non_exhaustive]
     WrongSizeChunk {
@@ -828,6 +831,21 @@ pub enum GridFsErrorKind {
         actual_number: u32,
         expected_number: u32,
     },
+
+    /// An error occurred when aborting a file upload.
+    #[non_exhaustive]
+    AbortError {
+        /// The original error. Only present if the abort occurred because of an error during a
+        /// GridFS operation.
+        original_error: Option<Error>,
+
+        /// The error that occurred when attempting to remove any orphaned chunks.
+        delete_error: Error,
+    },
+
+    /// A close operation was attempted on a [`GridFsUploadStream`] while a write was still in
+    /// progress.
+    WriteInProgress,
 }
 
 /// An identifier for a file stored in a GridFS bucket.

src/gridfs.rs

@@ -1,36 +1,34 @@
 // TODO(RUST-1395) Remove these allows.
-#![allow(dead_code, unused_variables)]
+#![allow(dead_code, unused_variables, missing_docs)]
 
 mod download;
 pub mod options;
 mod upload;
 
-use core::task::{Context, Poll};
-use std::{
-    pin::Pin,
-    sync::{atomic::AtomicBool, Arc},
-};
+use std::sync::{atomic::AtomicBool, Arc};
 
 use serde::{Deserialize, Serialize};
 use serde_with::skip_serializing_none;
 
 use crate::{
     bson::{doc, oid::ObjectId, Bson, DateTime, Document, RawBinaryRef},
     cursor::Cursor,
-    error::{ErrorKind, GridFsErrorKind, GridFsFileIdentifier, Result},
+    error::{Error, ErrorKind, GridFsErrorKind, GridFsFileIdentifier, Result},
     options::{CollectionOptions, FindOptions, ReadConcern, SelectionCriteria, WriteConcern},
     Collection,
     Database,
 };
 
-use options::*;
+pub use download::GridFsDownloadStream;
+pub use options::*;
+pub use upload::GridFsUploadStream;
 
 pub const DEFAULT_BUCKET_NAME: &str = "fs";
 pub const DEFAULT_CHUNK_SIZE_BYTES: u32 = 255 * 1024;
 
 // Contained in a "chunks" collection for each user file
 #[derive(Debug, Deserialize, Serialize)]
-struct Chunk<'a> {
+pub(crate) struct Chunk<'a> {
     #[serde(rename = "_id")]
     id: ObjectId,
     files_id: Bson,
@@ -97,70 +95,6 @@ pub struct GridFsBucket {
     inner: Arc<GridFsBucketInner>,
 }
 
-// TODO: RUST-1395 Add documentation and example code for this struct.
-pub struct GridFsUploadStream {
-    files_id: Bson,
-}
-
-impl GridFsUploadStream {
-    /// Gets the file `id` for the stream.
-    pub fn files_id(&self) -> &Bson {
-        &self.files_id
-    }
-
-    /// Consumes the stream and uploads data in the stream to the server.
-    pub async fn finish(self) {
-        todo!()
-    }
-
-    /// Aborts the upload and discards the upload stream.
-    pub async fn abort(self) {
-        todo!()
-    }
-}
-
-impl tokio::io::AsyncWrite for GridFsUploadStream {
-    fn poll_write(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &[u8],
-    ) -> Poll<tokio::io::Result<usize>> {
-        todo!()
-    }
-
-    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<tokio::io::Result<()>> {
-        todo!()
-    }
-
-    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<tokio::io::Result<()>> {
-        todo!()
-    }
-}
-
-impl futures_util::AsyncWrite for GridFsUploadStream {
-    fn poll_write(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &[u8],
-    ) -> Poll<core::result::Result<usize, futures_util::io::Error>> {
-        todo!()
-    }
-
-    fn poll_flush(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<core::result::Result<(), futures_util::io::Error>> {
-        todo!()
-    }
-
-    fn poll_close(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<core::result::Result<(), futures_util::io::Error>> {
-        todo!()
-    }
-}
-
 impl GridFsBucket {
     pub(crate) fn new(db: Database, mut options: GridFsBucketOptions) -> GridFsBucket {
         if options.read_concern.is_none() {
@@ -232,41 +166,15 @@ impl GridFsBucket {
     }
 
     /// Gets a handle to the files collection for the [`GridFsBucket`].
-    fn files(&self) -> &Collection<FilesCollectionDocument> {
+    pub(crate) fn files(&self) -> &Collection<FilesCollectionDocument> {
         &self.inner.files
     }
 
     /// Gets a handle to the chunks collection for the [`GridFsBucket`].
-    fn chunks(&self) -> &Collection<Chunk<'static>> {
+    pub(crate) fn chunks(&self) -> &Collection<Chunk<'static>> {
         &self.inner.chunks
     }
 
-    /// Opens a [`GridFsUploadStream`] that the application can write the contents of the file to.
-    /// The application provides a custom file id.
-    ///
-    /// Returns a [`GridFsUploadStream`] to which the application will write the contents.
-    pub async fn open_upload_stream_with_id(
-        &self,
-        id: Bson,
-        filename: String,
-        options: impl Into<Option<GridFsUploadOptions>>,
-    ) -> Result<GridFsUploadStream> {
-        todo!()
-    }
-
-    /// Opens a [`GridFsUploadStream`] that the application can write the contents of the file to.
-    /// The driver generates a unique [`Bson::ObjectId`] for the file id.
-    ///
-    /// Returns a [`GridFsUploadStream`] to which the application will write the contents.
-    pub async fn open_upload_stream(
-        &self,
-        filename: String,
-        options: impl Into<Option<GridFsUploadOptions>>,
-    ) -> Result<GridFsUploadStream> {
-        self.open_upload_stream_with_id(Bson::ObjectId(ObjectId::new()), filename, options)
-            .await
-    }
-
     /// Deletes the [`FilesCollectionDocument`] with the given `id `and its associated chunks from
     /// this bucket.
     pub async fn delete(&self, id: Bson) -> Result<()> {
@@ -322,3 +230,9 @@ impl GridFsBucket {
         Ok(())
     }
 }
+
+impl Error {
+    fn into_futures_io_error(self) -> futures_io::Error {
+        futures_io::Error::new(futures_io::ErrorKind::Other, self)
+    }
+}

src/gridfs/download.rs

@@ -163,9 +163,8 @@ pub struct GridFsDownloadStream {
 type GetBytesFuture = BoxFuture<'static, Result<(Vec<u8>, Box<Cursor<Chunk<'static>>>)>>;
 
 enum State {
-    // Idle stores these fields as options so that they can be moved into a GetBytesFuture
-    // without requiring ownership of the state. They will always store values and can be
-    // unwrapped safely.
+    // Idle is stored as an option so that its fields can be moved into a GetBytesFuture
+    // without requiring ownership of the state. It can always be unwrapped safely.
     Idle(Option<Idle>),
     Busy(GetBytesFuture),
     Done,
@@ -280,8 +279,7 @@ impl AsyncRead for GridFsDownloadStream {
             }
             Err(error) => {
                 stream.state = State::Done;
-                let error = futures_io::Error::new(futures_io::ErrorKind::Other, error);
-                Poll::Ready(Err(error))
+                Poll::Ready(Err(error.into_futures_io_error()))
             }
         }
     }

src/gridfs/options.rs

@@ -33,7 +33,7 @@ pub struct GridFsBucketOptions {
 /// [`GridFsBucket`].
 #[derive(Clone, Debug, Default, Deserialize, TypedBuilder)]
 #[serde(rename_all = "camelCase")]
-#[builder(field_defaults(setter(into)))]
+#[builder(field_defaults(default, setter(into)))]
 #[non_exhaustive]
 pub struct GridFsUploadOptions {
     /// The number of bytes per chunk of this file. Defaults to the `chunk_size_bytes` specified
@@ -47,7 +47,7 @@ pub struct GridFsUploadOptions {
 /// Contains the options for creating a [`GridFsDownloadStream`] to retrieve a stored file
 /// from a [`GridFsBucket`].
 #[derive(Clone, Debug, Default, Deserialize, TypedBuilder)]
-#[builder(field_defaults(setter(into)))]
+#[builder(field_defaults(default, setter(into)))]
 #[non_exhaustive]
 pub struct GridFsDownloadByNameOptions {
     /// Which revision (documents with the same filename and different `upload_date`)
@@ -65,7 +65,7 @@ pub struct GridFsDownloadByNameOptions {
 
 /// Contains the options for performing a find operation on a files collection.  
 #[derive(Clone, Debug, Default, Deserialize, TypedBuilder)]
-#[builder(field_defaults(setter(into)))]
+#[builder(field_defaults(default, setter(into)))]
 #[non_exhaustive]
 pub struct GridFsFindOptions {
     /// Enables writing to temporary files on the server. When set to true, the