How do we idiomatically clean up resources after a stream is done being consumed?

[jedwards1211]

Admittedly I'm a beginner at Rust, but it seems like Streams don't provide any way to signal when the consumer is done, so I'm wondering how I would clean up resources after that?

As a hand-wavy example let's say I subscribe to a redis topic while a client is subscribed:

    pub fn handle_client_subscription(mut self) -> impl Stream<Item = crate::Result<Message>> {
        let subscriber = self.mini_redis_client.subscribe(vec!["numbers".to_string()]).await?;
        return subscriber.into_stream();
        // how to self.mini_redis_client.unsubscribe once the consumer is done with this stream?
    }

Is there an idiomatic way I could make self.mini_redis_client unsubscribe from the topic once the stream is no longer being consumed (e.g. after a websocket handler receives an unsubscribe message from the client and stops consuming this stream)? Do people typically acquire resources via something that implements the Drop trait so that those resources automatically get cleaned up when they go out of scope (I'm assuming ownership of those resources would get moved into the stream, and they would get dropped when the stream goes out of scope, etc?)

My colleague expected a design where subscription methods return an implementation with an explicit method that gets called when the subscription is done. But Streams aren't designed this way so I'm hoping there's some other idiom for cleaning up once the stream is no longer consumed?