event factoring, part 1

src/app/mod.rs

@@ -7,7 +7,9 @@ use std::{
 
 use crossterm::event::{Event, EventStream, KeyCode, KeyEvent, KeyEventKind};
 use futures::{FutureExt, StreamExt};
-use jocalsend::{JocalEvent, JocalService, ReceiveDialog, ReceiveRequest, error::Result};
+use jocalsend::{
+    DEFAULT_INTERVAL, JocalEvent, JocalService, ReceiveDialog, ReceiveRequest, error::Result,
+};
 use julid::Julid;
 use log::{LevelFilter, debug, error, warn};
 use ratatui::{
@@ -124,12 +126,11 @@ impl App {
                             self.receive_requests.remove(&id);
                         }
                         JocalEvent::SendApproved(_id) => todo!(),
-                        JocalEvent::SendDenied(_id) => todo!(),
+                        JocalEvent::SendDenied => todo!(),
                         JocalEvent::Tick => {}
                     }
                 }
             }
-            _ = tokio::time::sleep(Duration::from_millis(200)) => {}
         }
 
         Ok(())

src/lib.rs

@@ -31,9 +31,11 @@ pub const DEFAULT_PORT: u16 = 53317;
 pub const MULTICAST_IP: Ipv4Addr = Ipv4Addr::new(224, 0, 0, 167);
 pub const LISTENING_SOCKET_ADDR: SocketAddrV4 =
     SocketAddrV4::new(Ipv4Addr::from_bits(0), DEFAULT_PORT);
-
 pub const DEFAULT_INTERVAL: Duration = Duration::from_millis(200);
 
+pub type Peers = Arc<Mutex<BTreeMap<String, (SocketAddr, Device)>>>;
+pub type Sessions = Arc<Mutex<BTreeMap<String, Session>>>; // Session ID to Session
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum JocalTasks {
     Udp,
@@ -48,11 +50,11 @@ pub enum ReceiveDialog {
     Deny,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub enum JocalEvent {
     ReceivedInbound(Julid),
-    SendApproved(Julid),
+    SendApproved(String),
-    SendDenied(Julid),
+    SendDenied,
     Cancelled(Julid),
     ReceiveRequest { id: Julid, request: ReceiveRequest },
     Tick,
@@ -65,6 +67,14 @@ pub struct ReceiveRequest {
     pub tx: UnboundedSender<ReceiveDialog>,
 }
 
+impl PartialEq for ReceiveRequest {
+    fn eq(&self, other: &Self) -> bool {
+        self.alias == other.alias
+    }
+}
+
+impl Eq for ReceiveRequest {}
+
 impl Debug for ReceiveRequest {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("ReceiveRequest")
@@ -77,8 +87,8 @@ impl Debug for ReceiveRequest {
 /// Contains the main network and backend state for an application session.
 #[derive(Clone)]
 pub struct JocalService {
-    pub peers: Arc<Mutex<BTreeMap<String, (SocketAddr, Device)>>>,
+    pub peers: Peers,
-    pub sessions: Arc<Mutex<BTreeMap<String, Session>>>, // Session ID to Session
+    pub sessions: Sessions,
     pub running_state: Arc<Mutex<RunningState>>,
     pub socket: Arc<UdpSocket>,
     pub client: reqwest::Client,

src/main.rs

@@ -1,7 +1,7 @@
 use std::{path::Path, str::FromStr, time::Duration};
 
 use clap::Parser;
-use jocalsend::{Config, JocalService, JocalTasks, error::Result};
+use jocalsend::{Config, DEFAULT_INTERVAL, JocalService, JocalTasks, error::Result};
 use log::{error, info};
 use ratatui::DefaultTerminal;
 use ratatui_explorer::FileExplorer;
@@ -55,7 +55,7 @@ async fn start_and_run(terminal: &mut DefaultTerminal, config: Config) -> Result
 
     let mut handles = JoinSet::new();
     app.service.start(&mut handles).await;
-    let mut tick = tokio::time::interval(Duration::from_millis(200));
+    let mut tick = tokio::time::interval(DEFAULT_INTERVAL);
     let shutdown = shutdown(&mut handles);
     let mut shutdown = std::pin::pin!(shutdown);
     loop {

src/transfer.rs

@@ -10,10 +10,10 @@ use axum::{
 use julid::Julid;
 use log::{debug, error, info, warn};
 use serde::{Deserialize, Serialize};
-use tokio::sync::mpsc::unbounded_channel;
+use tokio::sync::mpsc::{UnboundedSender, unbounded_channel};
 
 use crate::{
-    JocalEvent, JocalService, ReceiveDialog, ReceiveRequest,
+    JocalEvent, JocalService, Peers, ReceiveDialog, ReceiveRequest, Sessions,
     error::{LocalSendError, Result},
     models::{Device, FileMetadata},
 };
@@ -396,3 +396,64 @@ pub async fn handle_cancel(
 pub struct CancelParams {
     session_id: String,
 }
+
+async fn do_prepare_upload(
+    ourself: Device,
+    client: reqwest::Client,
+    tx: UnboundedSender<JocalEvent>,
+    peer: &str,
+    peers: Peers,
+    sessions: Sessions,
+    files: BTreeMap<String, FileMetadata>,
+) -> Result<PrepareUploadResponse> {
+    let Some((addr, device)) = peers.lock().await.get(peer).cloned() else {
+        return Err(LocalSendError::PeerNotFound);
+    };
+
+    log::debug!("preparing upload request");
+
+    let request = client
+        .post(format!(
+            "{}://{}/api/localsend/v2/prepare-upload",
+            device.protocol, addr
+        ))
+        .json(&PrepareUploadRequest {
+            info: ourself.clone(),
+            files: files.clone(),
+        })
+        .timeout(Duration::from_secs(30));
+
+    debug!("sending '{request:?}' to peer at {addr:?}");
+
+    // tokio::spawn(future);
+    let response = request.send().await?;
+
+    debug!("Response: {response:?}");
+
+    let response: PrepareUploadResponse = match response.json().await {
+        Err(e) => {
+            error!("got error deserializing response: {e:?}");
+            return Err(LocalSendError::RequestError(e));
+        }
+        Ok(r) => r,
+    };
+
+    debug!("decoded response: {response:?}");
+
+    let session = Session {
+        session_id: response.session_id.clone(),
+        files,
+        file_tokens: response.files.clone(),
+        receiver: device,
+        sender: ourself.clone(),
+        status: SessionStatus::Active,
+        addr,
+    };
+
+    sessions
+        .lock()
+        .await
+        .insert(response.session_id.clone(), session);
+
+    Ok(response)
+}