Split into gui and client crates (#30)

Reviewed-on: #30
Reviewed-by: restitux <restitux@ohea.xyz>
Co-authored-by: Sam Sartor <me@samsartor.com>
Co-committed-by: Sam Sartor <me@samsartor.com>

client/src/mainloop.rs

@@ -0,0 +1,442 @@
+use crate::msghtml::process_message_html;
+use crate::AudioSettings;
+use crate::Chat;
+use crate::Command;
+use crate::ConnectionState;
+use asynchronous_codec::FramedRead;
+use asynchronous_codec::FramedWrite;
+use color_eyre::eyre::{bail, Error};
+use dioxus_signals::ReadableExt as _;
+use dioxus_signals::WritableExt as _;
+use futures::select;
+use futures::AsyncRead;
+use futures::AsyncWrite;
+use futures::FutureExt as _;
+use futures::SinkExt as _;
+use futures::StreamExt as _;
+use futures_channel::mpsc::{UnboundedReceiver, UnboundedSender};
+use mumble_protocol::control::msgs;
+use mumble_protocol::control::ControlCodec;
+use mumble_protocol::control::ControlPacket;
+use mumble_protocol::voice::VoicePacket;
+use mumble_protocol::voice::VoicePacketPayload;
+use mumble_protocol::Clientbound;
+use mumble_protocol::Serverbound;
+use std::collections::hash_map::Entry;
+use std::collections::HashMap;
+use std::time::Duration;
+use tracing::error;
+use tracing::info;
+
+use crate::app::SharedState;
+use crate::app::State;
+use crate::effects::AudioProcessor;
+use crate::imp::{
+    spawn, AudioPlayer, AudioPlayerInterface as _, AudioSystem, AudioSystemInterface as _,
+    Platform, PlatformInterface as _,
+};
+
+pub async fn network_entrypoint(mut event_rx: UnboundedReceiver<Command>, state: SharedState) {
+    loop {
+        let Some(Command::Connect {
+            address,
+            username,
+            config,
+        }) = event_rx.next().await
+        else {
+            panic!("did not receive connect command")
+        };
+
+        *state.server.write_unchecked() = Default::default();
+        *state.status.write_unchecked() = ConnectionState::Connecting;
+        if let Err(error) =
+            Platform::network_connect(address, username, &mut event_rx, &config, state.clone())
+                .await
+        {
+            error!("could not connect {:?}", error);
+            *state.status.write_unchecked() = ConnectionState::Failed(error.to_string());
+        } else {
+            *state.status.write_unchecked() = ConnectionState::Disconnected;
+        }
+    }
+}
+
+pub(crate) async fn sender_loop<W: AsyncWrite + Unpin + 'static>(
+    mut outgoing: UnboundedReceiver<ControlPacket<Serverbound>>,
+    mut writer: FramedWrite<W, ControlCodec<Serverbound, Clientbound>>,
+) {
+    while let Some(msg) = outgoing.next().await {
+        if !matches!(msg, ControlPacket::Ping(_) | ControlPacket::UDPTunnel(_)) {
+            info!("sending packet {:#?}", msg);
+        }
+        if let Err(e) = writer.send(msg).await {
+            error!("error sending packet {:?}", e);
+            break;
+        }
+    }
+}
+
+pub(crate) async fn network_loop<R: AsyncRead + Unpin + 'static>(
+    username: String,
+    state: SharedState,
+    event_rx: &mut UnboundedReceiver<Command>,
+    mut outgoing: UnboundedSender<ControlPacket<Serverbound>>,
+    mut reader: FramedRead<R, ControlCodec<Serverbound, Clientbound>>,
+) -> Result<(), Error> {
+    let audio_settings = state.audio.read().clone();
+
+    // Get version packet
+    let version = match reader.next().await {
+        Some(Ok(v)) => v,
+        Some(Err(err)) => bail!("bad version packet: {err:?}"),
+        None => bail!("no version was recieved"),
+    };
+    info!("got version packet {:#?}", version);
+
+    // Send version packet
+    let mut msg = msgs::Version::new();
+    msg.set_version(0x000010204);
+    msg.set_release(format!("{} {}", "mumbleweb2", "6.9.0"));
+    //msg.set_os("Chrome".to_string());
+    outgoing.send(msg.into()).await.unwrap();
+
+    // Send authenticate packet
+    let mut msg = msgs::Authenticate::new();
+    msg.set_username(username);
+    msg.set_opus(true);
+    outgoing.send(msg.into()).await.unwrap();
+
+    // Spawn worker to send pings
+    {
+        let mut send_chan = outgoing.clone();
+        spawn(async move {
+            loop {
+                if let Err(_) = send_chan.send(msgs::Ping::new().into()).await {
+                    break;
+                }
+
+                Platform::sleep(Duration::from_millis(3000)).await;
+            }
+        });
+    }
+
+    let mut audio = AudioSystem::new().await?;
+    audio.set_processor(AudioProcessor::new(audio_settings.denoise));
+    {
+        let send_chan = outgoing.clone();
+        let mut sequence_num = 0;
+        if let Err(err) = audio.start_recording(move |opus_frame, is_terminator| {
+            let _ =
+                send_chan.unbounded_send(ControlPacket::UDPTunnel(Box::new(VoicePacket::Audio {
+                    _dst: std::marker::PhantomData,
+                    target: 0,
+                    session_id: (),
+                    seq_num: sequence_num,
+                    payload: VoicePacketPayload::Opus(opus_frame.into(), is_terminator),
+                    position_info: None,
+                })));
+            sequence_num = sequence_num.wrapping_add(2);
+        }) {
+            error!("could not begin recording: {err:?}")
+        }
+    }
+
+    // Create map of session_id -> AudioDecoder
+    let mut decoder_map = HashMap::new();
+
+    let mut reader_future = reader.next().fuse();
+    let mut command_future = event_rx.next();
+
+    loop {
+        select! {
+            packet = reader_future => {
+                reader_future = reader.next().fuse();
+                match packet {
+                    Some(Ok(msg)) => {
+                        if !matches!(msg, ControlPacket::UDPTunnel(_) | ControlPacket::Ping(_)) {
+                            info!("receiving packet {:#?}", msg);
+                        }
+                        let res = accept_packet(msg, &mut audio, &mut decoder_map, &state);
+                        if let Err(err) = res {
+                            error!("error accepting packet {:?}", err)
+                        }
+                    },
+                    Some(Err(err)) => {
+                        error!("error receiving packet {:?}", err)
+                    },
+                    None => break,
+                }
+            }
+            command = command_future => {
+                command_future = event_rx.next();
+                if let Some(command) = &command {
+                    info!("issuing command {:#?}", command);
+                }
+                match command {
+                    Some(Command::Disconnect) => break,
+                    Some(command) => {
+                        let res = accept_command(command, &mut outgoing, &mut audio, &state);
+                        if let Err(err) = res {
+                            info!("error accepting command {:?}", err)
+                        }
+                    }
+                    None => continue,
+                }
+            }
+        }
+    }
+    let _ = outgoing.close();
+
+    Ok(())
+}
+
+fn accept_command(
+    command: Command,
+    send_chan: &mut UnboundedSender<ControlPacket<mumble_protocol::Serverbound>>,
+    audio: &mut AudioSystem,
+    state: &State,
+) -> Result<(), Error> {
+    use Command::*;
+    let Some(session) = state.server.read().session else {
+        bail!("no session id")
+    };
+
+    match command {
+        SendChat { markdown, channels } => {
+            use markdown::*;
+            let blocks = tokenize(&markdown);
+            let html_text = match blocks.as_slice() {
+                [Block::Paragraph(par)] => match par.as_slice() {
+                    [Span::Text(text)] => text.to_string(),
+                    _ => to_html(&markdown)
+                        .trim()
+                        .strip_prefix("<p>")
+                        .unwrap()
+                        .strip_suffix("</p>")
+                        .unwrap()
+                        .to_string(),
+                },
+                _ => to_html(&markdown).trim().to_string(),
+            };
+
+            {
+                let mut server = state.server.write_unchecked();
+                let Some(me) = server.session else {
+                    bail!("not signed in with a session id")
+                };
+                server.chat.push(Chat {
+                    raw: markdown,
+                    dangerous_html: html_text.clone(),
+                    sender: Some(me),
+                })
+            }
+
+            let mut u = msgs::TextMessage::new();
+            u.set_message(html_text.to_string());
+            u.set_channel_id(channels);
+            let _ = send_chan.unbounded_send(u.into());
+        }
+        SendFile {
+            ref bytes,
+            name,
+            mime,
+            channels,
+        } => {
+            use base64::{display::Base64Display, prelude::BASE64_STANDARD};
+            let html = match mime {
+                Some(mime) if mime.type_() == "image" => format!(
+                    "<img src=\"data:{};base64,{}\" />",
+                    mime,
+                    Base64Display::new(bytes, &BASE64_STANDARD)
+                ),
+                Some(mime) => format!(
+                    "<a href=\"data:{};base64,{}\" download>{name}</a>",
+                    mime,
+                    Base64Display::new(bytes, &BASE64_STANDARD)
+                ),
+                None => format!(
+                    "<a href=\"data:application/octet-stream;base64,{}\" download>{name}</a>",
+                    Base64Display::new(bytes, &BASE64_STANDARD)
+                ),
+            };
+
+            {
+                let mut server = state.server.write_unchecked();
+                let Some(me) = server.session else {
+                    bail!("not signed in with a session id")
+                };
+                server.chat.push(Chat {
+                    raw: "".to_string(),
+                    dangerous_html: html.clone(),
+                    sender: Some(me),
+                })
+            }
+
+            let mut u = msgs::TextMessage::new();
+            u.set_message(html);
+            u.set_channel_id(channels);
+            let _ = send_chan.unbounded_send(u.into());
+        }
+        SetMute { mute } => {
+            let mut u = msgs::UserState::new();
+            u.set_session(session);
+            u.set_self_mute(mute);
+            let _ = send_chan.unbounded_send(u.into());
+        }
+        SetDeaf { deaf } => {
+            let mut u = msgs::UserState::new();
+            u.set_session(session);
+            u.set_self_deaf(deaf);
+            let _ = send_chan.unbounded_send(u.into());
+        }
+        EnterChannel { channel, user } => {
+            let mut u = msgs::UserState::new();
+            u.set_session(user);
+            u.set_channel_id(channel);
+            let _ = send_chan.unbounded_send(u.into());
+        }
+        Connect { .. } | Disconnect => (),
+        UpdateAudioSettings(AudioSettings { denoise }) => {
+            audio.set_processor(AudioProcessor::new(denoise));
+        }
+    }
+
+    Ok(())
+}
+
+fn accept_packet(
+    msg: ControlPacket<mumble_protocol::Clientbound>,
+    audio_context: &mut AudioSystem,
+    player_map: &mut HashMap<u32, AudioPlayer>,
+    state: &State,
+) -> Result<(), Error> {
+    match msg {
+        ControlPacket::UDPTunnel(u) => {
+            match *u.clone() {
+                mumble_protocol::voice::VoicePacket::Audio {
+                    _dst,
+                    target,
+                    session_id,
+                    seq_num,
+                    payload,
+                    position_info,
+                } => {
+                    // Get or create audio decoder for this user
+                    let audio_player = match player_map.entry(session_id) {
+                        Entry::Occupied(occupied_entry) => occupied_entry.into_mut(),
+                        Entry::Vacant(vacant_entry) => {
+                            vacant_entry.insert(audio_context.create_player()?)
+                        }
+                    };
+                    // This will over time (as users join and leave) leak
+                    // AudioDecoders, MediaStreamTrackGenerators, MediaStreams, and MediaStreamAudioSourceNodes.
+                    // A better way to handle this would be to delete and create all the audio
+                    // infra on channel join and update it as new users join the channel, dropping
+                    // any audio packets that come in the meantime.
+                    if let VoicePacketPayload::Opus(audio_payload, end_bit) = payload {
+                        audio_player.play_opus(audio_payload.as_ref());
+                        //console::log_1(&"Oueued audio chunk for decoding".into());
+                    }
+                }
+                _ => {
+                    unreachable!("TCP tunnels UDP packets should not contain pings");
+                    // I think?
+                }
+            }
+        }
+        ControlPacket::ChannelState(u) => {
+            let mut server = state.server.write_unchecked();
+            server.channels_state.update_from_channel_state(&u);
+        }
+        ControlPacket::ChannelRemove(u) => {
+            let mut server = state.server.write_unchecked();
+            server.channels_state.update_from_channel_remove(&u);
+        }
+        ControlPacket::UserState(u) => {
+            let mut server = state.server.write_unchecked();
+            let server = &mut *server;
+            let id = u.get_session();
+
+            let state_entry = server.users.entry(id);
+            let new = matches!(state_entry, std::collections::hash_map::Entry::Vacant(_));
+            let state = state_entry.or_default();
+            // the server might now send a channel_id if the user is in channel=0
+            if u.has_channel_id() || new {
+                if let Some(parent) = server.channels_state.channels.get_mut(&state.channel) {
+                    parent.users.remove(&id);
+                }
+
+                let channel_id = u.get_channel_id();
+                server
+                    .channels_state
+                    .channels
+                    .entry(channel_id)
+                    .or_default()
+                    .users
+                    .insert(id);
+                state.channel = channel_id;
+            }
+
+            if u.has_name() {
+                state.name = u.get_name().to_string();
+            }
+            if u.has_mute() {
+                state.mute = u.get_mute();
+            }
+            if u.has_deaf() {
+                state.deaf = u.get_deaf();
+            }
+            if u.has_suppress() {
+                state.suppress = u.get_suppress();
+            }
+            if u.has_self_mute() {
+                state.self_mute = u.get_self_mute();
+            }
+            if u.has_self_deaf() {
+                state.self_deaf = u.get_self_deaf();
+            }
+        }
+        ControlPacket::UserRemove(u) => {
+            let mut server = state.server.write_unchecked();
+            let id = u.get_session();
+            if let Some(state) = server.users.remove(&id) {
+                if let Some(parent) = server.channels_state.channels.get_mut(&state.channel) {
+                    parent.users.remove(&id);
+                }
+            }
+        }
+        ControlPacket::TextMessage(u) => {
+            let mut server = state.server.write_unchecked();
+            if u.has_message() {
+                let text = u.get_message().to_string();
+                server.chat.push(Chat {
+                    sender: if u.has_actor() {
+                        Some(u.get_actor())
+                    } else {
+                        None
+                    },
+                    dangerous_html: process_message_html(&text),
+                    raw: text,
+                });
+            }
+        }
+        ControlPacket::ServerSync(u) => {
+            *state.status.write_unchecked() = ConnectionState::Connected;
+            let mut server = state.server.write_unchecked();
+            if u.has_welcome_text() {
+                let text = u.get_welcome_text().to_string();
+                server.chat.push(Chat {
+                    sender: None,
+                    dangerous_html: process_message_html(&text),
+                    raw: text,
+                });
+            }
+            if u.has_session() {
+                server.session = Some(u.get_session());
+            }
+        }
+        _ => {}
+    }
+
+    Ok(())
+}