prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <issue>
  [gRPC] Does not work with stream-to-stream gRPC requests
  **Describe the bug**: I'm trying to add elastic apm for my gRPC service with unary-to-unary requests and stream-to-stream. Everything is fine with unary-to-unary requests. But stream-to-stream requests stop working.
  
  The service stops accepting streaming requests altogether.
  
  I think I saw somewhere that you wrote elastic does not support stream requests for gRPC. But I didn't think it would be a service-blocking problem. 
  
  Is there any way to get around it, to make custom processing for stream specifically?
  
  **To Reproduce**
  1. Create a gRPC service with stream-to-stream and unary-to-unary endpoints.
  ```python
  class TestService(test_pb2_grpc.TestService):
      async def Streaming(self, request_iterator, context):
             count_of_request = 0
             async for request in request_iterator:
                 count_of_request += 1
                 yield test_pb2.Response(message=f"#{count_of_request } - ok")
  
      async def Unary(self, request, context):
             return test_pb2.Response(message="ok")
  ```
  3. Connect elastic apm.
  ```python
  elasticapm.instrument()
  
  async def run_serve():
      apm_client = GRPCApmClient(
              {
                  "SERVICE_NAME": "grpc-test",
                  "SERVER_URL": "http://localhost:8200",
                  "ENVIRONMENT": "test",
                  "TRANSACTION_SAMPLE_RATE": 1.0,
                  "SECRET_TOKEN": "qwerty",
              }
      )
      server = grpc.aio.server(
          futures.ThreadPoolExecutor(max_workers=10),
      )
      test_pb2_grpc.add_TestServicer_to_server(
          TestService(), server
      )
      server.add_insecure_port("[::]:50051")
      await server.start()
      await server.wait_for_termination()
  
  
  if __name__ == "__main__":
      asyncio.run(run_serve())
  ```
  5. Make test requests
  6. Result: unary - works, stream - returns nothing, no logs in the service
  
  **Environment (please complete the following information)**
  - OS: Linux
  - Python version: 3.10
  - Framework and version: gRPC 1.43.0
  - APM Server version: 6.20.0
  
  **Additional context**
  
  Add any other context about the problem here.
  
  - `requirements.txt`:
    <details>
      <summary>Click to expand</summary>
  
      ```
      gcloud==0.18.3
      google-api-python-client==2.39.0
      grpcio-tools==1.43.0
      grpcio-health-checking==1.43.0
      setuptools==59.5.0
      elastic-apm
      sentry-sdk[grpcio]==1.31.0
      ```
    </details>
  
  
  </issue>
  <code>
  [start of elasticapm/contrib/grpc/async_server_interceptor.py]
  1 #  BSD 3-Clause License
  2 #
  3 #  Copyright (c) 2022, Elasticsearch BV
  4 #  All rights reserved.
  5 #
  6 #  Redistribution and use in source and binary forms, with or without
  7 #  modification, are permitted provided that the following conditions are met:
  8 #
  9 #  * Redistributions of source code must retain the above copyright notice, this
  10 #    list of conditions and the following disclaimer.
  11 #
  12 #  * Redistributions in binary form must reproduce the above copyright notice,
  13 #    this list of conditions and the following disclaimer in the documentation
  14 #    and/or other materials provided with the distribution.
  15 #
  16 #  * Neither the name of the copyright holder nor the names of its
  17 #    contributors may be used to endorse or promote products derived from
  18 #    this software without specific prior written permission.
  19 #
  20 #  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  21 #  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  22 #  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  23 #  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
  24 #  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  25 #  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
  26 #  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
  27 #  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  28 #  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  29 #  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  30 
  31 import inspect
  32 
  33 import grpc
  34 
  35 import elasticapm
  36 from elasticapm.contrib.grpc.server_interceptor import _ServicerContextWrapper, _wrap_rpc_behavior, get_trace_parent
  37 
  38 
  39 class _AsyncServerInterceptor(grpc.aio.ServerInterceptor):
  40     async def intercept_service(self, continuation, handler_call_details):
  41         def transaction_wrapper(behavior, request_streaming, response_streaming):
  42             async def _interceptor(request_or_iterator, context):
  43                 if request_streaming or response_streaming:  # only unary-unary is supported
  44                     return behavior(request_or_iterator, context)
  45                 tp = get_trace_parent(handler_call_details)
  46                 client = elasticapm.get_client()
  47                 transaction = client.begin_transaction("request", trace_parent=tp)
  48                 try:
  49                     result = behavior(request_or_iterator, _ServicerContextWrapper(context, transaction))
  50 
  51                     # This is so we can support both sync and async rpc functions
  52                     if inspect.isawaitable(result):
  53                         result = await result
  54 
  55                     if transaction and not transaction.outcome:
  56                         transaction.set_success()
  57                     return result
  58                 except Exception:
  59                     if transaction:
  60                         transaction.set_failure()
  61                     client.capture_exception(handled=False)
  62                     raise
  63                 finally:
  64                     client.end_transaction(name=handler_call_details.method)
  65 
  66             return _interceptor
  67 
  68         return _wrap_rpc_behavior(await continuation(handler_call_details), transaction_wrapper)
  69
  [end of elasticapm/contrib/grpc/async_server_interceptor.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: diff --git a/elasticapm/contrib/grpc/async_server_interceptor.py b/elasticapm/contrib/grpc/async_server_interceptor.py
  --- a/elasticapm/contrib/grpc/async_server_interceptor.py
  +++ b/elasticapm/contrib/grpc/async_server_interceptor.py
  @@ -33,20 +33,18 @@
   import grpc
   
   import elasticapm
  -from elasticapm.contrib.grpc.server_interceptor import _ServicerContextWrapper, _wrap_rpc_behavior, get_trace_parent
  +from elasticapm.contrib.grpc.server_interceptor import _ServicerContextWrapper, get_trace_parent
   
   
   class _AsyncServerInterceptor(grpc.aio.ServerInterceptor):
       async def intercept_service(self, continuation, handler_call_details):
  -        def transaction_wrapper(behavior, request_streaming, response_streaming):
  -            async def _interceptor(request_or_iterator, context):
  -                if request_streaming or response_streaming:  # only unary-unary is supported
  -                    return behavior(request_or_iterator, context)
  +        def wrap_unary_unary(behavior):
  +            async def _interceptor(request, context):
                   tp = get_trace_parent(handler_call_details)
                   client = elasticapm.get_client()
                   transaction = client.begin_transaction("request", trace_parent=tp)
                   try:
  -                    result = behavior(request_or_iterator, _ServicerContextWrapper(context, transaction))
  +                    result = behavior(request, _ServicerContextWrapper(context, transaction))
   
                       # This is so we can support both sync and async rpc functions
                       if inspect.isawaitable(result):
  @@ -65,4 +63,12 @@
   
               return _interceptor
   
  -        return _wrap_rpc_behavior(await continuation(handler_call_details), transaction_wrapper)
  +        handler = await continuation(handler_call_details)
  +        if handler.request_streaming or handler.response_streaming:
  +            return handler
  +
  +        return grpc.unary_unary_rpc_method_handler(
  +            wrap_unary_unary(handler.unary_unary),
  +            request_deserializer=handler.request_deserializer,
  +            response_serializer=handler.response_serializer,
  +        )
  
verification_info: {"golden_diff": "diff --git a/elasticapm/contrib/grpc/async_server_interceptor.py b/elasticapm/contrib/grpc/async_server_interceptor.py\n--- a/elasticapm/contrib/grpc/async_server_interceptor.py\n+++ b/elasticapm/contrib/grpc/async_server_interceptor.py\n@@ -33,20 +33,18 @@\n import grpc\n \n import elasticapm\n-from elasticapm.contrib.grpc.server_interceptor import _ServicerContextWrapper, _wrap_rpc_behavior, get_trace_parent\n+from elasticapm.contrib.grpc.server_interceptor import _ServicerContextWrapper, get_trace_parent\n \n \n class _AsyncServerInterceptor(grpc.aio.ServerInterceptor):\n     async def intercept_service(self, continuation, handler_call_details):\n-        def transaction_wrapper(behavior, request_streaming, response_streaming):\n-            async def _interceptor(request_or_iterator, context):\n-                if request_streaming or response_streaming:  # only unary-unary is supported\n-                    return behavior(request_or_iterator, context)\n+        def wrap_unary_unary(behavior):\n+            async def _interceptor(request, context):\n                 tp = get_trace_parent(handler_call_details)\n                 client = elasticapm.get_client()\n                 transaction = client.begin_transaction(\"request\", trace_parent=tp)\n                 try:\n-                    result = behavior(request_or_iterator, _ServicerContextWrapper(context, transaction))\n+                    result = behavior(request, _ServicerContextWrapper(context, transaction))\n \n                     # This is so we can support both sync and async rpc functions\n                     if inspect.isawaitable(result):\n@@ -65,4 +63,12 @@\n \n             return _interceptor\n \n-        return _wrap_rpc_behavior(await continuation(handler_call_details), transaction_wrapper)\n+        handler = await continuation(handler_call_details)\n+        if handler.request_streaming or handler.response_streaming:\n+            return handler\n+\n+        return grpc.unary_unary_rpc_method_handler(\n+            wrap_unary_unary(handler.unary_unary),\n+            request_deserializer=handler.request_deserializer,\n+            response_serializer=handler.response_serializer,\n+        )\n", "issue": "[gRPC] Does not work with stream-to-stream gRPC requests\n**Describe the bug**: I'm trying to add elastic apm for my gRPC service with unary-to-unary requests and stream-to-stream. Everything is fine with unary-to-unary requests. But stream-to-stream requests stop working.\r\n\r\nThe service stops accepting streaming requests altogether.\r\n\r\nI think I saw somewhere that you wrote elastic does not support stream requests for gRPC. But I didn't think it would be a service-blocking problem. \r\n\r\nIs there any way to get around it, to make custom processing for stream specifically?\r\n\r\n**To Reproduce**\r\n1. Create a gRPC service with stream-to-stream and unary-to-unary endpoints.\r\n```python\r\nclass TestService(test_pb2_grpc.TestService):\r\n    async def Streaming(self, request_iterator, context):\r\n           count_of_request = 0\r\n           async for request in request_iterator:\r\n               count_of_request += 1\r\n               yield test_pb2.Response(message=f\"#{count_of_request } - ok\")\r\n\r\n    async def Unary(self, request, context):\r\n           return test_pb2.Response(message=\"ok\")\r\n```\r\n3. Connect elastic apm.\r\n```python\r\nelasticapm.instrument()\r\n\r\nasync def run_serve():\r\n    apm_client = GRPCApmClient(\r\n            {\r\n                \"SERVICE_NAME\": \"grpc-test\",\r\n                \"SERVER_URL\": \"http://localhost:8200\",\r\n                \"ENVIRONMENT\": \"test\",\r\n                \"TRANSACTION_SAMPLE_RATE\": 1.0,\r\n                \"SECRET_TOKEN\": \"qwerty\",\r\n            }\r\n    )\r\n    server = grpc.aio.server(\r\n        futures.ThreadPoolExecutor(max_workers=10),\r\n    )\r\n    test_pb2_grpc.add_TestServicer_to_server(\r\n        TestService(), server\r\n    )\r\n    server.add_insecure_port(\"[::]:50051\")\r\n    await server.start()\r\n    await server.wait_for_termination()\r\n\r\n\r\nif __name__ == \"__main__\":\r\n    asyncio.run(run_serve())\r\n```\r\n5. Make test requests\r\n6. Result: unary - works, stream - returns nothing, no logs in the service\r\n\r\n**Environment (please complete the following information)**\r\n- OS: Linux\r\n- Python version: 3.10\r\n- Framework and version: gRPC 1.43.0\r\n- APM Server version: 6.20.0\r\n\r\n**Additional context**\r\n\r\nAdd any other context about the problem here.\r\n\r\n- `requirements.txt`:\r\n  <details>\r\n    <summary>Click to expand</summary>\r\n\r\n    ```\r\n    gcloud==0.18.3\r\n    google-api-python-client==2.39.0\r\n    grpcio-tools==1.43.0\r\n    grpcio-health-checking==1.43.0\r\n    setuptools==59.5.0\r\n    elastic-apm\r\n    sentry-sdk[grpcio]==1.31.0\r\n    ```\r\n  </details>\r\n\n", "before_files": [{"content": "#  BSD 3-Clause License\n#\n#  Copyright (c) 2022, Elasticsearch BV\n#  All rights reserved.\n#\n#  Redistribution and use in source and binary forms, with or without\n#  modification, are permitted provided that the following conditions are met:\n#\n#  * Redistributions of source code must retain the above copyright notice, this\n#    list of conditions and the following disclaimer.\n#\n#  * Redistributions in binary form must reproduce the above copyright notice,\n#    this list of conditions and the following disclaimer in the documentation\n#    and/or other materials provided with the distribution.\n#\n#  * Neither the name of the copyright holder nor the names of its\n#    contributors may be used to endorse or promote products derived from\n#    this software without specific prior written permission.\n#\n#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE\n#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE\n#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL\n#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR\n#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER\n#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,\n#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n#  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n\nimport inspect\n\nimport grpc\n\nimport elasticapm\nfrom elasticapm.contrib.grpc.server_interceptor import _ServicerContextWrapper, _wrap_rpc_behavior, get_trace_parent\n\n\nclass _AsyncServerInterceptor(grpc.aio.ServerInterceptor):\n    async def intercept_service(self, continuation, handler_call_details):\n        def transaction_wrapper(behavior, request_streaming, response_streaming):\n            async def _interceptor(request_or_iterator, context):\n                if request_streaming or response_streaming:  # only unary-unary is supported\n                    return behavior(request_or_iterator, context)\n                tp = get_trace_parent(handler_call_details)\n                client = elasticapm.get_client()\n                transaction = client.begin_transaction(\"request\", trace_parent=tp)\n                try:\n                    result = behavior(request_or_iterator, _ServicerContextWrapper(context, transaction))\n\n                    # This is so we can support both sync and async rpc functions\n                    if inspect.isawaitable(result):\n                        result = await result\n\n                    if transaction and not transaction.outcome:\n                        transaction.set_success()\n                    return result\n                except Exception:\n                    if transaction:\n                        transaction.set_failure()\n                    client.capture_exception(handled=False)\n                    raise\n                finally:\n                    client.end_transaction(name=handler_call_details.method)\n\n            return _interceptor\n\n        return _wrap_rpc_behavior(await continuation(handler_call_details), transaction_wrapper)\n", "path": "elasticapm/contrib/grpc/async_server_interceptor.py"}]}