Await asynchronous sinks when flushing collected values

[saitakarcesme]

collect.flush() currently drops the awaitables returned by downstream sinks. A synchronous caller can return before the sink runs, and await collector.flush() raises a TypeError.

Use emit() and return its result so flushing follows the stream's existing synchronous/asynchronous execution mode. Clear the captured batch before invoking downstream callbacks, since those callbacks can collect more values while emit() waits; release the captured metadata references in finally.

Fixes #468.

Validation:

• Three regression cases fail on the unchanged source: direct synchronous flush, flush triggered from another stream, and awaited asynchronous flush.
• An additional regression verifies that values collected by an asynchronous sink survive for the next flush (it fails if the cache is cleared after emit()).
• Core, native async, and sink suites: 131 passed, 2 skipped on Python 3.13.
• Full suite: 1038 passed, 438 skipped, 8 xfailed, 96 xpassed; two subprocess tests initially failed because python was not on PATH. Both pass when rerun with the virtualenv bin directory on PATH.
• All tracked non-test Python modules pass the repository flake8 configuration; git diff --check passes.

AI disclosure: This contribution was prepared and tested by OpenAI Codex under the account owner's authorization.

[martindurant]

@saitakarcesme , are you an active user of streamz?