How to gracefully unsubscribe

[initprism]

The subscriber is passing data to the receiver channel until the client terminates.
And when the client closes the connection, it wants the subscriber to close and the task to finish.
However, there seems to be no direct way to call unsubscribe from outside the subscriber is subscribing, even if the client terminates.

let mut sc = connection.subscribe(subject).await.unwrap;

tokio::spawn(async move {
    while let Some(m) = sc.next().await {
        tx.send(m);
        // sc.unsubscribe().await
    }
});

// sc.unsubscribe().await

Is there any way how to do it? Even if Arc is used, the subscriber cannot acquire the lock outside while waiting for next while holding the lock.

[YaZasnyal]

Unsubscribe is going to happen if the Subscription is dropped or by calling unsubscribe(). You need to signal a Task to unsubscribe and drain messages that are left. This can be done using tokio oneshot channel and select!ing both futures.

https://docs.rs/tokio/latest/tokio/sync/oneshot/fn.channel.html
https://docs.rs/tokio/latest/tokio/macro.select.html#examples