Simplify result stream by removing the RowOrSummary enum

knutwalker · knutwalker · commit 15692a8d8fc5 · 2025-02-11T07:29:02.000+01:00
The result summary can now only be returned from `finish`, which also consumes the stream.
The next_or_summary and *_items_* methods are removed.
This also simplifies the API for finish, the summary no longer has to be returned as an option.
Before it could have already been returned by a call to next_or_summary, which would result in a None being returned from finish.
Having the API like that is also closer to APIs from the product drivers
diff --git a/lib/include/result_summary.rs b/lib/include/result_summary.rs
@@ -22,45 +22,18 @@
     }
 
     //
-    // next_or_summary
+    // next + finish
 
     let mut stream = graph
         .execute(query("CREATE (n:Node {prop: 'frobnicate'}) RETURN n"))
         .await
         .unwrap();
 
-    let Ok(Some(row)) = stream.next_or_summary().await else { panic!() };
-    assert!(row.row().is_some());
-    assert!(row.summary().is_none());
+    let Ok(Some(row)) = stream.next().await else { panic!() };
+    assert_item(row.to().unwrap());
 
-    assert_item(row.row().unwrap().to().unwrap());
-
-    let Ok(Some(row)) = stream.next_or_summary().await else { panic!() };
-    assert!(row.row().is_none());
-    assert!(row.summary().is_some());
-
-    assert_summary(row.summary().unwrap());
-
-
-    //
-    // as_items
-
-    let mut stream = graph
-        .execute(query("CREATE (n:Node {prop: 'frobnicate'}) RETURN n"))
-        .await
-        .unwrap();
-
-    let items = stream.as_items::<N>()
-        .try_collect::<Vec<_>>()
-        .await
-        .unwrap();
-
-    for item in items {
-        match item {
-            RowItem::Row(row) => assert_item(row),
-            RowItem::Summary(summary) => assert_summary(&summary),
-        }
-    }
+    let Ok(summary) = stream.finish().await else { panic!() };
+    assert_summary(&summary);
 
 
     //
@@ -76,7 +49,7 @@
         .await
         .unwrap();
 
