Merge pull request #77 from michel/lib/send-upload-speed

feat(lib): report each finished upload's rate to the server

soulseek-rs-lib/src/client/uploads.rs

@@ -1,7 +1,9 @@
 use super::{
     ActiveUpload, Arc, Client, ClientContext, DownloadStatus, PeerMessage,
-    RwLock, RwLockExt, collect_failed_tokens, error, next_upload_token, thread,
+    RwLock, RwLockExt, ServerMessage, collect_failed_tokens, error,
+    next_upload_token, thread,
 };
+use crate::message::server::MessageFactory;
 use crate::types::UploadStatus;
 use std::sync::atomic::{AtomicBool, AtomicU64};
 
@@ -156,6 +158,7 @@ impl Client {
                     UploadStatus::Failed(e.to_string())
                 }
             };
+            let mut report = None;
             if let Ok(mut ctx) = context.write_safe()
                 && let Some(upload) = ctx.active_uploads.get_mut(&token)
             {
@@ -164,9 +167,18 @@ impl Client {
                 if let Some(bytes) = streamed
                     && secs > 0.0
                 {
-                    ctx.last_upload_speed = (bytes as f64 / secs) as u32;
+                    let speed = (bytes as f64 / secs) as u32;
+                    ctx.last_upload_speed = speed;
+                    report = ctx.server_sender.clone().map(|s| (s, speed));
                 }
             }
+            // The server folds each finished upload's rate into the average
+            // speed it shows other users for us.
+            if let Some((server, speed)) = report {
+                let _ = server.send(ServerMessage::SendMessage(
+                    MessageFactory::build_send_upload_speed(speed),
+                ));
+            }
             // The slot this transfer held is free now, so whoever is next in
             // line gets it without waiting for another request to arrive.
             Self::pump_upload_queue(&context);

soulseek-rs-lib/src/message/mod.rs

@@ -252,6 +252,7 @@ impl Message {
                 100 => Ok("AcceptChildren"),
                 102 => Ok("PossibleParents"),
                 104 => Ok("WishlistInterval"),
+                121 => Ok("SendUploadSpeed"),
                 160 => Ok("ExcludedSearchPhrases"),
                 1001 => Ok("CantConnectToPeer"),
                 _ => Err(Error(format!("Unknown server message code: {code}"))),

soulseek-rs-lib/src/message/server/message_factory.rs

@@ -143,6 +143,16 @@ impl MessageFactory {
         Message::new().write_int32(7).write_string(username).clone()
     }
 
+    /// Tell the server (code 121) what a finished upload averaged, in bytes
+    /// per second; it folds that into the speed other users see for us.
+    #[must_use]
+    pub fn build_send_upload_speed(bytes_per_second: u32) -> Message {
+        Message::new()
+            .write_int32(121)
+            .write_int32(bytes_per_second)
+            .clone()
+    }
+
     /// Ask the server (code 36) for a user's share statistics.
     #[must_use]
     pub fn build_get_user_stats(username: &str) -> Message {
@@ -474,3 +484,12 @@ fn test_build_unwatch_user() {
     let expect: Vec<u8> = [6, 0, 0, 0, 3, 0, 0, 0, b'b', b'o', b'b'].to_vec();
     assert_eq!(expect, message.get_data());
 }
+
+#[test]
+fn an_upload_speed_report_carries_the_rate() {
+    let message = MessageFactory::build_send_upload_speed(900);
+    let mut decoded = Message::new_with_data(message.get_buffer());
+    assert_eq!(decoded.get_message_code(), 121);
+    decoded.set_pointer(8);
+    assert_eq!(decoded.read_int32(), 900);
+}

soulseek-rs-lib/tests/e2e.rs

@@ -1769,6 +1769,22 @@ fn two_real_clients_search_and_download() {
         "a finished upload sets the advertised speed"
     );
 
+    // The server hears about it too (SendUploadSpeed), which is where every
+    // other client's `user` lookup and the server's own ranking read it.
+    let deadline = Instant::now() + Duration::from_secs(10);
+    let mut recorded = 0;
+    while Instant::now() < deadline && recorded == 0 {
+        leecher
+            .request_user_info("e2e_sharer")
+            .expect("ask about the sharer");
+        std::thread::sleep(Duration::from_millis(250));
+        recorded = leecher
+            .user_info("e2e_sharer")
+            .and_then(|info| info.stats)
+            .map_or(0, |stats| stats.average_speed);
+    }
+    assert!(recorded > 0, "the server should record the upload's speed");
+
     let _ = std::fs::remove_dir_all(share_dir);
     let _ = std::fs::remove_dir_all(download_dir);
 }