class_docstr: Helper class that wraps the RPC server that middlemans the connection
  between the client and the real ray server. Useful for injecting
  errors between a client and server pair.
class_docstr_tok: ["Helper","class","that","wraps","the","RPC","server","that","middlemans","the","connection","between","the","client","and","the","real","ray","server",".","Useful","for","injecting","errors","between","a","client","and","server","pair","."]
docstr: Args:
      listen_addr: The address the middleman server will listen on
      real_addr: The address of the real ray server
      on_log_response: Optional hook to inject errors before sending back
          a log response
      on_data_response: Optional hook to inject errors before sending
          back a data response
      on_task_request: Optional hook to inject errors before forwarding
          a raylet driver request
      on_task_response: Optional hook to inject errors before sending
          back a raylet driver response
docstr_tok: ["Args",":","listen_addr",":","The","address","the","middleman","server","will","listen","on","real_addr",":","The","address","of","the","real","ray","server","on_log_response",":","Optional","hook","to","inject","errors","before","sending","back","a","log","response","on_data_response",":","Optional","hook","to","inject","errors","before","sending","back","a","data","response","on_task_request",":","Optional","hook","to","inject","errors","before","forwarding","a","raylet","driver","request","on_task_response",":","Optional","hook","to","inject","errors","before","sending","back","a","raylet","driver","response"]
code: def __init__(
      self,
      listen_addr: str,
      real_addr,
      on_log_response: Optional[Hook] = None,
      on_data_request: Optional[Hook] = None,
      on_data_response: Optional[Hook] = None,
      on_task_request: Optional[Hook] = None,
      on_task_response: Optional[Hook] = None,
  ):
      """
      Args:
          listen_addr: The address the middleman server will listen on
          real_addr: The address of the real ray server
          on_log_response: Optional hook to inject errors before sending back
              a log response
          on_data_response: Optional hook to inject errors before sending
              back a data response
          on_task_request: Optional hook to inject errors before forwarding
              a raylet driver request
          on_task_response: Optional hook to inject errors before sending
              back a raylet driver response
      """
      self.listen_addr = listen_addr
      self.real_addr = real_addr
      self.server = grpc.server(
          futures.ThreadPoolExecutor(
              max_workers=CLIENT_SERVER_MAX_THREADS
          ),
          options=GRPC_OPTIONS,
      )
      self.task_servicer = MiddlemanRayletServicer(
          on_response=on_task_response, on_request=on_task_request
      )
      self.data_servicer = MiddlemanDataServicer(
          on_response=on_data_response, on_request=on_data_request
      )
      self.logs_servicer = MiddlemanLogServicer(
          on_response=on_log_response
      )
      ray_client_pb2_grpc.add_RayletDriverServicer_to_server(
          self.task_servicer, self.server
      )
      ray_client_pb2_grpc.add_RayletDataStreamerServicer_to_server(
          self.data_servicer, self.server
      )
      ray_client_pb2_grpc.add_RayletLogStreamerServicer_to_server(
          self.logs_servicer, self.server
      )
      self.server.add_insecure_port(self.listen_addr)
      self.channel = None
      self.reset_channel()
  
imports: from concurrent import futures
  import contextlib
  import os
  import threading
  import sys
  import grpc
  from mock import Mock
  import numpy
  import time
  import random
  import pytest
  from typing import Any, Callable, Optional
  from unittest.mock import patch
  import ray
  from ray._private.utils import get_or_create_event_loop
  import ray.core.generated.ray_client_pb2
  import ray.core.generated.ray_client_pb2_grpc
  from ray.tests.conftest import call_ray_start_context
  from ray.util.client.common import CLIENT_SERVER_MAX_THREADS, GRPC_OPTIONS
label_desc: Use image node_id 1 to create a new MiddlemanServer object  with example: obj = MiddlemanServer(listen_addr, real_addr, on_log_response, on_data_request, on_data_response, on_task_request, on_task_response)