Issue with Notification stream blocking thread indefinitely.

[aaarkid]

I needed to have a Vec of notification streams to 1. check notifications periodically and 2. update the Vec whenever I add a new device. However the notification streams appear to be Stream and not TryStream so is there any way for me to check the notification without blocking the thread indefinitely?

[aaarkid]

Here is some code for example and it won't proceed the flow if there is None notification:

while let Some(notification) = notifications.next().await {
        debug!("Received notification from {}", device_name);
        let device = device_name.clone();
        let notification = DiceNotification {device, notification};
        sink.send(notification).await.unwrap();
}

The sink is Tokio MPSC btw, but this implementation has the same issue without it.

[BeaconBrigade]

Hey @aaarkid, I've run into the same issue. Did you come up with a solution?

[aaarkid]

Hey @aaarkid, I've run into the same issue. Did you come up with a solution?

What OS are you trying this on?

[BeaconBrigade]

What OS are you trying this on?

MacOS.

I ended up solving the problem by creating multiple tokio tasks. One task calls notification_stream.next().await and sends the data through a channel to another task. The second task tokio::select!s on the bluetooth channel and a command channel. This task handles the bluetooth data or commands (including a stop command). Then there's a handle returned from the function that spawned these tasks which can send commands to the second task.

Sorry if that explanation wasn't really clear. Essentially there's three parts, the ble task which reads data, the command task which processes data and commands and the main task where the programmer sends commands.

[aaarkid]

It makes complete sense. I solved this issue the same way. Too bad the devs didn't abide by the async contract in this occasion, but there's probably a good reason behind it.

[BeaconBrigade]

Fair enough

[blandger]

Author

Did you have a code (link) to show the idea and apporach ?

[BeaconBrigade]

I made a proof of concept here but I have a more real implementation in this file for a project I'm contributing to which has the real btleplug.

[aaarkid]

Hey @BeaconBrigade. I sent you an email to the address in your GitHub about, but I'm not sure if it went to spam. Can you please check it and get back to me?

[BeaconBrigade]

I see your email now

[aaarkid]

I'm working on a solution I'll publish soon once it's ready. Fortunately, I can say that these streams could be fixed without any breaking changes.

[dmtrKovalenko]

I did solve it without additional threads, you can just tokio timeout on the notification_sream.next()

while let Some(data) = timeout(TIMEOUT, notification_stream.next()).await? {

[BeaconBrigade]

That makes sense. I thougt about doing something like tokio::select! on the notification stream and like a command queue but was worried what the effect of cancelling the notification_stream.next() future would be. Would I lose data? But of course this still requires you to spawn a task.

Good idea.

[qdot]

It makes complete sense. I solved this issue the same way. Too bad the devs didn't abide by the async contract in this occasion, but there's probably a good reason behind it.

I honestly can't remember if I had a good reason behind it so it's something that could at least be explored if it doesn't break too much. Mostly I just don't have time for dev on this right now.

That makes sense. I thougt about doing something like tokio::select! on the notification stream and like a command queue but was worried what the effect of cancelling the notification_stream.next() future would be. Would I lose data? But of course this still requires you to spawn a task.

I... think this is how I handle things in my projects that use btleplug, but I also spawn tons of tasks already.