wip: renaming functions based on events.

Author: eternalcuriouslearner

instrumentation-genai/opentelemetry-instrumentation-openai-v2/src/opentelemetry/instrumentation/openai_v2/chat_wrappers.py

@@ -144,10 +144,10 @@ def _set_output_messages(self):
 
         self._self_invocation.output_messages = output_messages
 
-    def _stop_stream(self) -> None:
+    def _on_stream_end(self) -> None:
         self._cleanup()
 
-    def _fail_stream(self, error: BaseException) -> None:
+    def _on_stream_error(self, error: BaseException) -> None:
         self._cleanup(error)
 
     def parse(self):

instrumentation-genai/opentelemetry-instrumentation-openai-v2/tests/test_chat_completions.py

@@ -1115,7 +1115,7 @@ def test_chat_completion_streaming_instrumentation_finalize_errors_swallowed(
         def stop_raises():
             raise RuntimeError("instrumentation failure")
 
-        monkeypatch.setattr(response, "_stop_stream", stop_raises)
+        monkeypatch.setattr(response, "_on_stream_end", stop_raises)
         response.close()
 
     assert span_exporter.get_finished_spans() == ()

util/opentelemetry-util-genai/src/opentelemetry/util/genai/stream.py

@@ -67,8 +67,8 @@ class SyncStreamWrapper(
     Subclass this when wrapping a provider SDK stream that is consumed with
     normal iteration. The subclass should pass the SDK stream to
     ``super().__init__(stream)`` and implement the three telemetry hooks:
-    ``_process_chunk`` for per-chunk state, ``_stop_stream`` for successful
-    finalization, and ``_fail_stream`` for failure finalization.
+    ``_process_chunk`` for per-chunk state, ``_on_stream_end`` for successful
+    finalization, and ``_on_stream_error`` for failure finalization.
 
     Users should consume subclasses as normal streams, for example with
     ``for chunk in wrapper`` or ``with wrapper``. The hook methods are called
@@ -137,13 +137,13 @@ def _finalize_success(self) -> None:
         if self._self_finalized:
             return
         self._self_finalized = True
-        self._stop_stream()
+        self._on_stream_end()
 
     def _finalize_failure(self, error: BaseException) -> None:
         if self._self_finalized:
             return
         self._self_finalized = True
-        self._fail_stream(error)
+        self._on_stream_error(error)
 
     def _safe_finalize_success(self) -> None:
         try:
@@ -168,11 +168,11 @@ def _process_chunk(self, chunk: ChunkT) -> None:
         """Process one stream chunk for telemetry."""
 
     @abstractmethod
-    def _stop_stream(self) -> None:
+    def _on_stream_end(self) -> None:
         """Finalize the stream successfully."""
 
     @abstractmethod
-    def _fail_stream(self, error: BaseException) -> None:
+    def _on_stream_error(self, error: BaseException) -> None:
         """Finalize the stream with failure."""
 
     @staticmethod
@@ -193,8 +193,8 @@ class AsyncStreamWrapper(
     Subclass this when wrapping a provider SDK stream that is consumed with
     async iteration. The subclass should pass the SDK stream to
     ``super().__init__(stream)`` and implement the three telemetry hooks:
-    ``_process_chunk`` for per-chunk state, ``_stop_stream`` for successful
-    finalization, and ``_fail_stream`` for failure finalization.
+    ``_process_chunk`` for per-chunk state, ``_on_stream_end`` for successful
+    finalization, and ``_on_stream_error`` for failure finalization.
 
     Users should consume subclasses as normal async streams, for example with
     ``async for chunk in wrapper`` or ``async with wrapper``. The hook methods
@@ -266,13 +266,13 @@ def _finalize_success(self) -> None:
         if self._self_finalized:
             return
         self._self_finalized = True
-        self._stop_stream()
+        self._on_stream_end()
 
     def _finalize_failure(self, error: BaseException) -> None:
         if self._self_finalized:
             return
         self._self_finalized = True
-        self._fail_stream(error)
+        self._on_stream_error(error)
 
     def _safe_finalize_success(self) -> None:
         try:
@@ -297,11 +297,11 @@ def _process_chunk(self, chunk: ChunkT) -> None:
         """Process one stream chunk for telemetry."""
 
     @abstractmethod
-    def _stop_stream(self) -> None:
+    def _on_stream_end(self) -> None:
         """Finalize the stream successfully."""
 
     @abstractmethod
-    def _fail_stream(self, error: BaseException) -> None:
+    def _on_stream_error(self, error: BaseException) -> None:
         """Finalize the stream with failure."""
 
     @staticmethod

util/opentelemetry-util-genai/tests/test_stream.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+# pylint: disable=abstract-class-instantiated
+
 import asyncio
 import inspect
 
@@ -26,8 +28,8 @@
 def test_stream_wrapper_abstract_method_signatures_match():
     method_names = (
         "_process_chunk",
-        "_stop_stream",
-        "_fail_stream",
+        "_on_stream_end",
+        "_on_stream_error",
         "_handle_process_chunk_error",
     )
 
@@ -86,10 +88,10 @@ def __init__(self, stream):
     def _process_chunk(self, chunk):
         self._self_processed.append(chunk)
 
-    def _stop_stream(self):
+    def _on_stream_end(self):
         self._self_stop_count += 1
 
-    def _fail_stream(self, error):
+    def _on_stream_error(self, error):
         self._self_failures.append(error)
 
 
@@ -99,13 +101,13 @@ def _process_chunk(self, chunk):
 
 
 class _FailingSyncStopStreamWrapper(_TestSyncStreamWrapper):
-    def _stop_stream(self):
+    def _on_stream_end(self):
         self._self_stop_count += 1
         raise ValueError("instrumentation failed")
 
 
 class _FailingSyncFailStreamWrapper(_TestSyncStreamWrapper):
-    def _fail_stream(self, error):
+    def _on_stream_error(self, error):
         self._self_failures.append(error)
         raise ValueError("instrumentation failed")
 
@@ -325,10 +327,10 @@ def __init__(self, stream):
     def _process_chunk(self, chunk):
         self._self_processed.append(chunk)
 
-    def _stop_stream(self):
+    def _on_stream_end(self):
         self._self_stop_count += 1
 
-    def _fail_stream(self, error):
+    def _on_stream_error(self, error):
         self._self_failures.append(error)
 
 
@@ -338,13 +340,13 @@ def _process_chunk(self, chunk):
 
 
 class _FailingAsyncStopStreamWrapper(_TestAsyncStreamWrapper):
-    def _stop_stream(self):
+    def _on_stream_end(self):
         self._self_stop_count += 1
         raise ValueError("instrumentation failed")
 
 
 class _FailingAsyncFailStreamWrapper(_TestAsyncStreamWrapper):
-    def _fail_stream(self, error):
+    def _on_stream_error(self, error):
         self._self_failures.append(error)
         raise ValueError("instrumentation failed")