Fix AsyncIter implementation of stream. (#1648)

When the `safe_iterators` feature isn't implemented, the `Stream` implementation of AsyncIter should return T, not `RedisResult<T>`.

Author: nihohit

Makefile

@@ -12,6 +12,11 @@ test:
 	@echo "===================================================================="
 	@RUSTFLAGS="-D warnings" REDISRS_SERVER_TYPE=tcp RUST_BACKTRACE=1 cargo nextest run --locked -p redis --no-default-features -F safe_iterators -E 'not test(test_module)'
 
+	@echo "===================================================================="
+	@echo "Testing Connection Type TCP without safe_iterators, but with async"
+	@echo "===================================================================="
+	@RUSTFLAGS="-D warnings -A deprecated" REDISRS_SERVER_TYPE=tcp RUST_BACKTRACE=1 cargo nextest run --locked -p redis --no-default-features -F tokio-comp -E 'not test(test_module)'
+
 	@echo "===================================================================="
 	@echo "Testing Connection Type TCP with all features and RESP2"
 	@echo "===================================================================="

redis/examples/async-scan.rs

@@ -10,6 +10,9 @@ async fn main() -> redis::RedisResult<()> {
     let _: () = con.set("async-key2", b"foo").await?;
 
     let iter: AsyncIter<String> = con.scan().await?;
+    #[cfg(not(feature = "safe_iterators"))]
+    let mut keys: Vec<String> = iter.collect().await;
+    #[cfg(feature = "safe_iterators")]
     let mut keys: Vec<_> = iter.map(std::result::Result::unwrap).collect().await;
 
     keys.sort();

redis/src/cmd.rs

@@ -262,13 +262,16 @@ impl<'a, T: FromRedisValue + 'a + Unpin + Send> AsyncIter<'a, T> {
     #[cfg(not(feature = "safe_iterators"))]
     #[inline]
     pub async fn next_item(&mut self) -> Option<T> {
-        StreamExt::next(self).await?.ok()
+        StreamExt::next(self).await
     }
 }
 
 #[cfg(feature = "aio")]
 impl<'a, T: FromRedisValue + Unpin + Send + 'a> Stream for AsyncIter<'a, T> {
+    #[cfg(feature = "safe_iterators")]
     type Item = RedisResult<T>;
+    #[cfg(not(feature = "safe_iterators"))]
+    type Item = T;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let this = self.get_mut();
@@ -289,7 +292,11 @@ impl<'a, T: FromRedisValue + Unpin + Send + 'a> Stream for AsyncIter<'a, T> {
                 }
                 Poll::Ready((iter, value)) => {
                     this.inner = IterOrFuture::Iter(iter);
-                    Poll::Ready(value)
+
+                    #[cfg(feature = "safe_iterators")]
+                    return Poll::Ready(value);
+                    #[cfg(not(feature = "safe_iterators"))]
+                    Poll::Ready(value.map(|res| res.ok()).flatten())
                 }
             },
             IterOrFuture::Empty => unreachable!(),

redis/tests/test_async.rs

@@ -939,6 +939,9 @@ mod basic_async {
                 }
 
                 let iter: redis::AsyncIter<String> = con.scan().await.unwrap();
+                #[cfg(not(feature = "safe_iterators"))]
+                let mut keys_from_redis: Vec<String> = iter.collect().await;
+                #[cfg(feature = "safe_iterators")]
                 let mut keys_from_redis: Vec<_> =
                     iter.map(std::result::Result::unwrap).collect().await;
                 keys_from_redis.sort();