-    let Ok(Some(summary)) = stream.finish().await else { panic!() };
+    let Ok(summary) = stream.finish().await else { panic!() };
 
     for item in items {
         assert_item(item);
diff --git a/lib/src/stream.rs b/lib/src/stream.rs
@@ -27,7 +27,7 @@ type BoxedSummary = Box<ResultSummary>;
 type BoxedSummary = ();
 
 #[cfg(feature = "unstable-bolt-protocol-impl-v2")]
-type FinishResult = Option<ResultSummary>;
+type FinishResult = ResultSummary;
 #[cfg(not(feature = "unstable-bolt-protocol-impl-v2"))]
 type FinishResult = ();
 
@@ -84,7 +84,7 @@ impl DetachedRowStream {
 pub enum RowItem<T = Row> {
     Row(T),
     #[cfg(feature = "unstable-result-summary")]
-    Summary(Box<ResultSummary>),
+    Summary(BoxedSummary),
 }
 
 impl<T> RowItem<T> {
@@ -113,7 +113,7 @@ impl<T> RowItem<T> {
     }
 
     #[cfg(feature = "unstable-result-summary")]
-    pub fn into_summary(self) -> Option<Box<ResultSummary>> {
+    pub fn into_summary(self) -> Option<BoxedSummary> {
         match self {
             RowItem::Summary(summary) => Some(summary),
             _ => None,
@@ -125,22 +125,10 @@ impl RowStream {
     /// A call to next() will return a row from an internal buffer if the buffer has any entries,
     /// if the buffer is empty and the server has more rows left to consume, then a new batch of rows
     /// are fetched from the server (using the fetch_size value configured see [`crate::ConfigBuilder::fetch_size`])
-    pub async fn next(&mut self, handle: impl TransactionHandle) -> Result<Option<Row>> {
-        self.next_or_summary(handle)
-            .await
-            .map(|item| item.and_then(RowItem::into_row))
-    }
-
-    /// A call to next_or_summary() will return a row from an internal buffer if the buffer has any entries,
-    /// if the buffer is empty and the server has more rows left to consume, then a new batch of rows
-    /// are fetched from the server (using the fetch_size value configured see [`crate::ConfigBuilder::fetch_size`])
-    pub async fn next_or_summary(
-        &mut self,
-        mut handle: impl TransactionHandle,
-    ) -> Result<Option<RowItem>> {
+    pub async fn next(&mut self, mut handle: impl TransactionHandle) -> Result<Option<Row>> {
         loop {
             if let Some(row) = self.buffer.pop_front() {
-                return Ok(Some(RowItem::Row(row)));
+                return Ok(Some(row));
             }
 
             #[cfg(feature = "unstable-bolt-protocol-impl-v2")]
@@ -167,16 +155,13 @@ impl RowStream {
                             Response::Success(Streaming::HasMore) => break State::Ready,
                             Response::Success(Streaming::Done(mut s)) => {
                                 s.set_t_first(self.available_after);
-                                break State::Complete(Some(s));
+                                break State::Complete(s);
                             }
                             otherwise => return Err(otherwise.into_error("PULL")),
                         }
                     };
-                } else if let State::Complete(ref mut summary) = self.state {
-                    break match summary.take() {
-                        Some(summary) => Ok(Some(RowItem::Summary(summary))),
-                        None => Ok(None),
-                    };
+                } else if let State::Complete(_) = self.state {
+                    break Ok(None);
                 }
             }
 
@@ -227,25 +212,21 @@ impl RowStream {
                 }?;
                 let summary = match summary {
                     Summary::Success(s) => match s.metadata {
-                        Streaming::Done(summary) => Some(*summary),
+                        Streaming::Done(summary) => *summary,
                         Streaming::HasMore => {
-                            // this should never happen
-                            None
+                            unreachable!("Query returned has_more after a discard_all");
                         }
                     },
                     Summary::Ignored => {
-                        self.state = State::Complete(None);
                         return Err(Error::RequestIgnoredError);
                     }
                     Summary::Failure(f) => {
-                        self.state = State::Complete(None);
                         return Err(f.into_error());
                     }
                 };
-                self.state = State::Complete(None);
                 Ok(summary)
             }
-            State::Complete(summary) => Ok(summary.map(|o| *o)),
+            State::Complete(summary) => Ok(*summary),
         }
 
         #[cfg(not(feature = "unstable-bolt-protocol-impl-v2"))]
@@ -282,17 +263,6 @@ impl RowStream {
         self.convert_rows(handle, Ok)
     }
 
-    /// Turns this RowStream into a [`futures::stream::TryStream`] where
-    /// every element is a [`RowItem`].
-    ///
-    /// The stream can only be converted once.
-    pub fn as_row_items<'this, 'db: 'this>(
-        &'this mut self,
-        handle: impl TransactionHandle + 'db,
-    ) -> impl TryStream<Ok = RowItem, Error = Error> + 'this {
-        self.convert_with_summary(handle, Ok)
-    }
-
     /// Turns this RowStream into a [`futures::stream::TryStream`] where
     /// every row is converted into a `T` by calling [`crate::row::Row::to`].
     ///
@@ -306,17 +276,6 @@ impl RowStream {
         self.convert_rows(handle, |row| row.to::<T>())
     }
 
-    /// Turns this RowStream into a [`futures::stream::TryStream`] where
-    /// every row is converted into a [`RowItem<T>`] by calling [`crate::row::Row::to`].
-    ///
-    /// The stream can only be converted once.
-    pub fn as_items<'this, 'db: 'this, T: DeserializeOwned + 'this>(
-        &'this mut self,
-        handle: impl TransactionHandle + 'db,
-    ) -> impl TryStream<Ok = RowItem<T>, Error = Error> + 'this {
-        self.convert_with_summary(handle, |row| row.to::<T>())
-    }
-
     /// Turns this RowStream into a [`futures::stream::TryStream`] where
     /// the value at the given column is converted into a `T`
     /// by calling [`crate::row::Row::get`].
@@ -331,58 +290,18 @@ impl RowStream {
         self.convert_rows(handle, move |row| row.get::<T>(column))
     }
 
-    /// Turns this RowStream into a [`futures::stream::TryStream`] where
-    /// the value at the given column is converted into a [`RowItem<T>`]
-    /// by calling [`crate::row::Row::get`].
-    ///
-    /// The stream can only be converted once.
-    pub fn column_to_items<'this, 'db: 'this, T: DeserializeOwned + 'db>(
-        &'this mut self,
-        handle: impl TransactionHandle + 'db,
-        column: &'db str,
-    ) -> impl TryStream<Ok = RowItem<T>, Error = Error> + 'this {
-        self.convert_with_summary(handle, move |row| row.get::<T>(column))
-    }
-
     fn convert_rows<'this, 'db: 'this, T: 'this>(
         &'this mut self,
         handle: impl TransactionHandle + 'db,
         convert: impl Fn(Row) -> Result<T, DeError> + 'this,
     ) -> impl TryStream<Ok = T, Error = Error> + 'this {
         try_unfold((self, handle, convert), |(stream, mut hd, de)| async move {
-            match stream.next_or_summary(&mut hd).await {
-                Ok(Some(RowItem::Row(row))) => match de(row) {
+            match stream.next(&mut hd).await? {
+                Some(row) => match de(row) {
                     Ok(res) => Ok(Some((res, (stream, hd, de)))),
                     Err(e) => Err(Error::DeserializationError(e)),
                 },
-                #[cfg(feature = "unstable-result-summary")]
-                Ok(Some(RowItem::Summary(summary))) => {
-                    stream.state = State::Complete(Some(summary));
-                    Ok(None)
-                }
-                Ok(None) => Ok(None),
-                Err(e) => Err(e),
-            }
-        })
-    }
-
-    fn convert_with_summary<'this, 'db: 'this, T>(
-        &'this mut self,
-        handle: impl TransactionHandle + 'db,
-        convert: impl Fn(Row) -> Result<T, DeError> + 'this,
-    ) -> impl TryStream<Ok = RowItem<T>, Error = Error> + 'this {
-        try_unfold((self, handle, convert), |(stream, mut hd, de)| async move {
-            match stream.next_or_summary(&mut hd).await {
-                Ok(Some(RowItem::Row(row))) => match de(row) {
-                    Ok(res) => Ok(Some((RowItem::Row(res), (stream, hd, de)))),
-                    Err(e) => Err(Error::DeserializationError(e)),
-                },
-                #[cfg(feature = "unstable-result-summary")]
-                Ok(Some(RowItem::Summary(summary))) => {
-                    Ok(Some((RowItem::Summary(summary), (stream, hd, de))))
-                }
-                Ok(None) => Ok(None),
-                Err(e) => Err(e),
+                None => Ok(None),
             }
         })
     }
