**INTERNAL** Terminal cleanup for the streaming op: end the observability span/meter and release the executor. Invoked from every terminal branch of ``__anext__`` -- including cancellation -- so that ``asyncio.CancelledError`` (a ``BaseException``, not an ``Excepti
(self, exc_val=None)
| 108 | self._tp_executor.shutdown(wait=False) |
| 109 | |
| 110 | def _finalize(self, exc_val=None): |
| 111 | """ |
| 112 | **INTERNAL** |
| 113 | |
| 114 | Terminal cleanup for the streaming op: end the observability span/meter and release the |
| 115 | executor. Invoked from every terminal branch of ``__anext__`` -- including cancellation -- |
| 116 | so that ``asyncio.CancelledError`` (a ``BaseException``, not an ``Exception``) cannot bypass |
| 117 | teardown and leak the span/meter or orphan the streaming result and its executor thread. |
| 118 | """ |
| 119 | self._process_core_span(exc_val=exc_val) |
| 120 | if exc_val is not None and self._streaming_result is not None: |
| 121 | # Cancellation/error: unblock a worker still waiting on the C++ core so its executor |
| 122 | # thread is released promptly instead of waiting for the whole server-side operation |
| 123 | # to finish. Normal completion (exc_val is None) skips this so trailing metadata can |
| 124 | # still be read from the streaming result. |
| 125 | self._streaming_result.cancel() |
| 126 | self._shutdown_executor() |
| 127 | |
| 128 | async def __anext__(self): |
| 129 | return await stream_anext(self) |