Logic around sensor _on_stream_event_in differs between cython and python implementations

[Tim-Thomas]

Checklist

• I have included information about relevant versions
• I have verified that the issue persists when using the master branch of Faust.

I recently added prometheus metrics instrumentation to one of my teams projects that's using faust-streaming. However, many of the per-agent metrics I was expecting to appear did not appear. After extensive debugging, I realized that this is due to agents using yield and sink were not correctly invoking _on_stream_event_in per event.

I dug through the cython and python implementations of the stream handling and discovered that in cython, _on_stream_event_in is being invoked conditionally alongside _on_message_in:
https://github.com/faust-streaming/faust/blob/master/faust/_cython/streams.pyx#L177-L183
but in the python implementation, the _on_stream_event_in is being invoked all the time in this code path:
https://github.com/faust-streaming/faust/blob/master/faust/streams.py#L1182-L1190

These cannot both be correct. I believe the correct behavior is the python one.

Expected behavior

I expect that _on_stream_event_in should be called per stream event in both the cython and python impls.

Actual behavior

_on_stream_event_in is not invoked per stream event when using the cython implementation.