RUST-1719 Fix panic after SessionCursor::with_type is called (#928)

Co-authored-by: Isabel Atkinson <isabel.atkinson@mongodb.com>

src/cursor/session.rs

@@ -309,17 +309,15 @@ impl<T> SessionCursor<T> {
     where
         D: Deserialize<'a>,
     {
-        let out = SessionCursor {
+        SessionCursor {
             client: self.client.clone(),
             info: self.info.clone(),
             state: Some(self.take_state()),
             drop_address: self.drop_address.take(),
             _phantom: Default::default(),
             #[cfg(test)]
             kill_watcher: self.kill_watcher.take(),
-        };
-        self.mark_exhausted(); // prevent a `kill_cursor` call in `drop`
-        out
+        }
     }
 
     pub(crate) fn address(&self) -> &ServerAddress {
@@ -346,12 +344,8 @@ impl<T> SessionCursor<T> {
 }
 
 impl<T> SessionCursor<T> {
-    fn mark_exhausted(&mut self) {
-        self.state.as_mut().unwrap().exhausted = true;
-    }
-
     pub(crate) fn is_exhausted(&self) -> bool {
-        self.state.as_ref().unwrap().exhausted
+        self.state.as_ref().map_or(true, |state| state.exhausted)
     }
 
     #[cfg(test)]

src/test/cursor.rs

@@ -235,3 +235,33 @@ async fn borrowed_deserialization() {
         i += 1;
     }
 }
+
+#[cfg_attr(feature = "tokio-runtime", tokio::test)]
+#[cfg_attr(feature = "async-std-runtime", async_std::test)]
+async fn session_cursor_with_type() {
+    let _guard: RwLockReadGuard<()> = LOCK.run_concurrently().await;
+    let client = TestClient::new().await;
+
+    let mut session = client.start_session(None).await.unwrap();
+    let coll = client.database("db").collection("coll");
+    coll.drop_with_session(None, &mut session).await.unwrap();
+
+    coll.insert_many_with_session(
+        vec![doc! { "x": 1 }, doc! { "x": 2 }, doc! { "x": 3 }],
+        None,
+        &mut session,
+    )
+    .await
+    .unwrap();
+
+    let mut cursor: crate::SessionCursor<bson::Document> = coll
+        .find_with_session(doc! {}, None, &mut session)
+        .await
+        .unwrap();
+
+    let _ = cursor.next(&mut session).await.unwrap().unwrap();
+
+    let mut cursor_with_type: crate::SessionCursor<bson::RawDocumentBuf> = cursor.with_type();
+
+    let _ = cursor_with_type.next(&mut session).await.unwrap().unwrap();
+}