Repository: jina-ai/jina
File: jina/serve/streamer.py (Python)

Code before:
        :return: Returns an instance of `GatewayStreamer`
        """
        if 'JINA_STREAMER_ARGS' in os.environ:
            args_dict = json.loads(os.environ['JINA_STREAMER_ARGS'])
            return GatewayStreamer(**args_dict)
        else:
            raise OSError('JINA_STREAMER_ARGS environment variable is not set')

    @staticmethod
    def _set_env_streamer_args(**kwargs):
        os.environ['JINA_STREAMER_ARGS'] = json.dumps(kwargs)


class ExecutorStreamer:
    def __init__(self, connection_pool: GrpcConnectionPool) -> None:
        self._connection_pool: GrpcConnectionPool = connection_pool

    def __getitem__(self, executor_name: str):
        async def _acall_executor(
            docs: DocumentArray,
            request_size: int = 100,
            exec_endpoint: Optional[str] = None,
            parameters: Optional[Dict] = None,
            **kwargs,
        ):
            from jina.clients.request import request_generator
            resp, _ = await self._connection_pool.send_requests_once(
                requests=list(request_generator(
                    exec_endpoint=exec_endpoint,
                    data=docs,
                    request_size=request_size,
                    target_executor=executor_name,
                    parameters=parameters,
                    **kwargs
                )),
                deployment=executor_name,
                head=True,
                endpoint=exec_endpoint
            )
            
            return resp.docs
        
        return _acall_executor
            

Review comment (refactor):
avoid this `request_generator`. Create a Request using the `Request` object. This `generator I would love to remove` since it is too complex

Code after:
        The streamer object is contructed using runtime arguments stored in the `JINA_STREAMER_ARGS` environment variable.
        If this method is used outside a Jina context (process not controlled/orchestrated by jina), this method will
        raise an error.
        The streamer object does not have tracing/instrumentation capabilities.

        :return: Returns an instance of `GatewayStreamer`
        """
        if 'JINA_STREAMER_ARGS' in os.environ:
            args_dict = json.loads(os.environ['JINA_STREAMER_ARGS'])
            return GatewayStreamer(**args_dict)
        else:
            raise OSError('JINA_STREAMER_ARGS environment variable is not set')

    @staticmethod
    def _set_env_streamer_args(**kwargs):
        os.environ['JINA_STREAMER_ARGS'] = json.dumps(kwargs)

    async def warmup(self, stop_event: threading.Event):
        '''Executes warmup task on each deployment. This forces the gateway to establish connection and open a
        gRPC channel to each executor so that the first request doesn't need to experience the penalty of
        eastablishing a brand new gRPC channel.
        :param stop_event: signal to indicate if an early termination of the task is required for graceful teardown.
        '''
        self.logger.debug(f'Running GatewayRuntime warmup')
        deployments = {key for key in self._executor_addresses.keys()}

        try:
            deployment_warmup_tasks = []
            for deployment in deployments:
                deployment_warmup_tasks.append(
                    asyncio.create_task(
                        self._connection_pool.warmup(
                            deployment=deployment, stop_event=stop_event
                        )
                    )
                )

            await asyncio.gather(*deployment_warmup_tasks, return_exceptions=True)
        except Exception as ex:
            self.logger.error(f'error with GatewayRuntime warmup up task: {ex}')
            return


class _ExecutorStreamer:
    def __init__(self, connection_pool: GrpcConnectionPool, executor_name: str) -> None:
        self._connection_pool: GrpcConnectionPool = connection_pool
        self.executor_name = executor_name

    async def post(
        self,
        inputs: DocumentArray,