remove sub-pipeline

Signed-off-by: Shoham Elias <[email protected]>

glide-core/redis-rs/redis/src/cluster_async/mod.rs

@@ -291,6 +291,7 @@ where
                     offset,
                     count,
                     route: route.into(),
+                    sub_pipeline: false,
                 },
                 sender,
             })
@@ -612,6 +613,7 @@ enum CmdArg<C> {
         offset: usize,
         count: usize,
         route: InternalSingleNodeRouting<C>,
+        sub_pipeline: bool,
     },
     ClusterScan {
         // struct containing the arguments for the cluster scan command - scan state cursor, match pattern, count and object type.
@@ -2115,8 +2117,9 @@ where
                 offset,
                 count,
                 route,
+                sub_pipeline,
             } => {
-                if pipeline.is_atomic() || pipeline.is_sub_pipeline() {
+                if pipeline.is_atomic() || sub_pipeline {
                     // If the pipeline is atomic (i.e., a transaction) or if the pipeline is already splitted into sub-pipelines, we can send it as is, with no need to split it into sub-pipelines.
                     Self::try_pipeline_request(
                         pipeline,

glide-core/redis-rs/redis/src/cluster_async/pipeline_routing.rs

@@ -39,10 +39,8 @@ pub type NodePipelineMap<C> = HashMap<String, NodePipelineContext<C>>;
 
 impl<C> NodePipelineContext<C> {
     fn new(connection: C) -> Self {
-        let mut pipeline = Pipeline::new();
-        pipeline.sub_pipeline();
         Self {
-            pipeline,
+            pipeline: Pipeline::new(),
             connection,
             command_indices: Vec::new(),
         }
@@ -311,6 +309,8 @@ where
                         address: address.clone(),
                         conn: async { context.connection }.boxed().shared(),
                     },
+                    // mark it as a sub-pipeline mode
+                    sub_pipeline: true,
                 },
             },
         });

glide-core/redis-rs/redis/src/pipeline.rs

@@ -12,7 +12,6 @@ pub struct Pipeline {
     commands: Vec<Cmd>,
     transaction_mode: bool,
     ignored_commands: HashSet<usize>,
-    is_sub_pipeline: bool,
 }
 
 /// A pipeline allows you to send multiple commands in one go to the
@@ -49,7 +48,6 @@ impl Pipeline {
             commands: Vec::with_capacity(capacity),
             transaction_mode: false,
             ignored_commands: HashSet::new(),
-            is_sub_pipeline: false,
         }
     }
 
@@ -72,13 +70,6 @@ impl Pipeline {
         self
     }
 
-    /// Enables sub-pipeline mode, indicating that this pipeline is part of a larger pipeline
-    /// split across multiple nodes.
-    pub fn sub_pipeline(&mut self) -> &mut Pipeline {
-        self.is_sub_pipeline = true;
-        self
-    }
-
     /// Returns the encoded pipeline commands.
     pub fn get_packed_pipeline(&self) -> Vec<u8> {
         encode_pipeline(&self.commands, self.transaction_mode)
@@ -230,11 +221,6 @@ impl Pipeline {
     pub fn is_empty(&self) -> bool {
         self.commands.is_empty()
     }
-
-    /// Returns whether the pipeline is in sub-pipeline mode.
-    pub fn is_sub_pipeline(&self) -> bool {
-        self.is_sub_pipeline
-    }
 }
 
 fn encode_pipeline(cmds: &[Cmd], atomic: bool) -> Vec<u8> {