implement flush-responses approach

Author: ikonst

redis/connection.py

@@ -559,6 +559,7 @@ def __init__(
         self.set_parser(parser_class)
         self._connect_callbacks = []
         self._buffer_cutoff = 6000
+        self._expected_responses = 0
 
     def __repr__(self):
         repr_args = ",".join([f"{k}={v}" for k, v in self.repr_pieces()])
@@ -739,6 +740,7 @@ def disconnect(self, *args):
         except OSError:
             pass
         self._sock = None
+        self._expected_responses = 0
 
     def _send_ping(self):
         """Send PING, expect PONG in return"""
@@ -762,11 +764,17 @@ def send_packed_command(self, command, check_health=True):
         # guard against health check recursion
         if check_health:
             self.check_health()
+
+        # flush un-retrieved responses to any previous commands
+        while self._expected_responses:
+            _ = self.read_response(disable_decoding=True)
+
         try:
             if isinstance(command, str):
                 command = [command]
             for item in command:
                 self._sock.sendall(item)
+                self._expected_responses += 1
         except socket.timeout:
             self.disconnect()
             raise TimeoutError("Timeout writing to socket")
@@ -820,6 +828,8 @@ def read_response(self, disable_decoding=False):
             self.disconnect()
             raise
 
+        self._expected_responses -= 1
+
         if self.health_check_interval:
             self.next_health_check = time() + self.health_check_interval