Pass a callback into a Stream that will be called when the Stream is dropped

drop_stream.rs

use ::pin_project::{pin_project, pinned_drop};
use futures::{stream::Fuse, StreamExt};
use std::pin::Pin;

#[pin_project(PinnedDrop)]
pub struct DropStream<St, Item, C>
where
    St: futures::Stream<Item = Item>,
    C: FnMut(),
{
    drop: C,
    #[pin]
    stream: Fuse<St>,
}

/// Returns a stream that with a callback when the Stream is dropped.
pub fn drop_stream<St, Item, C>(stream: St, drop: C) -> DropStream<St, Item, C>
where
    St: futures::Stream<Item = Item>,
    C: FnMut(),
{
    DropStream::<St, Item, C> {
        drop: drop,
        stream: stream.fuse(),
    }
}

impl<St, Item, C> futures::Stream for DropStream<St, Item, C>
where
    St: futures::Stream<Item = Item>,
    C: FnMut(),
{
    type Item = St::Item;

    fn poll_next(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        self.project().stream.poll_next(cx)
    }
}

#[pinned_drop]
impl<St, Item, C> PinnedDrop for DropStream<St, Item, C>
where
    St: futures::Stream<Item = Item>,
    C: FnMut(),
{
    fn drop(self: Pin<&mut Self>) {
        let this = self.project();
        (this.drop)()
    }
}