code: def test_multiple_clients(
      prefetch,
      protocol,
      monkeypatch,
      simple_graph_dict_indexer,
      use_stream,
      prefetch_implementation,
  ):
      if not use_stream and protocol != "grpc":
          return
  
      GOOD_CLIENTS = 5
      GOOD_CLIENT_NUM_DOCS = 20
      MALICIOUS_CLIENT_NUM_DOCS = 50
  
      port = random_port()
      gateway_kwargs = {
          "protocol": protocol,
          "port": port,
          "graph_dict": simple_graph_dict_indexer,
      }
      client_prefetch = None
  
      if prefetch_implementation == "gateway":
          gateway_kwargs["prefetch"] = prefetch
      else:
          client_prefetch = prefetch
  
      def get_document(i):
          return Document(
              id=f"{multiprocessing.current_process().name}_{i}",
              text=str(bytes(bytearray(os.urandom(512 * 4)))),
          )
  
      async def good_client_gen():
          for i in range(GOOD_CLIENT_NUM_DOCS):
              yield get_document(i)
              await asyncio.sleep(0.1)
  
      async def malicious_client_gen():
          for i in range(1000, 1000 + MALICIOUS_CLIENT_NUM_DOCS):
              yield get_document(i)
  
      def client(gen, port):
          Client(protocol=protocol, port=port).post(
              on="/index",
              inputs=gen,
              request_size=1,
              stream=use_stream,
              prefetch=client_prefetch,
          )
  
      monkeypatch.setattr(
          networking.GrpcConnectionPool,
          "send_requests_once",
          DummyMockConnectionPool.send_requests_once,
      )
      monkeypatch.setattr(
          networking.GrpcConnectionPool,
          "send_discover_endpoint",
          DummyMockConnectionPool.send_discover_endpoint,
      )
  
      pool = []
  
      runtime_process = multiprocessing.Process(
          target=create_runtime,
          kwargs=gateway_kwargs,
      )
      runtime_process.start()
      time.sleep(1.0)
      # We have 5 good clients connecting to the same gateway. They have controlled requests.
      # Each client sends `GOOD_CLIENT_NUM_DOCS` (20) requests and sleeps after each request.
      for i in range(GOOD_CLIENTS):
          cp = multiprocessing.Process(
              target=partial(client, good_client_gen, port),
              name=f"goodguy_{i}",
          )
          cp.start()
          pool.append(cp)
  
      # and 1 malicious client, sending lot of requests (trying to block others)
      cp = multiprocessing.Process(
          target=partial(client, malicious_client_gen, port),
          name="badguy",
      )
      cp.start()
      pool.append(cp)
  
      for p in pool:
          p.join()
  
      order_of_ids = list(
          Client(protocol=protocol, port=port)
          .post(on="/status", inputs=[Document()], stream=use_stream)[0]
          .tags["ids"]
      )
      # There must be total 150 docs indexed.
  
      runtime_process.terminate()
      runtime_process.join()
      assert (
          len(order_of_ids)
          == GOOD_CLIENTS * GOOD_CLIENT_NUM_DOCS
          + MALICIOUS_CLIENT_NUM_DOCS
      )
      (
          malicious_client_first_response_idx,
          malicious_client_last_response_idx,
      ) = _search_first_and_last_prefix_occurrence(
          "badguy", order_of_ids
      )
  
      """
      If prefetch is disabled, clients can freeflow requests. No client is blocked.
      Hence last 20 requests go from `goodguy`.
      (Ideally last 30 requests should be validated, to avoid flaky CI, we test last 20)
  
      When there are no rules, badguy wins! With rule, you find balance in the world.
      """
      # first response will always be from one of the good clients
      assert order_of_ids[0].split("_")[0] == "goodguy"
      # first response from malicilious client will (mostly) appear after the first response from a good client
      assert malicious_client_first_response_idx >= 0
      if prefetch == 0:
          # mailicious client will finish before all the good clients
          assert malicious_client_last_response_idx < (
              len(order_of_ids) - 1
          )
  
      if prefetch == 5:
          if prefetch_implementation == "gateway" and not use_stream:
              # gateway with prefetch will complete the full iteration from the gateway recieves
              # the full malicious client request iterator which is processed with prefetch
              assert set(
                  map(lambda x: x.split("_")[0], order_of_ids[-10:])
              ) == {"goodguy"}
              # mailicious client will finish before all the good clients
              assert malicious_client_last_response_idx < (
                  len(order_of_ids) - 1
              )
          elif (
              prefetch_implementation == "gateway"
              and protocol == "http"
          ):
              # gateway with prefetch will complete the full iteration from the gateway recieves
              # the full malicious client request iterator which is processed with prefetch
              # and due to the request <-> response coupling
              assert set(
                  map(lambda x: x.split("_")[0], order_of_ids[-10:])
              ) == {"goodguy"}
              # mailicious client will finish before all the good clients
              assert malicious_client_last_response_idx < (
                  len(order_of_ids) - 1
              )
  
          else:
              assert set(
                  map(lambda x: x.split("_")[0], order_of_ids[-10:])
              ) == {"badguy"}
              # mailicious client will finish last
              assert malicious_client_last_response_idx == (
                  len(order_of_ids) - 1
              )
  
imports: import asyncio
  import copy
  import multiprocessing
  import os
  import time
  from datetime import datetime
  from functools import partial
  from typing import Dict, List
  import pytest
  from docarray import Document, DocumentArray
  from jina.clients import Client
  from jina.helper import random_port
  from jina.parsers import set_gateway_parser
  from jina.serve import networking
  from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
  from jina.serve.runtimes.gateway.request_handling import GatewayRequestHandler
label_desc: Use image node_id 13 for calling a global function with example usage: test_multiple_clients(prefetch, protocol, monkeypatch, simple_graph_dict_indexer, use_stream, prefetch_implementation) and returns: Document