Sending to a subscription outside of .subscription(

[bivald]

Hi,

Sorry for all the questions :)

I have a subscription via RSPC (using Tauri) and I'm trying to send a message through it, but outside of the .subscription( callback (from another thread doing background processing work). I've tried various was of using mpsc patterns from various attack points.

My initial plan was to create a receiver and attach it in subscription, something like

.subscription("pings", |t| {
                    async_stream::stream! {
                        // A loop that takes output from the async process and sends it
                        // to the webview via a Tauri Event
                        while let Some(input) = rx.recv().await {
                            println!("Got message I should send back to JS: {:?}", input);
                            yield input;
                        }
                    }
});

But it seems this won't work due to the "single consumer" part, because it looks like the subscription get's called twice internally (one to get the output format? One once in running?).

I tried doing a more advanced logic where I the first time create the receiver, and then the second time re-use the same but then it falters on no two async blocks, even if identical, have the same type because the async blocks aren't actually the same.

My second (or more like 10th) try was to setup a loop inside rspc_tauri2 and pass it the receiver:

            {
                let app_handle = app_handle.clone();
                let response_sender = resp_tx.clone();
                {
                    tauri::async_runtime::spawn(async move {
                        while let Some(input) = to_js_receiver.recv().await {
                            let input = jsonrpc::ResponseInner::Event(serde_json::Value::String(input));
                            response_sender.send(jsonrpc::Response{
                                 jsonrpc: "2.0",
                                 id: jsonrpc::RequestId::Null,
                                 result: input,
                            });
                        }
                    });
                }
            }

This works, in the sense that a message get sent to the front-end but it's not send using the subscriptions ID (which means). I can work around that, by assuming that the first key of the HashMap is the subscription, but we're getting down into hacky territory :)

Regards,
Niklas

[bivald]

I probably made a misstake in setting up the subscription. I used merge pattern having each router in a separate file. When I moved it into the top level router it worked. In case anyone is interested, here is how the subscription now looks:

    let router = Router::<LightfindState>::new()
         // ... //
        .subscription("pings", |t| {
            t(|ctx, input: ()| {
                let (tx, mut rx) = mpsc::unbounded_channel::<String>();

                let mut to_js_sender = ctx.to_js_sender.write().unwrap();
                *to_js_sender = Some(tx);
                drop(to_js_sender);

                async_stream::stream! {
                    // A loop that takes output from the async process and sends it
                    // to the webview via a Tauri Event
                    while let Some(input) = rx.recv().await {
                        println!("Got message I should send back to JS: {:?}", input);
                        yield input;
                    }
                }
            })
        })

ctx is my state, which has a to_js_sender variable which is a Arc<RwLock<Option<tokio::sync::mpsc::UnboundedSender<String>>>>

[oscartbeaumont]

You should probally consider using a broadcast instead as it allows Clone on the receiver and sender. broadcast is a mpmc channel.

subscription get's called twice internally (one to get the output format? One once in running?).

A subscriptions function is called once and then the stream will being polled until it's complete like regular Rust code. The polling won't result in the same piece of code being run twice as this would likely break Rust's safety guarantees. I wonder if you have multiple browser tabs open with your site, or more likely if your using React we may be creating the subscription twice in development due to how StrictMode works.