Subscription Close detection mechanism / event?

[AadamZ5]

I'm trying to employ Juniper with some real-data subscriptions. I'm getting a change feed from my database, and mapping that stream into a User object that Juniper can resolve. It's working great! I now need to know when the subscription has ended so I can close the stream from my database.

Is there any parameter that my Subscription function can intake to know when the client has ended the subscription? I need some sort of mechanism to know when the subscription has ended.

For reference, my Subscription struct:

pub struct Subscription;

#[graphql_subscription(context = Context)]
impl Subscription {
    #[graphql()]
    async fn watch_user(
        #[graphql(context)] ctx: &Context,
        #[graphql(description = "The ID of the user to watch")] id: String,
    ) -> GraphStream<User> {
        log::debug!("Starting stream watching user ID \"{}\"", id);

        //TODO: Rust might not appreciate the "service" paradigm, wonder if something else fits better...
        let inner_stream = ctx.user_service.watch_user(id.clone());

        let out_stream = inner_stream
            .map(|u| {
                u.map_err(|e| {
                    FieldError::new(e.to_string(), Value::null() as Value<DefaultScalarValue>)
                })
            })
            .boxed();

        out_stream
    }
}

I need to supply some sort of logic (maybe a oneshot tx/rx) to my Database implementation to close the connection. I can pass this in as an argument to my ctx.user_service.watch_user function. How can I get some sort of notification or future to await? Ideally, it would be something I can move or clone to a different separate running async spawned task.

Edit:
Just found #1066 which talks about using Drop for this. I am going to checkout stream-cancel which implements some wrappers for triggering logic on Drop. Otherwise, any other possibilities?

[AadamZ5]

Possible answer for now

I've implemented a quick Stream wrapper called DropStream<T>.

use futures::{
    channel::oneshot::{Receiver, Sender},
    Stream,
};
use std::pin::Pin;

/// Stream wrapper that sends a message to a oneshot reciever upon being dropped.
#[derive(Debug)]
pub struct DropStream<T>
where
    T: Stream + ?Sized,
{
    pub stream: Pin<Box<T>>,
    pub tx: Option<Sender<()>>,
}

impl<T> DropStream<T>
where
    T: Stream + ?Sized,
{
    pub fn new(stream: Pin<Box<T>>) -> (DropStream<T>, Receiver<()>) {
        let (tx, rx) = futures::channel::oneshot::channel();

        let myself = Self {
            stream,
            tx: Some(tx),
        };

        return (myself, rx);
    }

    ///Given an existing `Stream` and `Sender`, create a `DropStream` wrapper.
    pub fn from_existing_tx(stream: Pin<Box<T>>, tx: Sender<()>) -> DropStream<T> {
        let myself = Self {
            stream,
            tx: Some(tx),
        };

        return myself;
    }
}

impl<T> Drop for DropStream<T>
where
    T: Stream + ?Sized,
{
    fn drop(&mut self) {
        if let Some(tx) = self.tx.take() {
            let _ = tx.send(());
        };
    }
}

impl<T> Stream for DropStream<T>
where
    T: Stream + ?Sized,
{
    type Item = T::Item;

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

Implementation details

In my implementation, I create a oneshot sender and reciever. I pass the reciever to my back-end where it can listen for the single message.

Subscription Object:

#[graphql_subscription(context = Context)]
impl Subscription {
    #[graphql()]
    async fn watch_user(
        #[graphql(context)] ctx: &Context,
        #[graphql(description = "The ID of the user to watch")] id: String,
    ) -> GraphStream<User> {
        log::debug!("Starting stream watching user ID \"{}\"", id);

        let (tx, rx) = futures::channel::oneshot::channel();

        //TODO: Rust might not appreciate the "service" paradigm, wonder if something else fits better...
        let inner_stream = ctx.user_service.watch_user(id.clone(), rx);

        let mapped_stream = inner_stream.map(|u| {
            u.map_err(|e| {
                FieldError::new(e.to_string(), Value::null() as Value<DefaultScalarValue>)
            })
        });

        let wired_stream = DropStream::from_existing_tx(mapped_stream.boxed(), tx);

        let wired_stream = wired_stream.boxed();

        wired_stream
    }
}

Database connection logic

fn watch_user(
    &self,
    id: String,
    close_oneshot: Receiver<()>,
) -> crate::models::user::UserStream {
    let conn_result = self.connection();
    let mut conn = conn_result.unwrap();

    log::debug!("Watching user: {}", id);

    let error_title = "Error in watch_user:";

    let err_fn = move |e| match e {
        reql::Error::Driver(e) => {
            let newmsg = format!("Driver error: {:?}", e);
            log::error!("{}{}", error_title, newmsg);
            newmsg
        }
        reql::Error::Runtime(e) => {
            let newmsg = format!("Runtime error: {}", e.to_string());
            log::error!("{}{}", error_title, newmsg);
            newmsg
        }
        any_e => {
            let newmsg = format!("Unknown error: {}", any_e.to_string());
            log::error!("{}{}", error_title, newmsg);
            newmsg
        }
    };

    let change_conn = conn.clone();

    let stream = r
        .db("main")
        .table("users")
        .get(id)
        .changes(())
        .run::<Connection, reql::types::Change<User, User>>(change_conn)
        .map(move |change_result| {
            log::debug!("Change result {:?}", change_result);
            match change_result {
                Ok(change) => {
                    if change.new_val.is_none() {
                        Err("User not found".to_string())
                    } else {
                        Ok(change.new_val.unwrap())
                    }
                }
                Err(e) => Err(err_fn(e)),
            }
        })
        .map_err(|e| e.to_string());
    log::debug!("Created changes stream...");

    actix_web::rt::spawn(async move {
        let _msg = close_oneshot.await;
        log::debug!("Closing stream...");
        let _close_success = conn.close(()).await;
    });

    return stream.boxed();
}

[ilslv]

@AadamZ5 #1066 also mentions a watchdog approach, which involves running a background process checking some persistent storage. This storage is also updated from by the Stream returned in the Subscription, while connection is running. Watchdog solution differs from the Drop in ensuring that some code will be eventually executed on the Subscription close, because Drop isn't executed for example in case of a power outage. But with watchdog you'll eventually find stale value in persistent storage, which isn't updated by the failed instance of your app.

But for what you are trying to achieve, Drop should work fine.

[LegNeato]

We should document something like this in the book!