[fix] make Pipeline handle returned bulks correctly

redis/src/aio/multiplexed_connection.rs

@@ -129,7 +129,7 @@ where
             match result {
                 Ok(item) => {
                     entry.buffer = Some(match entry.buffer.take() {
-                        Some(Value::Bulk(mut values)) => {
+                        Some(Value::Bulk(mut values)) if entry.current_response_count > 1 => {
                             values.push(item);
                             Value::Bulk(values)
                         }

redis/tests/test_async.rs

@@ -1,7 +1,7 @@
 use futures::{prelude::*, StreamExt};
 use redis::{
     aio::{ConnectionLike, MultiplexedConnection},
-    cmd, AsyncCommands, ErrorKind, RedisResult,
+    cmd, pipe, AsyncCommands, ErrorKind, RedisResult,
 };
 
 use crate::support::*;
@@ -181,6 +181,23 @@ fn test_error(con: &MultiplexedConnection) -> impl Future<Output = RedisResult<(
     }
 }
 
+#[test]
+fn test_pipe_over_multiplexed_connection() {
+    let ctx = TestContext::new();
+    block_on_all(async move {
+        let mut con = ctx.multiplexed_async_connection().await?;
+        let mut pipe = pipe();
+        pipe.zrange("zset", 0, 0);
+        pipe.zrange("zset", 0, 0);
+        let frames = con.send_packed_commands(&pipe, 0, 2).await?;
+        assert_eq!(frames.len(), 2);
+        assert!(matches!(frames[0], redis::Value::Bulk(_)));
+        assert!(matches!(frames[1], redis::Value::Bulk(_)));
+        RedisResult::Ok(())
+    })
+    .unwrap();
+}
+
 #[test]
 fn test_args_multiplexed_connection() {
     let ctx = TestContext::new();