Added inflight_requests_limit

Signed-off-by: GilboaAWS <gilboabg@amazon.com>

glide-core/src/socket_listener.rs

@@ -496,7 +496,7 @@ fn handle_request(
             |current_inflight| {
                 if current_inflight as u32 >= client.get_inflight_requests_limit() {
                     // In case counter reached the limit, don't update and return Err.
-                    return None;
+                    None
                 } else {
                     Some(current_inflight + 1)
                 }

glide-core/tests/test_socket_listener.rs

@@ -1069,7 +1069,7 @@ mod socket_listener {
             write_blpop(
                 &mut buffer,
                 &mut write_blpop_socket,
-                i as u32,
+                i,
                 "nonexistingkeylist",
                 0,
             );

python/python/glide/config.py

@@ -389,7 +389,7 @@ class GlideClusterClientConfiguration(BaseClientConfiguration):
             This limit is used to control the memory usage and prevent the client from overwhelming the server or getting stuck in case of a queue backlog.
             If not set, a default value will be used.
 
-            
+
 
     Notes:
         Currently, the reconnection strategy in cluster mode is not configurable, and exponential backoff

python/python/tests/conftest.py

@@ -246,7 +246,7 @@ async def create_client(
             protocol=protocol,
             request_timeout=timeout,
             pubsub_subscriptions=cluster_mode_pubsub,
-            inflight_requests_limit=inflight_requests_limit
+            inflight_requests_limit=inflight_requests_limit,
         )
         return await GlideClusterClient.create(cluster_config)
     else:

python/python/tests/test_async_client.py

@@ -311,7 +311,7 @@ async def test_allow_opt_in_to_resp2_protocol(self, glide_client: TGlideClient):
         assert int(result[b"proto"]) == 2
 
     # Testing the inflight_requests_limit parameter in glide. Sending the allowed amount + 1 of requests
-    # to glide, using blocking commands, and checking the N+1 request returns immediately with error. 
+    # to glide, using blocking commands, and checking the N+1 request returns immediately with error.
     @pytest.mark.parametrize("cluster_mode", [False, True])
     @pytest.mark.parametrize("protocol", [ProtocolVersion.RESP2, ProtocolVersion.RESP3])
     @pytest.mark.parametrize("inflight_requests_limit", [5, 100, 1500])
@@ -320,27 +320,30 @@ async def test_inflight_request_limit(
     ):
         key1 = f"{{nonexistinglist}}:1-{get_random_string(10)}"
         test_client = await create_client(
-            request=request, protocol=protocol, cluster_mode=cluster_mode, inflight_requests_limit=inflight_requests_limit
+            request=request,
+            protocol=protocol,
+            cluster_mode=cluster_mode,
+            inflight_requests_limit=inflight_requests_limit,
         )
-        
+
         tasks = []
         for i in range(inflight_requests_limit + 1):
-            tasks.append(test_client.blpop([key1], 0))
+            coro = test_client.blpop([key1], 0)
+            task = asyncio.create_task(coro)
+            tasks.append(task)
 
-        done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
+        done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
 
         for task in done:
-            try:
-                result = await task
-                print(f"Completed task: {result}")
-            except Exception as e:
-                assert isinstance(e, RequestError)
-                print(f"Error in task: {e}")
-                break
+            with pytest.raises(RequestError) as e:
+                await task
+            assert "maximum inflight requests" in str(e)
 
         for task in pending:
             task.cancel()
 
+        await test_client.close()
+
     @pytest.mark.parametrize("cluster_mode", [True, False])
     @pytest.mark.parametrize("protocol", [ProtocolVersion.RESP2, ProtocolVersion.RESP3])
     async def test_conditional_set(self, glide_client: TGlideClient):