async def METHOD_NAME(self):
    """Wraps the unary rpc call with retry loop based on the retry params.
    :yields: Responses received from the target.
    """
    stub = jina_pb2_grpc.JinaSingleDataRequestRPCStub(self.channel)
    def _request_handler(
        request: 'Request',
    ) -> 'Tuple[asyncio.Future, Optional[asyncio.Future]]':
        async def _with_retry(req: 'Request'):
            for attempt in range(1, self.max_attempts + 1):
                try:
                    return await stub.process_single_data(
                        req,
                        compression=self.compression,
                        metadata=self.metadata,
                        credentials=self.kwargs.get('credentials', None),
                        timeout=self.kwargs.get('timeout', None),
                    )
                except (
                    grpc.aio.AioRpcError,
                    InternalNetworkError,
                ) as err:
                    await wait_or_raise_err(
                        attempt=attempt,
                        err=err,
                        max_attempts=self.max_attempts,
                        backoff_multiplier=self.backoff_multiplier,
                        initial_backoff=self.initial_backoff,
                        max_backoff=self.max_backoff,
                    )
        return (
            asyncio.ensure_future(_with_retry(request)),
            None,
        )
    def _result_handler(resp):
        callback_exec(
            response=resp,
            logger=self.logger,
            on_error=self.on_error,
            on_done=self.on_done,
            on_always=self.on_always,
            continue_on_error=self.continue_on_error,
        )
        return resp
    streamer_args = vars(self.client_args)
    if self.prefetch:
        streamer_args['prefetch'] = self.prefetch
    streamer = RequestStreamer(
        request_handler=_request_handler,
        result_handler=_result_handler,
        iterate_sync_in_thread=False,
        logger=self.logger,
        **streamer_args,
    )
    async for response in streamer.stream(
        request_iterator=self.req_iter, results_in_order=self.results_in_order
    ):
        if self.show_progress:
            self.p_bar.update()
        yield response