How to cancel a Udp stream

[Sheap]

I'm using Tokio to handle udp connections in a tauri app, interfacing with the frontend via an event system.

I need the frontend to be able to close the udp connection, but it's unclear how to do this.

I've tried a few different methods, e.g. that outlined here, but all have some sort of error. In the case of the linked solution, I was receiving errors that the channel could not be implicitly typed due to being in an async function. I've also tried using something like stream_cancel, where I was unable to drop the trigger within the listen handle function, due to borrow rules. I couldn't think of a way around this.

Here is my current code, with no attempted solution, as all of my attempts so far have been much harder to read.

#[tauri::command]
async fn start_udp_server(app_handle: tauri::AppHandle, address: String) -> Result<(), String> {
  match UdpSocket::bind(address.to_string()).await {
    Ok(socket) => {
      let framed = UdpFramed::new(socket, BytesCodec::new());
      let (_write, read) = framed.split();

      let udp_listener = {
        read.for_each(|udp_message| async {
          match udp_message {
            Ok((bytes, _socket_address)) => {
              app_handle.emit_all("udp_trigger", bytes.to_ascii_lowercase()).unwrap();
            },
            Err(error) => {
              println!("udp error: {}", error);
            }
          }
        })
      };

      app_handle.listen_global("udp_close", |_| {
        println!("got close command");
        // how to close?
      });

      app_handle.emit_all("udp_listening", true).unwrap();
      udp_listener.await;
      println!("closed");
      Ok(())
    },
    Err(_) => Err("Error binding to address".into())
  }
}

[Darksonn]

Give this a try:

let (send, mut recv) = tokio::sync::mpsc::channel::<()>();
app_handle.listen_global("udp_close", move |_| {
    println!("got close command");
    send.send(());
});

app_handle.emit_all("udp_listening", true).unwrap();

tokio::select! {
    _ = udp_listener => {},
    _ = recv.recv() => {},
}
println!("closed");
Ok(())

[Sheap]

That didn't quite work. Mixing compiler help with a little of your answer in the similar thread, I have something which compiles:

async fn the_loop (app_handle: &tauri::AppHandle, read: SplitStream<UdpFramed<BytesCodec>>) {
  read.for_each(|udp_message| async {
    match udp_message {
      Ok((bytes, _socket_address)) => {
        println!("received udp message");
        app_handle.emit_all("udp_trigger", bytes.to_ascii_lowercase()).unwrap();
      },
      Err(error) => {
        println!("udp error: {}", error);
      }
    }
  }).await;
}

#[tauri::command]
async fn start_udp_server(app_handle: tauri::AppHandle, address: String) -> Result<(), String> {
  match UdpSocket::bind(address.to_string()).await {
    Ok(socket) => {
      println!("socket created");
      let framed = UdpFramed::new(socket, BytesCodec::new());
      let (_write, read) = framed.split();
      println!("frame created");

      let (send, mut recv) = tokio::sync::mpsc::channel::<()>(1);

      app_handle.listen_global("udp_close", move |_| {
        println!("got close command");
        send.send(());
      });
      println!("listener created");

      app_handle.emit_all("udp_listening", true).unwrap();
      tokio::select! {
        _ = the_loop(&app_handle, read) => {},
        _ = recv.recv() => {},
      }
      println!("selected?");

      println!("closed");
      Ok(())
    },
    Err(_) => Err("Error binding to address".into())
  }
}

But I get this warning:

warning: unused implementer of `futures_util::Future` that must be used
  --> src/main.rs:70:9
   |
70 |         send.send(());
   |         ^^^^^^^^^^^^^^
   |
   = note: `#[warn(unused_must_use)]` on by default
   = note: futures do nothing unless you `.await` or poll them

awaiting seems to be the appropriate thing here, but I can't use an async closure, for a couple of reasons.

error[E0658]: async closures are unstable
  --> src/main.rs:68:45
   |
68 |       app_handle.listen_global("udp_close", async move |_| {
   |                                             ^^^^^
   |
   = note: see issue #62290 <https://github.com/rust-lang/rust/issues/62290> for more information
   = help: add `#![feature(async_closure)]` to the crate attributes to enable
   = help: to use an async block, remove the `||`: `async {`

error[E0308]: mismatched types
  --> src/main.rs:68:45
   |
68 |       app_handle.listen_global("udp_close", async move |_| {
   |                                             ^^^^^^^^^^^^^^ expected `()`, found opaque type
   |
  ::: /home/stuart/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/mod.rs:61:43
   |
61 | pub const fn from_generator<T>(gen: T) -> impl Future<Output = T::Return>
   |                                           ------------------------------- the found opaque type
   |
   = note: expected unit type `()`
            found opaque type `impl futures_util::Future`

[Darksonn]

Use an unbounded mpsc channel instead.

[Sheap]

That worked! Thanks a lot!

For anybody in the future with a similar issue, this is the code which worked:

async fn the_loop (app_handle: &tauri::AppHandle, read: SplitStream<UdpFramed<BytesCodec>>) {
  read.for_each(|udp_message| async {
    match udp_message {
      Ok((bytes, _socket_address)) => {
        println!("received udp message");
        app_handle.emit_all("udp_trigger", bytes.to_ascii_lowercase()).unwrap();
      },
      Err(error) => {
        println!("udp error: {}", error);
      }
    }
  }).await;
}

#[tauri::command]
async fn start_udp_server(app_handle: tauri::AppHandle, address: String) -> Result<(), String> {
  match UdpSocket::bind(address.to_string()).await {
    Ok(socket) => {
      println!("socket created");
      let framed = UdpFramed::new(socket, BytesCodec::new());
      let (_write, read) = framed.split();
      println!("frame created");

      let (send, mut recv) = tokio::sync::mpsc::unbounded_channel();

      app_handle.listen_global("udp_close", move |_| {
        println!("got close command");
        send.send(());
      });
      println!("listener created");

      app_handle.emit_all("udp_listening", true).unwrap();
      tokio::select! {
        _ = the_loop(&app_handle, read) => {},
        _ = recv.recv() => {},
      }

      println!("closed");
      Ok(())
    },
    Err(_) => Err("Error binding to address".into())
  }
}