@@ -396,13 +315,6 @@ impl DetachedRowStream {
         self.stream.next(&mut self.connection).await
     }
 
-    /// A call to next_or_summary() will return a row from an internal buffer if the buffer has any entries,
-    /// if the buffer is empty and the server has more rows left to consume, then a new batch of rows
-    /// are fetched from the server (using the fetch_size value configured see [`crate::ConfigBuilder::fetch_size`])
-    pub async fn next_or_summary(&mut self) -> Result<Option<RowItem>> {
-        self.stream.next_or_summary(&mut self.connection).await
-    }
-
     /// Stop consuming the stream and return a summary, if available.
     /// Stopping the stream will also discard any messages on the server side.
     pub async fn finish(mut self) -> Result<FinishResult> {
@@ -419,14 +331,6 @@ impl DetachedRowStream {
         self.stream.into_stream(&mut self.connection)
     }
 
-    /// Turns this RowStream into a [`futures::stream::TryStream`] where
-    /// every element is a [`RowItem`].
-    ///
-    /// The stream can only be converted once.
-    pub fn as_row_items(&mut self) -> impl TryStream<Ok = RowItem, Error = Error> + '_ {
-        self.stream.as_row_items(&mut self.connection)
-    }
-
     /// Turns this RowStream into a [`futures::stream::TryStream`] where
     /// every row is converted into a `T` by calling [`crate::row::Row::to`].
     ///
@@ -439,16 +343,6 @@ impl DetachedRowStream {
         self.stream.into_stream_as(&mut self.connection)
     }
 
-    /// Turns this RowStream into a [`futures::stream::TryStream`] where
-    /// every row is converted into a [`RowItem<T>`] by calling [`crate::row::Row::to`].
-    ///
-    /// The stream can only be converted once.
-    pub fn as_items<'this, T: DeserializeOwned + 'this>(
-        &'this mut self,
-    ) -> impl TryStream<Ok = RowItem<T>, Error = Error> + 'this {
-        self.stream.as_items(&mut self.connection)
-    }
-
     /// Turns this RowStream into a [`futures::stream::TryStream`] where
     /// the value at the given column is converted into a `T`
     /// by calling [`crate::row::Row::get`].
@@ -461,22 +355,10 @@ impl DetachedRowStream {
     ) -> impl TryStream<Ok = T, Error = Error> + 'this {
         self.stream.column_into_stream(&mut self.connection, column)
     }
-
-    /// Turns this RowStream into a [`futures::stream::TryStream`] where
-    /// the value at the given column is converted into a [`RowItem<T>`]
-    /// by calling [`crate::row::Row::get`].
-    ///
-    /// The stream can only be converted once.
-    pub fn column_to_items<'this, 'db: 'this, T: DeserializeOwned + 'db>(
-        &'this mut self,
-        column: &'db str,
-    ) -> impl TryStream<Ok = RowItem<T>, Error = Error> + 'this {
-        self.stream.column_to_items(&mut self.connection, column)
-    }
 }
 
 #[derive(Clone, PartialEq, Debug)]
 enum State {
     Ready,
-    Complete(Option<BoxedSummary>),
+    Complete(BoxedSummary),
 }
