code: def _mk_request_handler(
      self,
      runner_method: RunnerMethod[t.Any, t.Any, t.Any],
      batching: bool = True,
  ) -> t.Callable[[Request], t.Coroutine[None, None, Response]]:
      from starlette.responses import Response
  
      server_str = f"BentoML-Runner/{self.runner.name}/{runner_method.name}/{self.worker_index}"
  
      if runner_method.config.is_stream:
          # Streaming does not have batching implemented yet
          async def infer_stream(
              paramss: t.Sequence[Params[t.Any]],
          ) -> t.Sequence[t.AsyncGenerator[str, None]]:
              async def inner():
                  # This is a workaround to allow infer stream to return a iterable of
                  # async generator, to align with how non stream inference works
                  params = paramss[0].map(AutoContainer.from_payload)
                  try:
                      ret = runner_method.async_stream(
                          *params.args, **params.kwargs
                      )
                  except Exception:
                      traceback.print_exc()
                      raise
                  async for data in ret:
                      yield data
  
              return (inner(),)
  
          infer = self.dispatchers[runner_method.name](infer_stream)
      else:
          if batching:
  
              async def infer_batch(
                  params_list: t.Sequence[Params[t.Any]],
              ) -> list[Payload] | list[tuple[Payload, ...]]:
                  self.adaptive_batch_size_hist.labels(  # type: ignore
                      runner_name=self.runner.name,
                      worker_index=self.worker_index,
                      method_name=runner_method.name,
                      service_version=component_context.bento_version,
                      service_name=component_context.bento_name,
                  ).observe(len(params_list))
  
                  if not params_list:
                      return []
  
                  (
                      input_batch_dim,
                      output_batch_dim,
                  ) = runner_method.config.batch_dim
  
                  (
                      batched_params,
                      indices,
                  ) = payload_paramss_to_batch_params(
                      params_list, input_batch_dim
                  )
  
                  try:
                      batch_ret = await runner_method.async_run(
                          *batched_params.args, **batched_params.kwargs
                      )
                  except Exception:
                      traceback.print_exc()
                      raise
  
                  # multiple output branch
                  if LazyType["tuple[t.Any, ...]"](tuple).isinstance(
                      batch_ret
                  ):
                      output_num = len(batch_ret)
                      payloadss = tuple(
                          AutoContainer.batch_to_payloads(
                              batch_ret[idx],
                              indices,
                              batch_dim=output_batch_dim,
                          )
                          for idx in range(output_num)
                      )
                      ret = list(zip(*payloadss))
                      return ret
  
                  # single output branch
                  payloads = AutoContainer.batch_to_payloads(
                      batch_ret,
                      indices,
                      batch_dim=output_batch_dim,
                  )
                  return payloads
  
              infer = self.dispatchers[runner_method.name](infer_batch)
          else:
  
              async def infer_single(
                  paramss: t.Sequence[Params[t.Any]],
              ):
                  assert len(paramss) == 1
  
                  params = paramss[0].map(AutoContainer.from_payload)
  
                  try:
                      ret = await runner_method.async_run(
                          *params.args, **params.kwargs
                      )
                  except Exception:
                      traceback.print_exc()
                      raise
  
                  payload = AutoContainer.to_payload(ret, 0)
                  return (payload,)
  
              infer = self.dispatchers[runner_method.name](infer_single)
  
      async def _request_handler(request: Request) -> Response:
          assert self._is_ready
  
          arg_num = int(request.headers["args-number"])
          r_: bytes = await request.body()
  
          if arg_num == 1:
              params: Params[t.Any] = _deserialize_single_param(
                  request, r_
              )
  
          else:
              params: Params[t.Any] = pickle.loads(r_)
  
          try:
              payload = await infer(params)
          except BentoMLException as e:
              # pass known exceptions to the client
              return Response(
                  status_code=e.error_code,
                  content=str(e),
                  headers={
                      PAYLOAD_META_HEADER: "{}",
                      "Content-Type": "application/vnd.bentoml.error",
                      "Server": server_str,
                  },
              )
  
          if isinstance(payload, ServiceUnavailable):
              return Response(
                  "Service Busy",
                  status_code=payload.error_code,
                  headers={
                      PAYLOAD_META_HEADER: json.dumps({}),
                      "Content-Type": "application/vnd.bentoml.error",
                      "Server": server_str,
                  },
              )
          if isinstance(payload, Payload):
              return Response(
                  payload.data,
                  headers={
                      PAYLOAD_META_HEADER: json.dumps(payload.meta),
                      "Content-Type": f"application/vnd.bentoml.{payload.container}",
                      "Server": server_str,
                  },
              )
          if isinstance(payload, tuple):
              # a tuple, which means user runnable has multiple outputs
              return Response(
                  pickle.dumps(payload),
                  headers={
                      PAYLOAD_META_HEADER: json.dumps({}),
                      "Content-Type": "application/vnd.bentoml.multiple_outputs",
                      "Server": server_str,
                  },
              )
          from starlette.responses import StreamingResponse
  
          if runner_method.config.is_stream:
              return StreamingResponse(
                  stream_encoder(payload),
                  media_type="text/plain",
                  headers={
                      PAYLOAD_META_HEADER: json.dumps({}),
                      "Content-Type": "application/vnd.bentoml.stream_outputs",
                      "Server": server_str,
                  },
              )
  
          raise BentoMLException(
              f"Unexpected payload type: {type(payload)}, {payload}"
          )
  
      return _request_handler
  
imports: from __future__ import annotations
  import functools
  import json
  import logging
  import pickle
  import traceback
  import typing
  from typing import TYPE_CHECKING
  from simple_di import Provide
  from simple_di import inject
  from bentoml.exceptions import BentoMLException
  from bentoml.exceptions import ServiceUnavailable
  from ..configuration.containers import BentoMLContainer
  from ..context import component_context
  from ..context import trace_context
  from ..marshal.dispatcher import CorkDispatcher
  from ..runner.container import AutoContainer
  from ..runner.container import Payload
  from ..runner.utils import PAYLOAD_META_HEADER
  from ..runner.utils import Params
  from ..runner.utils import payload_paramss_to_batch_params
  from ..server.base_app import BaseAppFactory
  from ..types import LazyType
  from ..utils.metrics import exponential_buckets
label_desc: Use image node_id 8 for calling the RunnerAppFactory obj's underlying member method code with example usage: obj._mk_request_handler(runner_method, batching) and returns: _request_handler, inner, Response, Response, Response, StreamingResponse, payloads, payload, Response, list, ret