aquatic_ws: clean torrent state periodically

TODO.md

@@ -6,7 +6,6 @@
   * handshake: deregister stream when applicable
 * test
   * test full torrent transfer (offer-answer exchange)
-* torrent state cleaning
 * log crate instead of println/eprintln?
 * privdrop
 * config

aquatic_ws/src/lib/common.rs

@@ -82,6 +82,7 @@ impl Default for TorrentData {
 pub type TorrentMap = HashMap<InfoHash, TorrentData>;
 
 
+#[derive(Clone)]
 pub struct State {
     pub torrents: Arc<Mutex<TorrentMap>>,
 }

aquatic_ws/src/lib/config.rs

@@ -49,7 +49,7 @@ pub struct HandlerConfig {
 #[serde(default)]
 pub struct CleaningConfig {
     /// Clean peers this often (seconds)
-    pub interval: u64, // FIXME: implement
+    pub interval: u64,
     /// Remove peers that haven't announced for this long (seconds)
     pub max_peer_age: u64,
     /// Remove connections that are older than this (seconds)

aquatic_ws/src/lib/lib.rs

@@ -1,11 +1,14 @@
 //! There is not much point in doing more work until more clarity on
 //! exact protocol is achieved
 
+use std::time::Duration;
+
 pub mod common;
 pub mod config;
 pub mod handler;
 pub mod network;
 pub mod protocol;
+pub mod tasks;
 
 use common::*;
 use config::Config;
@@ -38,16 +41,23 @@ pub fn run(config: Config){
 
     let out_message_sender = OutMessageSender::new(out_message_senders);
 
-    ::std::thread::spawn(move || {
-        handler::run_request_worker(
-            config,
-            state,
-            in_message_receiver,
-            out_message_sender,
-        );
-    });
+    {
+        let config = config.clone();
+        let state = state.clone();
+
+        ::std::thread::spawn(move || {
+            handler::run_request_worker(
+                config,
+                state,
+                in_message_receiver,
+                out_message_sender,
+            );
+        });
+    }
 
     loop {
-        ::std::thread::sleep(::std::time::Duration::from_secs(60));
+        ::std::thread::sleep(Duration::from_secs(config.cleaning.interval));
+
+        tasks::clean_torrents(&state);
     }
 }

aquatic_ws/src/lib/tasks.rs

@@ -0,0 +1,20 @@
+use std::time::Instant;
+
+use crate::common::*;
+
+
+pub fn clean_torrents(state: &State){
+    let mut torrents = state.torrents.lock();
+
+    let now = Instant::now();
+
+    torrents.retain(|_, torrent_data| {
+        torrent_data.peers.retain(|_, peer| {
+            peer.valid_until.0 >= now
+        });
+
+        !torrent_data.peers.is_empty()
+    });
+
+    torrents.shrink_to_fit();
+}