Add a Receiver.close() method

[shsms]

This method would allow individual receivers to be closed, without affecting the underlying channel, if there is one.

[shsms]

Still need to add tests.

[llucax]

LGTM but the CI is failing, and this reminds me that it would be good to add tests at least for the receivers that actually do something when close() is called.

And oh, yes! As usual names, names, names 😆

TL;DR: All good the name actually, just some other comments about future directions.

I was going to suggest to rename close() to stop(), for once to match the ReceiverStoppedError but also to match the upcoming Service interface.

But I see close actually wait for the current buffer to be consumed, my idea for stop() would be that stuff gets just dropped.

Maybe to make it play better with Service we can keep close() as you implemented it, as a "graceful" shutdown, and when we introduce Service we can use cancel() to drop the current buffers. Then maybe stop() can take an option to decide if close() or cancel() is called (stop is usually implemented as self.cancel(); await self).

[shsms]

LGTM but the CI is failing, and this reminds me that it would be good to add tests at least for the receivers that actually do something when close() is called.

Yes, I'm already working on tests in this same branch, just made an early PR because the code is ready for review.

Maybe to make it play better with Service we can keep close() as you implemented it, as a "graceful" shutdown, and when we introduce Service we can use cancel() to drop the current buffers. Then maybe stop() can take an option to decide if close() or cancel() is called (stop is usually implemented as self.cancel(); await self).

I think stop() doesn't make sense for receivers, because it is not a task. Idiomatically, we should have stop() only for things that look like there must be an activity there. So it is great for Actor, etc. But for a stream, I like close() better.

[llucax]

I think stop() doesn't make sense for receivers, because it is not a task. Idiomatically, we should have stop() only for things that look like there must be an activity there. So it is great for Actor, etc. But for a stream, I like close() better.

Mmm, receiver don't need to have task, but many do (like Merger, Timer and FileWatcher). If we don't add a task-like interface, it means we can't stop these receivers using the Receiver interface, which is the only way to access them if we apply a filter or map to them for example.

But maybe there are other solutions to this. We can discuss this when the time comes.

[llucax]

Please let me know when this is ready for a final review :)

[shsms]

Mmm, receiver don't need to have task, but many do (like Merger, Timer and FileWatcher). If we don't add a task-like interface, it means we can't stop these receivers using the Receiver interface, which is the only way to access them if we apply a filter or map to them for example.

That doesn't sound like an issue, because people still call close() on the receiver, and if the underlying thing is a FileWatcher for example, the close() would just call the stop(), which is what this PR is doing already.

And calling close() on Map and Filter would call close() of the underlying receiver. This PR has that too.

[llucax]

That doesn't sound like an issue, because people still call close() on the receiver, and if the underlying thing is a FileWatcher for example, the close() would just call the stop(), which is what this PR is doing already.

But stop() is async, and it is important to be able to wait until the thing actually stopped to avoid those annoying warnings in test we were talking about 😆

If we make close() async, then fine, but then we are ACKing that receivers can have background running tasks, and then why shouldn't they follow the Service interface?

[shsms]

This is ready, the only big change is that Receiver.close() is now async.

[llucax]

Last push to have a (Background)Service-compatible interface.

If we make close() async, then fine, but then we are ACKing that receivers can have background running tasks, and then why shouldn't they follow the Service interface?

[llucax]

BTW, the method can't be abstract, otherwise this is a breaking change. We need to provide a default empty implementation I guess. We can create an issue to make the method abstract for 2.0.

[llucax]

I didn't review everything, because I think the review will change a lot depending on what we do with close vs stop() (stop() being close()/cancel() + __await__).

cancel() and __await__ are actually the 2 primitives used by (Background)Service too, based on Task, and the more I look at it, the more I think receivers should follow this interface too.

[llucax]

We should probably do a await self._awatch.aclose() here to make sure all the threads from the awatch are stopped too. Maybe even mark the receiver somehow as stopped so ready() doesn ´t try to use self._awatch again? Not sure what would happen otherwise, what kind of exception would be raised.

[llucax]

I think here we should also break the ready() if it is waiting for the condition variable. I guess this would mean adding a an event and then async.wait() on the first one to complete in ready()?

I guess this is actually the difference between closing and stopping. Close is basically analogous to Task.cancel(), it is like it schedules the stopping. And it is actually probably a good idea to have it separated, but we are still missing the await then, making stop() async but then not actually awaiting for anything seems weird to me.