feat: add typed cursor

This will convert the Document to the Collection type and return the oid
as a tuple.

Author: fscc-alexkornitzer

mongod/src/async/client.rs

@@ -14,6 +14,7 @@ use mongodb::options::{
 };
 use url::Url;
 
+use super::TypedCursor;
 use crate::collection::Collection;
 use crate::filter::{AsFilter, Filter};
 use crate::query;
@@ -533,7 +534,7 @@ impl Client {
     /// # Errors
     ///
     /// This method fails if the mongodb encountered an error.
-    pub async fn find<C, F>(&self, filter: Option<F>) -> crate::Result<mongodb::Cursor<Document>>
+    pub async fn find<C, F>(&self, filter: Option<F>) -> crate::Result<TypedCursor<C>>
     where
         C: AsFilter<F> + Collection,
         F: Filter,
@@ -553,7 +554,7 @@ impl Client {
     /// # Errors
     ///
     /// This method fails if the mongodb encountered an error, or if the found document is invalid.
-    pub async fn find_one<C, F>(&self, filter: F) -> crate::Result<Option<C>>
+    pub async fn find_one<C, F>(&self, filter: F) -> crate::Result<Option<(ObjectId, C)>>
     where
         C: AsFilter<F> + Collection,
         F: Filter,
@@ -562,9 +563,7 @@ impl Client {
         let find: query::Find<C> = query::Find::new();
         let mut cursor = find.filter(filter)?.query(&self).await?;
         if let Some(res) = cursor.next().await {
-            let doc = res.map_err(crate::error::mongodb)?;
-            let doc = C::from_document(doc)?;
-            return Ok(Some(doc));
+            return Ok(Some(res?));
         }
         Ok(None)
     }

mongod/src/async/cursor.rs

@@ -2,28 +2,37 @@ use std::marker::PhantomData;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-use bson::Document;
+use bson::{oid::ObjectId, Document};
 use futures::Stream;
-use mongodb::Cursor;
 
 use crate::collection::Collection;
 
-/// A typed blocking cursor.
+/// A typed cursor.
 ///
-/// This wraps the blocking `Cursor` so that is can be automatically return typed documents.
+/// This wraps the `Cursor` so that it can be automatically return typed documents.
 pub struct TypedCursor<T>
 where
     T: Collection,
 {
-    cursor: Cursor<Document>,
+    cursor: mongodb::Cursor<Document>,
     document_type: PhantomData<T>,
 }
 
-impl<T> From<Cursor<Document>> for TypedCursor<T>
+impl<T> TypedCursor<T>
 where
     T: Collection,
 {
-    fn from(cursor: Cursor<Document>) -> Self {
+    /// Allow access to the wrapped [`mongodb::Cursor`](https://docs.rs/mongodb/2.0.0/mongodb/struct.Cursor.html).
+    pub fn into_inner(self) -> mongodb::Cursor<Document> {
+        self.cursor
+    }
+}
+
+impl<T> From<mongodb::Cursor<Document>> for TypedCursor<T>
+where
+    T: Collection,
+{
+    fn from(cursor: mongodb::Cursor<Document>) -> Self {
         TypedCursor {
             cursor: cursor,
             document_type: PhantomData,
@@ -35,15 +44,15 @@ impl<T> Stream for TypedCursor<T>
 where
     T: Collection,
 {
-    type Item = crate::Result<T>;
+    type Item = crate::Result<(ObjectId, T)>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let next = Pin::new(&mut self.cursor).poll_next(cx);
         match next {
             Poll::Ready(opt) => Poll::Ready(opt.map(|result| {
-                result
-                    .map_err(crate::error::mongodb)
-                    .and_then(|doc| T::from_document(doc))
+                let doc = result.map_err(crate::error::mongodb)?;
+                let oid = doc.get_object_id("_id").map_err(crate::error::bson)?;
+                Ok((oid, T::from_document(doc)?))
             })),
             Poll::Pending => Poll::Pending,
         }

mongod/src/blocking/client.rs

@@ -10,7 +10,7 @@ use mongodb::options::{
 };
 use mongodb::results::{DeleteResult, InsertManyResult, UpdateResult};
 
-use super::cursor::Cursor;
+use super::cursor::{Cursor, TypedCursor};
 use crate::collection::Collection;
 use crate::filter::{AsFilter, Filter};
 use crate::query;
@@ -255,7 +255,7 @@ impl Client {
     /// # Errors
     ///
     /// This method fails if the mongodb encountered an error.
-    pub fn find<C, F>(&self, filter: Option<F>) -> crate::Result<Cursor>
+    pub fn find<C, F>(&self, filter: Option<F>) -> crate::Result<TypedCursor<C>>
     where
         C: AsFilter<F> + Collection,
         F: Filter,
@@ -275,7 +275,7 @@ impl Client {
     /// # Errors
     ///
     /// This method fails if the mongodb encountered an error, or if the found document is invalid.
-    pub fn find_one<C, F>(&self, filter: F) -> crate::Result<Option<C>>
+    pub fn find_one<C, F>(&self, filter: F) -> crate::Result<Option<(ObjectId, C)>>
     where
         C: AsFilter<F> + Collection,
         F: Filter,
@@ -284,9 +284,7 @@ impl Client {
         let find: query::Find<C> = query::Find::new();
         let mut cursor = find.filter(filter)?.blocking(&self)?;
         if let Some(res) = cursor.next() {
-            let document = res.map_err(crate::error::mongodb)?;
-            let document = C::from_document(document)?;
-            return Ok(Some(document));
+            return Ok(Some(res?));
         }
         Ok(None)
     }

mongod/src/blocking/cursor.rs

@@ -1,6 +1,6 @@
 use std::marker::PhantomData;
 
-use bson::Document;
+use bson::{oid::ObjectId, Document};
 use futures::stream::StreamExt;
 
 use crate::collection::Collection;
@@ -13,7 +13,7 @@ enum Response {
     Next(Option<crate::Result<Document>>),
 }
 
-/// A blocking version of the [`mongodb::Cursor`](https://docs.rs/mongodb/1.1.1/mongodb/struct.Cursor.html).
+/// A blocking version of the [`mongodb::Cursor`](https://docs.rs/mongodb/2.0.0/mongodb/struct.Cursor.html).
 ///
 /// This wraps the async `Cursor` so that is can be called in a synchronous fashion, please see the
 /// asynchronous description for more information about the cursor.
@@ -66,9 +66,18 @@ pub struct TypedCursor<T>
 where
     T: Collection,
 {
+    cursor: Cursor,
     document_type: PhantomData<T>,
+}
 
-    tx: tokio::sync::mpsc::UnboundedSender<(Request, std::sync::mpsc::Sender<Response>)>,
+impl<T> TypedCursor<T>
+where
+    T: Collection,
+{
+    /// Allow access to the wrapped blocking `Cursor`
+    pub fn into_inner(self) -> Cursor {
+        self.cursor
+    }
 }
 
 impl<T> From<Cursor> for TypedCursor<T>
@@ -77,8 +86,8 @@ where
 {
     fn from(cursor: Cursor) -> Self {
         TypedCursor {
+            cursor,
             document_type: PhantomData,
-            tx: cursor.tx,
         }
     }
 }
@@ -87,21 +96,14 @@ impl<T> Iterator for TypedCursor<T>
 where
     T: Collection,
 {
-    type Item = crate::Result<T>;
+    type Item = crate::Result<(ObjectId, T)>;
     fn next(&mut self) -> Option<Self::Item> {
-        let (tx, rx) = std::sync::mpsc::channel();
-        self.tx
-            .send((Request::Next, tx))
-            .expect("core thread panicked");
-        let res = rx
-            .recv()
-            .expect("could not get response from mongo runtime");
-        let Response::Next(c) = res;
-        let resp = match c {
-            Some(Ok(b)) => Some(T::from_document(b)),
-            Some(Err(e)) => Some(Err(crate::error::mongodb(e))),
-            None => None,
-        };
-        resp
+        let next = self.cursor.next();
+
+        next.map(|res| {
+            let doc = res?;
+            let oid = doc.get_object_id("_id").map_err(crate::error::bson)?;
+            Ok((oid, T::from_document(doc)?))
+        })
     }
 }

mongod/src/blocking/mod.rs

@@ -38,4 +38,4 @@ mod cursor;
 
 pub use self::client::{Client, ClientBuilder};
 pub(crate) use self::client::{Request, Response};
-pub use self::cursor::Cursor;
+pub use self::cursor::{Cursor, TypedCursor};

mongod/src/lib.rs

@@ -89,8 +89,8 @@
 //!
 //! let mut cursor = client.find::<User, _>(None).await.unwrap();
 //! while let Some(res) = cursor.next().await {
-//!     if let Ok(user) = res {
-//!         println!("{:?}", user);
+//!     if let Ok((id, user)) = res {
+//!         println!("{} - {:?}", id, user);
 //!     }
 //! }
 //! # Ok(())

mongod/src/query/find.rs

@@ -3,12 +3,11 @@ use std::time::Duration;
 
 use mongodb::bson::Document;
 use mongodb::options::{Collation, CursorType, FindOptions, Hint, ReadConcern, SelectionCriteria};
-use mongodb::Cursor;
 
 use crate::collection::Collection;
 use crate::field::{AsField, Field};
 use crate::filter::{AsFilter, Filter};
-use crate::r#async::Client;
+use crate::r#async::{Client, TypedCursor};
 use crate::sort::Sort;
 
 /// A querier to find documents in a MongoDB collection.
@@ -38,8 +37,8 @@ use crate::sort::Sort;
 ///     .await
 ///     .unwrap();
 /// while let Some(res) = cursor.next().await {
-///     if let Ok(user) = res {
-///         println!("{:?}", user);
+///     if let Ok((id, user)) = res {
+///         println!("{} - {:?}", id, user);
 ///     }
 /// }
 /// # Ok(())
@@ -253,12 +252,13 @@ impl<C: Collection> Find<C> {
     /// # Errors
     ///
     /// This method fails if the mongodb encountered an error.
-    pub async fn query(self, client: &Client) -> crate::Result<Cursor<Document>> {
+    pub async fn query(self, client: &Client) -> crate::Result<TypedCursor<C>> {
         client
             .database()
             .collection::<Document>(C::COLLECTION)
             .find(self.filter, self.options)
             .await
+            .map(|cur| TypedCursor::from(cur))
             .map_err(crate::error::mongodb)
     }
 
@@ -275,14 +275,14 @@ impl<C: Collection> Find<C> {
     pub fn blocking(
         self,
         client: &crate::blocking::Client,
-    ) -> crate::Result<crate::blocking::Cursor> {
+    ) -> crate::Result<crate::blocking::TypedCursor<C>> {
         let resp = client.execute(crate::blocking::Request::Find(
             C::COLLECTION,
             self.filter,
             self.options,
         ))?;
         if let crate::blocking::Response::Find(r) = resp {
-            return Ok(r);
+            return Ok(crate::blocking::TypedCursor::from(r));
         }
         Err(crate::error::runtime(
             "incorrect response from blocking client",

mongod/tests/async.rs

@@ -1,5 +1,5 @@
 use futures::stream::StreamExt;
-use mongod::{AsFilter, AsUpdate, Comparator, TypedCursor, Updates};
+use mongod::{AsFilter, AsUpdate, Comparator, Updates};
 
 use user::User;
 
@@ -25,10 +25,9 @@ async fn async_client() {
 
     // Fetch
     let mut count: u32 = 0;
-    let cursor = client.find::<User, _>(None).await.unwrap();
-    let mut cursor: TypedCursor<User> = TypedCursor::from(cursor);
+    let mut cursor = client.find::<User, _>(None).await.unwrap();
     while let Some(res) = cursor.next().await {
-        let _user = res.unwrap();
+        let (_id, _user) = res.unwrap();
         count += 1;
     }
     assert_eq!(count, 2);

mongod/tests/blocking.rs

@@ -30,7 +30,7 @@ mod wrapper {
         let mut count: u32 = 0;
         let mut cursor = client.find::<User, _>(None).unwrap();
         while let Some(res) = cursor.next() {
-            let _user = res.unwrap();
+            let (_id, _user) = res.unwrap();
             count += 1;
         }
         assert_eq!(count, 2);