Change TryStream to Stream

cathay4t · cathay4t · commit f2ac86c1117b · 2026-02-03T15:01:59.000+08:00
* Align with `rtnetlink` crate.
* `Stream` allow us to do map, join, select actions while TryStream
  cannot.

Signed-off-by: Gris Ge &lt;fge@redhat.com&gt;
diff --git a/.gitignore b/.gitignore
@@ -1,5 +1,5 @@
 Cargo.lock
 target
 vendor/
-
+tags
 *.swp
diff --git a/examples/dump_channels.rs b/examples/dump_channels.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::stream::TryStreamExt;
+use futures_util::stream::StreamExt;
 
 // Once we find a way to load netsimdev kernel module in CI, we can convert this
 // to a test
@@ -16,10 +16,11 @@ async fn get_channels(iface_name: Option<&str>) {
     let (connection, mut handle, _) = ethtool::new_connection().unwrap();
     tokio::spawn(connection);
 
-    let mut channel_handle = handle.channel().get(iface_name).execute().await;
+    let mut channel_handle =
+        handle.channel().get(iface_name).execute().await.unwrap();
 
     let mut msgs = Vec::new();
-    while let Some(msg) = channel_handle.try_next().await.unwrap() {
+    while let Some(Ok(msg)) = channel_handle.next().await {
         msgs.push(msg);
     }
     assert!(!msgs.is_empty());
diff --git a/examples/dump_coalesce.rs b/examples/dump_coalesce.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::stream::TryStreamExt;
+use futures_util::stream::StreamExt;
 
 // Once we find a way to load netsimdev kernel module in CI, we can convert this
 // to a test
@@ -16,11 +16,12 @@ async fn get_coalesce(iface_name: Option<&str>) {
     let (connection, mut handle, _) = ethtool::new_connection().unwrap();
     tokio::spawn(connection);
 
-    let mut coalesce_handle = handle.coalesce().get(iface_name).execute().await;
+    let mut coalesce_handle =
+        handle.coalesce().get(iface_name).execute().await.unwrap();
 
     let mut msgs = Vec::new();
-    while let Some(msg) = coalesce_handle.try_next().await.unwrap() {
-        msgs.push(msg);
+    while let Some(Ok(msg)) = coalesce_handle.next().await {
+        msgs.push(msg.payload);
     }
     assert!(!msgs.is_empty());
     for msg in msgs {
diff --git a/examples/dump_eeprom_page.rs b/examples/dump_eeprom_page.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::stream::TryStreamExt;
+use futures_util::stream::StreamExt;
 
 fn main() {
     let rt = tokio::runtime::Builder::new_current_thread()
@@ -19,10 +19,11 @@ async fn get_eeprom(iface_name: Option<&str>) {
         .eeprom()
         .get(iface_name, 0, 1, 0, 0, 0x50)
         .execute()
-        .await;
+        .await
+        .unwrap();
 
     let mut msgs = Vec::new();
-    while let Some(msg) = eeprom_handle.try_next().await.unwrap() {
+    while let Some(Ok(msg)) = eeprom_handle.next().await {
         msgs.push(msg);
     }
     assert!(!msgs.is_empty());
diff --git a/examples/dump_features.rs b/examples/dump_features.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::stream::TryStreamExt;
+use futures_util::stream::StreamExt;
 
 // Once we find a way to load netsimdev kernel module in CI, we can convert this
 // to a test
@@ -16,10 +16,11 @@ async fn get_feature(iface_name: Option<&str>) {
     let (connection, mut handle, _) = ethtool::new_connection().unwrap();
     tokio::spawn(connection);
 
-    let mut feature_handle = handle.feature().get(iface_name).execute().await;
+    let mut feature_handle =
+        handle.feature().get(iface_name).execute().await.unwrap();
 
     let mut msgs = Vec::new();
-    while let Some(msg) = feature_handle.try_next().await.unwrap() {
+    while let Some(Ok(msg)) = feature_handle.next().await {
         msgs.push(msg);
     }
     assert!(!msgs.is_empty());
diff --git a/examples/dump_fec.rs b/examples/dump_fec.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::stream::TryStreamExt;
+use futures_util::stream::StreamExt;
 
 // Once we find a way to load netsimdev kernel module in CI, we can convert this
 // to a test
@@ -17,10 +17,10 @@ async fn get_fec(iface_name: Option<&str>) {
     let (connection, mut handle, _) = ethtool::new_connection().unwrap();
     tokio::spawn(connection);
 
-    let mut fec_handle = handle.fec().get(iface_name).execute().await;
+    let mut fec_handle = handle.fec().get(iface_name).execute().await.unwrap();
 
     let mut msgs = Vec::new();
-    while let Some(msg) = fec_handle.try_next().await.unwrap() {
+    while let Some(Ok(msg)) = fec_handle.next().await {
         msgs.push(msg);
     }
     assert!(!msgs.is_empty());
diff --git a/examples/dump_link_mode.rs b/examples/dump_link_mode.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::stream::TryStreamExt;
+use futures_util::stream::StreamExt;
 
 // Once we find a way to load netsimdev kernel module in CI, we can convert this
 // to a test
@@ -17,10 +17,10 @@ async fn get_link_mode(iface_name: Option<&str>) {
     tokio::spawn(connection);
 
     let mut link_mode_handle =
-        handle.link_mode().get(iface_name).execute().await;
+        handle.link_mode().get(iface_name).execute().await.unwrap();
 
     let mut msgs = Vec::new();
-    while let Some(msg) = link_mode_handle.try_next().await.unwrap() {
+    while let Some(Ok(msg)) = link_mode_handle.next().await {
         msgs.push(msg);
     }
     assert!(!msgs.is_empty());
diff --git a/examples/dump_pause.rs b/examples/dump_pause.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::stream::TryStreamExt;
+use futures_util::stream::StreamExt;
 
 // Once we find a way to load netsimdev kernel module in CI, we can convert this
 // to a test
@@ -17,10 +17,11 @@ async fn get_pause(iface_name: Option<&str>) {
     let (connection, mut handle, _) = ethtool::new_connection().unwrap();
     tokio::spawn(connection);
 
-    let mut pause_handle = handle.pause().get(iface_name).execute().await;
+    let mut pause_handle =
+        handle.pause().get(iface_name).execute().await.unwrap();
 
     let mut msgs = Vec::new();
-    while let Some(msg) = pause_handle.try_next().await.unwrap() {
+    while let Some(Ok(msg)) = pause_handle.next().await {
         msgs.push(msg);
     }
     assert!(!msgs.is_empty());
diff --git a/examples/dump_rings.rs b/examples/dump_rings.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::stream::TryStreamExt;
+use futures_util::stream::StreamExt;
 
 // Once we find a way to load netsimdev kernel module in CI, we can convert this
 // to a test
@@ -16,10 +16,11 @@ async fn get_ring(iface_name: Option<&str>) {
     let (connection, mut handle, _) = ethtool::new_connection().unwrap();
     tokio::spawn(connection);
 
-    let mut ring_handle = handle.ring().get(iface_name).execute().await;
+    let mut ring_handle =
+        handle.ring().get(iface_name).execute().await.unwrap();
 
     let mut msgs = Vec::new();
-    while let Some(msg) = ring_handle.try_next().await.unwrap() {
+    while let Some(Ok(msg)) = ring_handle.next().await {
         msgs.push(msg);
     }
     assert!(!msgs.is_empty());
diff --git a/examples/dump_tsinfo.rs b/examples/dump_tsinfo.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::stream::TryStreamExt;
+use futures_util::stream::StreamExt;
 
 // Once we find a way to load netsimdev kernel module in CI, we can convert this
 // to a test
@@ -17,10 +17,11 @@ async fn get_tsinfo(iface_name: Option<&str>) {
     let (connection, mut handle, _) = ethtool::new_connection().unwrap();
     tokio::spawn(connection);
 
-    let mut tsinfo_handle = handle.tsinfo().get(iface_name).execute().await;
+    let mut tsinfo_handle =
+        handle.tsinfo().get(iface_name).execute().await.unwrap();
 
     let mut msgs = Vec::new();
-    while let Some(msg) = tsinfo_handle.try_next().await.unwrap() {
+    while let Some(Ok(msg)) = tsinfo_handle.next().await {
         msgs.push(msg);
     }
     assert!(!msgs.is_empty());
diff --git a/src/channel/get.rs b/src/channel/get.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::TryStream;
+use futures_util::Stream;
 use netlink_packet_generic::GenlMessage;
 
 use crate::{ethtool_execute, EthtoolError, EthtoolHandle, EthtoolMessage};
@@ -20,8 +20,10 @@ impl EthtoolChannelGetRequest {
 
     pub async fn execute(
         self,
-    ) -> impl TryStream<Ok = GenlMessage<EthtoolMessage>, Error = EthtoolError>
-    {
+    ) -> Result<
+        impl Stream<Item = Result<GenlMessage<EthtoolMessage>, EthtoolError>>,
+        EthtoolError,
+    > {
         let EthtoolChannelGetRequest {
             mut handle,
             iface_name,
diff --git a/src/coalesce/get.rs b/src/coalesce/get.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::TryStream;
+use futures_util::Stream;
 use netlink_packet_generic::GenlMessage;
 
 use crate::{ethtool_execute, EthtoolError, EthtoolHandle, EthtoolMessage};
@@ -20,8 +20,10 @@ impl EthtoolCoalesceGetRequest {
 
     pub async fn execute(
         self,
-    ) -> impl TryStream<Ok = GenlMessage<EthtoolMessage>, Error = EthtoolError>
-    {
+    ) -> Result<
+        impl Stream<Item = Result<GenlMessage<EthtoolMessage>, EthtoolError>>,
+        EthtoolError,
+    > {
         let EthtoolCoalesceGetRequest {
             mut handle,
             iface_name,
diff --git a/src/eeprom/get.rs b/src/eeprom/get.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::TryStream;
+use futures_util::Stream;
 use netlink_packet_generic::GenlMessage;
 
 use crate::{ethtool_execute, EthtoolError, EthtoolHandle, EthtoolMessage};
@@ -38,8 +38,10 @@ impl EthtoolModuleEEPROMGetRequest {
 
     pub async fn execute(
         self,
-    ) -> impl TryStream<Ok = GenlMessage<EthtoolMessage>, Error = EthtoolError>
-    {
+    ) -> Result<
+        impl Stream<Item = Result<GenlMessage<EthtoolMessage>, EthtoolError>>,
+        EthtoolError,
+    > {
         let EthtoolModuleEEPROMGetRequest {
             mut handle,
             iface_name,
diff --git a/src/feature/get.rs b/src/feature/get.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::TryStream;
+use futures_util::Stream;
 use netlink_packet_generic::GenlMessage;
 
 use crate::{ethtool_execute, EthtoolError, EthtoolHandle, EthtoolMessage};
@@ -20,8 +20,10 @@ impl EthtoolFeatureGetRequest {
 
     pub async fn execute(
         self,
-    ) -> impl TryStream<Ok = GenlMessage<EthtoolMessage>, Error = EthtoolError>
-    {
+    ) -> Result<
+        impl Stream<Item = Result<GenlMessage<EthtoolMessage>, EthtoolError>>,
+        EthtoolError,
+    > {
         let EthtoolFeatureGetRequest {
             mut handle,
             iface_name,
diff --git a/src/fec/get.rs b/src/fec/get.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::TryStream;
+use futures_util::Stream;
 use netlink_packet_generic::GenlMessage;
 
 use crate::{ethtool_execute, EthtoolError, EthtoolHandle, EthtoolMessage};
@@ -20,8 +20,10 @@ impl EthtoolFecGetRequest {
 
     pub async fn execute(
         self,
-    ) -> impl TryStream<Ok = GenlMessage<EthtoolMessage>, Error = EthtoolError>
-    {
+    ) -> Result<
+        impl Stream<Item = Result<GenlMessage<EthtoolMessage>, EthtoolError>>,
+        EthtoolError,
+    > {
         let EthtoolFecGetRequest {
             mut handle,
             iface_name,
diff --git a/src/handle.rs b/src/handle.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::{future::Either, FutureExt, Stream, StreamExt, TryStream};
+use futures_util::{Stream, StreamExt};
 use genetlink::GenetlinkHandle;
 use netlink_packet_core::DecodeError;
 use netlink_packet_core::{
@@ -83,7 +83,10 @@ pub(crate) async fn ethtool_execute(
     handle: &mut EthtoolHandle,
     is_dump: bool,
     ethtool_msg: EthtoolMessage,
-) -> impl TryStream<Ok = GenlMessage<EthtoolMessage>, Error = EthtoolError> {
+) -> Result<
+    impl Stream<Item = Result<GenlMessage<EthtoolMessage>, EthtoolError>>,
+    EthtoolError,
+> {
     let nl_header_flags = if is_dump {
         // The NLM_F_ACK is required due to bug of kernel:
         //  https://bugzilla.redhat.com/show_bug.cgi?id=1953847
@@ -101,16 +104,8 @@ pub(crate) async fn ethtool_execute(
 
     nl_msg.header.flags = nl_header_flags;
 
-    match handle.request(nl_msg).await {
-        Ok(response) => {
-            Either::Left(response.map(move |msg| Ok(try_ethtool!(msg))))
-        }
-        Err(e) => Either::Right(
-            futures_util::future::err::<
-                GenlMessage<EthtoolMessage>,
-                EthtoolError,
-            >(e)
-            .into_stream(),
-        ),
-    }
+    Ok(handle
+        .request(nl_msg)
+        .await?
+        .map(move |msg| Ok(try_ethtool!(msg))))
 }
diff --git a/src/link_mode/get.rs b/src/link_mode/get.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::TryStream;
+use futures_util::Stream;
 use netlink_packet_generic::GenlMessage;
 
 use crate::{ethtool_execute, EthtoolError, EthtoolHandle, EthtoolMessage};
@@ -20,8 +20,10 @@ impl EthtoolLinkModeGetRequest {
 
     pub async fn execute(
         self,
-    ) -> impl TryStream<Ok = GenlMessage<EthtoolMessage>, Error = EthtoolError>
-    {
+    ) -> Result<
+        impl Stream<Item = Result<GenlMessage<EthtoolMessage>, EthtoolError>>,
+        EthtoolError,
+    > {
         let EthtoolLinkModeGetRequest {
             mut handle,
             iface_name,
diff --git a/src/pause/get.rs b/src/pause/get.rs
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 
-use futures_util::TryStream;
+use futures_util::Stream;
 use netlink_packet_generic::GenlMessage;
 
 use crate::{ethtool_execute, EthtoolError, EthtoolHandle, EthtoolMessage};
@@ -20,8 +20,10 @@ impl EthtoolPauseGetRequest {
 
     pub async fn execute(
         self,
-    ) -> impl TryStream<Ok = GenlMessage<EthtoolMessage>, Error = EthtoolError>
-    {
+    ) -> Result<
+        impl Stream<Item = Result<GenlMessage<EthtoolMessage>, EthtoolError>>,
+        EthtoolError,
+    > {
         let EthtoolPauseGetRequest {
             mut handle,
             iface_name,
diff --git a/src/ring/get.rs b/src/ring/get.rs
diff --git a/src/tsinfo/get.rs b/src/tsinfo/get.rs
diff --git a/tests/dump_link_modes.rs b/tests/dump_link_modes.rs
diff --git a/tests/get_features_lo.rs b/tests/get_features_lo.rs

-Original file line number
+Diff line change
@@ @@ -1,5 +1,5 @@ @@
 Cargo.lock
 target
 vendor/
+-
 +tags
 *.swp