Remove _message_iterator since not used by obj store anymore

Signed-off-by: Waldemar Quevedo <[email protected]>

Author: wallyqs

nats/src/nats/aio/subscription.py

@@ -94,8 +94,6 @@ def __init__(
             self._pending_next_msgs_calls = None
         self._pending_size = 0
         self._wait_for_msgs_task = None
-        # For compatibility with tests that expect _message_iterator
-        self._message_iterator = None
 
         # For JetStream enabled subscriptions.
         self._jsi: Optional[JetStreamContext._JSI] = None
@@ -147,14 +145,6 @@ async def _message_generator(self) -> AsyncIterator[Msg]:
                 if self._closed:
                     break
 
-                # Check if wrapper was cancelled (for compatibility with tests).
-                if (
-                    hasattr(self, "_message_iterator")
-                    and self._message_iterator
-                    and self._message_iterator._unsubscribed_future.done()
-                ):
-                    break
-
                 # Check max message limit based on how many we've yielded so far.
                 if self._max_msgs > 0 and yielded_count >= self._max_msgs:
                     break
@@ -177,9 +167,6 @@ async def _message_generator(self) -> AsyncIterator[Msg]:
 
                 # Check if we should auto-unsubscribe after yielding this message.
                 if self._max_msgs > 0 and yielded_count >= self._max_msgs:
-                    # Cancel the wrapper too for consistency.
-                    if hasattr(self, "_message_iterator") and self._message_iterator:
-                        self._message_iterator._cancel()
                     break
         except asyncio.CancelledError:
             pass
@@ -271,8 +258,7 @@ def _start(self, error_cb):
             pass
         else:
             # For async iteration, we now use a generator directly via the messages property
-            # But we create a compatibility wrapper for tests
-            self._message_iterator = _CompatibilityIteratorWrapper(self)
+            pass
 
     async def drain(self):
         """
@@ -343,8 +329,6 @@ def _stop_processing(self) -> None:
         """
         if self._wait_for_msgs_task and not self._wait_for_msgs_task.done():
             self._wait_for_msgs_task.cancel()
-        if hasattr(self, "_message_iterator") and self._message_iterator:
-            self._message_iterator._cancel()
 
         # Only put sentinel if there are active async generators
         try:
@@ -394,18 +378,3 @@ async def _wait_for_msgs(self, error_cb) -> None:
                     self._stop_processing()
             except asyncio.CancelledError:
                 break
-
-
-class _CompatibilityIteratorWrapper:
-    """
-    Compatibility wrapper that provides the same interface as the old _SubscriptionMessageIterator
-    but uses the more efficient generator internally.
-    """
-
-    def __init__(self, sub: Subscription) -> None:
-        self._sub = sub
-        self._unsubscribed_future: asyncio.Future[bool] = asyncio.Future()
-
-    def _cancel(self) -> None:
-        if not self._unsubscribed_future.done():
-            self._unsubscribed_future.set_result(True)

nats/tests/test_client.py

@@ -567,8 +567,8 @@ async def iterator_func(sub):
         self.assertEqual(0, sub.pending_bytes)
         await nc.close()
 
-        # Confirm that iterator is done.
-        self.assertTrue(sub._message_iterator._unsubscribed_future.done())
+        # Confirm that subscription is closed.
+        self.assertTrue(sub._closed)
 
     @async_test
     async def test_subscribe_async_generator(self):