How can I make subscriptions with Redis?

[Teajey]

Another subscription question.

I have a problem with my subscription here where each currentTriviaQuestion GraphQL request generates a new pubsub subscription in Redis, without an opportunity to call pubsub.unsubscribe. So even when I've terminated my frontend, and there are no longer any active websockets, all of those Redis subscriptions continue to exist; and streams too, for that matter. It also seems like this subscription's getting in a deadlock if it's called more than once.

What's the proper way to handle a subscription like this, and clean up afterwards?

async fn current_trivia_question(context: &Context) -> UnansweredTriviaQuestionStream {
    let redis_url = context.config.redis_url.clone();

    let stream = async_stream::stream! {
        let mut con = data::connect_redis(&redis_url)?;
        let mut pubsub = con.as_pubsub();
        pubsub.subscribe("current_trivia_question")?;

        loop {
            let msg = pubsub.get_message()?;
            let trivia_question: TriviaQuestion = msg.get_payload()?;
            yield Ok(UnansweredTriviaQuestion::from(trivia_question));
        }
    };

    Box::pin(stream)
}

[tyranron]

@Teajey as for me, the best option would be for the UnansweredTriviaQuestionStream, you're returning, to call to call pubsub.unsubscribe on Drop. This is the natural way for cancellation and resource cleaning in Rust.

If you wonder how one could call async in Drop which is sync, the most common patter for the moment (of AsyncDrop absence) is something like that:

impl Drop for MyStream {
    fn drop(&mut self) {
        tokio::spawn(async move {  // or `actix::spawn`, depending on what you're using
            // you async actions
        });
    }
}

[Teajey]

UnansweredTriviaQuestionStream is just a type, difficult to impl on:

type UnansweredTriviaQuestionStream =
    Pin<Box<dyn Stream<Item = Result<UnansweredTriviaQuestion, FieldError>> + Send>>;

I tried putting it in a tuple struct, but that didn't really work

[Teajey]

I guess I would need pubsub and the UnansweredTriviaQuestionStream to be members of a struct, but then I'm not sure how the stream would access pubsub...

[tyranron]

@Teajey yes, just type alias won't be enough here, if the inner stream is not capable of cancellation by itself. So, you should introduce a newtype pattern here (something like that):

struct UnansweredTriviaQuestionStream {
    stream: Pin<Box<dyn Stream<Item = Result<UnansweredTriviaQuestion, FieldError>> + Send>>,
    pubsub: PubSub,
}

impl Stream for UnansweredTriviaQuestionStream {
    //...
}

impl Drop for UnansweredTriviaQuestionStream {
    //...
}

[Teajey]

Yeah... now I'm just reckoning with ownership issues. I've tried passing con on as a member of UnansweredTriviaQuestionStream, but Rust doesn't seem to be able to recognize that that would give con and pubsub the same lifetime.

struct UnansweredTriviaQuestionStream<'a> {
    pubsub: PubSub<'a>,
}

impl<'a> UnansweredTriviaQuestionStream<'a> {
    fn try_new(mut con: Connection) -> FieldResult<Self> {
        let pubsub = con.as_pubsub();
                     --------------- `con` is borrowed here
        Ok(UnansweredTriviaQuestionStream { pubsub })
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ returns a value referencing data owned by the current function
    }
}

impl<'a> Stream for UnansweredTriviaQuestionStream<'a> {
    type Item = FieldResult<UnansweredTriviaQuestion>;

    fn poll_next(
        self: Pin<&mut Self>,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        let msg = Pin::into_inner(self).pubsub.get_message()?;
        let payload = msg.get_payload()?;
        std::task::Poll::Ready(Some(Ok(payload)))
    }
}

impl<'a> Drop for UnansweredTriviaQuestionStream<'a> {
    fn drop(&mut self) {
        self.pubsub.unsubscribe("new_trivia_question");
    }
}

pub struct SubscriptionRoot;

#[graphql_subscription(context = Context)]
impl SubscriptionRoot {
    async fn new_trivia_question<'a>(
        context: &Context,
    ) -> FieldResult<Pin<Box<UnansweredTriviaQuestionStream<'a>>>> {
        let con = data::connect_redis(&context.config.redis_url)?;
        let stream = UnansweredTriviaQuestionStream::try_new(con);

        match stream {
            Ok(stream) => Ok(Box::pin(stream)),
            Err(err) => Err(err),
        }
    }
}

[tyranron]

@Teajey well, the problem is that PubSub is a reference, yet a stateful one. Not having "owned" variant of the one seems like a bad design choice for me. I see this can be resolved in two ways:

1. Either you put the created redis::Connection into the Context, and refer to it in the stream via the signature:

   #[graphql_subscription(context = Context)]
   impl SubscriptionRoot {
       async fn new_trivia_question<'a>(
           context: &'a Context,
       ) -> FieldResult<Pin<Box<UnansweredTriviaQuestionStream<'a>>>> {

   Due to PubSub<'_> holding &mut reference there should be probably additional problems and squats with interior mutability.

2. Or use something like rental to baked into your stream both redis::Connection and redis::PubSub simultaneously.

....oh, wait! The whole thing is wrong!

As I see, you're using synchronous redis::Connection in async fn. This is bad and will block your thread away!

Instead, you should use redis::aio::Connection, which has the required .into_pubsub() method returning the owned version as you require, so you may remove the nasty lifetime from you Stream altogether.

[Teajey]

Awesome! Thanks to @tyranron's help I seem to have got it working with an asynchronous solution using redis::aio:

    async fn new_trivia_question(
        context: &Context,
    ) -> FieldResult<Pin<Box<dyn Stream<Item = FieldResult<UnansweredTriviaQuestion>> + Send>>>
    {
        let redis_url = context.config.redis_url.clone();

        let mut pubsub = data::async_connect_redis(&redis_url).await?.into_pubsub();
        pubsub.subscribe("new_trivia_question").await?;

        let stream = pubsub
            .into_on_message()
            .map(|m| {
                m.get_payload::<UnansweredTriviaQuestion>()
                    .map_err(|e| FieldError::from(e))
            })
            .boxed();

        Ok(stream)
    }