Ack filtered messages when using .take() (#488)

Co-authored-by: Artem Ilin <a.ilin@arammeem.com>

Author: artem-ilin

faust/_cython/streams.pyx

@@ -109,8 +109,7 @@ cdef class StreamIterator:
             object consumer
         consumer = self.consumer
         last_stream_to_ack = False
-        # if do_ack and event is not None:
-        if event is not None and (do_ack or event.value is self._skipped_value):
+        if do_ack and event is not None:
             message = event.message
             if not message.acked:
                 refcount = message.refcount

faust/streams.py

@@ -1065,7 +1065,7 @@ async def _c_aiter(self) -> AsyncIterator[T_co]:  # pragma: no cover
                         yield value
                 finally:
                     event, self.current_event = self.current_event, None
-                    it.after(event, do_ack, sensor_state)
+                    it.after(event, do_ack or value is skipped_value, sensor_state)
         except StopAsyncIteration:
             # We are not allowed to propagate StopAsyncIteration in __aiter__
             # (if we do, it'll be converted to RuntimeError by CPython).

tests/functional/test_streams.py

@@ -1,6 +1,6 @@
 import asyncio
 from copy import copy
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import pytest
 from mode import label
@@ -236,6 +236,30 @@ async def test_stream_filter_acks_filtered_out_messages(app, event_loop):
     assert len(app.consumer.unacked) == 0
 
 
+@pytest.mark.asyncio
+async def test_acks_filtered_out_messages_when_using_take(app, event_loop):
+    """
+    Test the filter function acknowledges the filtered out messages when using take().
+    """
+    initial_values = [1000, 999, 3000, 99, 5000, 3, 9999]
+    expected_values = [v for v in initial_values if v > 1000]
+    original_function = app.create_event
+    # using patch to intercept message objects, to check if they are acked later
+    with patch("faust.app.base.App.create_event") as create_event_mock:
+        create_event_mock.side_effect = original_function
+        async with new_stream(app) as stream:
+            for value in initial_values:
+                await stream.channel.send(value=value)
+            async for values in stream.filter(lambda x: x > 1000).take(
+                len(expected_values), within=5
+            ):
+                assert values == expected_values
+                break
+    messages = [call[0][3] for call in create_event_mock.call_args_list]
+    acked = [m.acked for m in messages if m.acked]
+    assert len(acked) == len(initial_values)
+
+
 @pytest.mark.asyncio
 async def test_events(app):
     async with new_stream(app) as stream: