refactor: process all streaming events for forward compatibility

Changes based on review feedback from @karpetrosyan:

1. Removed MESSAGE_EVENTS filtering - now processes any event received
2. Removed early response closing (response.close/aclose) - will be ported to codegen
3. Removed performance test file - no longer applicable

This change improves forward compatibility by processing any event type
the API sends, rather than filtering to a predefined set. The new
structure handles completion, ping, and error events explicitly, while
processing all other events generically.

Closes review feedback in anthropics#993

Author: fede-kamel

src/anthropic/_streaming.py

@@ -19,18 +19,6 @@
 
 _T = TypeVar("_T")
 
-# Centralized event type definitions for O(1) lookup performance
-# frozenset prevents accidental modification and is slightly faster than set
-MESSAGE_EVENTS = frozenset(
-    {
-        "message_start",
-        "message_delta",
-        "message_stop",
-        "content_block_start",
-        "content_block_delta",
-        "content_block_stop",
-    }
-)
 
 
 class _SyncStreamMeta(abc.ABCMeta):
@@ -93,21 +81,13 @@ def __stream__(self) -> Iterator[_T]:
         for sse in iterator:
             if sse.event == "completion":
                 yield process_data(data=sse.json(), cast_to=cast_to, response=response)
-            # Single, fast membership test instead of multiple string comparisons
-            if sse.event in MESSAGE_EVENTS:
-                data = sse.json()
-                if is_dict(data) and "type" not in data:
-                    data["type"] = sse.event
-
-                yield process_data(data=data, cast_to=cast_to, response=response)
 
             if sse.event == "ping":
                 continue
 
             if sse.event == "error":
                 body = sse.data
 
-                # Extract meaningful error messages and use specific exception handling
                 try:
                     body = sse.json()
                     err_msg = f"{body}"
@@ -120,9 +100,12 @@ def __stream__(self) -> Iterator[_T]:
                     response=self.response,
                 )
 
-        # Explicitly close the response to release the connection
-        # Immediately releases connection instead of consuming remaining stream
-        self.response.close()
+            # Process any other event for forward compatibility
+            data = sse.json()
+            if is_dict(data) and "type" not in data:
+                data["type"] = sse.event
+
+            yield process_data(data=data, cast_to=cast_to, response=response)
 
     def __enter__(self) -> Self:
         return self
@@ -205,13 +188,6 @@ async def __stream__(self) -> AsyncIterator[_T]:
         async for sse in iterator:
             if sse.event == "completion":
                 yield process_data(data=sse.json(), cast_to=cast_to, response=response)
-            # Single, fast membership test instead of multiple string comparisons
-            if sse.event in MESSAGE_EVENTS:
-                data = sse.json()
-                if is_dict(data) and "type" not in data:
-                    data["type"] = sse.event
-
-                yield process_data(data=data, cast_to=cast_to, response=response)
 
             if sse.event == "ping":
                 continue
@@ -231,9 +207,12 @@ async def __stream__(self) -> AsyncIterator[_T]:
                     response=self.response,
                 )
 
-        # Explicitly close the response to release the connection
-        # Immediately releases connection instead of consuming remaining stream
-        await self.response.aclose()
+            # Process any other event for forward compatibility
+            data = sse.json()
+            if is_dict(data) and "type" not in data:
+                data["type"] = sse.event
+
+            yield process_data(data=data, cast_to=cast_to, response=response)
 
     async def __aenter__(self) -> Self:
         return self