WIP: aquatic_ws: start trying to understand and document protocol

aquatic_ws/Cargo.toml

@@ -5,9 +5,9 @@ authors = ["Joakim Frostegård <joakim.frostegard@gmail.com>"]
 edition = "2018"
 license = "Apache-2.0"
 
-# [lib]
+[lib]
-# name = "aquatic_ws"
+name = "aquatic_ws"
-# path = "src/lib/lib.rs"
+path = "src/lib/lib.rs"
 
 [[bin]]
 name = "aquatic_ws"

aquatic_ws/src/bin/main.rs

@@ -1,216 +1,6 @@
-//! There is not much point in doing more work until more clarity on
+use aquatic_ws;
-//! exact protocol is achieved
-
-use std::net::SocketAddr;
-use std::time::{Duration, Instant};
-use std::io::ErrorKind;
-use std::option::Option;
-
-use slab::Slab;
-use tungstenite::{Message, WebSocket};
-
-use mio::{Events, Poll, Interest, Token};
-use mio::net::{TcpListener, TcpStream};
-
-
-pub struct PeerConnection {
-    pub ws: WebSocket<TcpStream>,
-    pub peer_socket_addr: SocketAddr,
-    pub valid_until: Instant,
-}
-
-
-/// First thoughts on what to send to handler
-pub struct HandlerMessage<T> {
-    /// Index of socket worker that read this request. Required for sending
-    /// back response through correct channel to correct worker.
-    pub socket_worker_index: usize,
-    /// FIXME: Should this be parsed request?
-    pub message: T, 
-    /// SocketAddr of peer
-    pub peer_socket_addr: SocketAddr,
-    /// Slab index of PeerConnection
-    pub peer_connection_index: usize,
-}
-
-
-fn run_network_worker(){
-    let address: SocketAddr = "0.0.0.0:3000".parse().unwrap();
-
-    let mut listener = TcpListener::bind(address).unwrap();
-    let mut poll = Poll::new().expect("create poll");
-
-    poll.registry()
-        .register(&mut listener, Token(0), Interest::READABLE)
-        .unwrap();
-
-    let mut events = Events::with_capacity(1024);
-
-    let timeout = Duration::from_millis(50);
-
-    let mut connections: Slab<Option<PeerConnection>> = Slab::new();
-
-    // Insert empty first entry to prevent assignment of index 0
-    assert_eq!(connections.insert(None), 0);
-
-    loop {
-        poll.poll(&mut events, Some(timeout))
-            .expect("failed polling");
-        
-        let valid_until = Instant::now() + Duration::from_secs(600);
-
-        for event in events.iter(){
-            let token = event.token();
-
-            if token.0 == 0 {
-                loop {
-                    match listener.accept(){
-                        Ok((mut stream, src)) => {
-                            let entry = connections.vacant_entry();
-                            let token = Token(entry.key());
-
-                            poll.registry()
-                                .register(&mut stream, token, Interest::READABLE)
-                                .unwrap();
-                            
-                            // FIXME: will this cause issues due to blocking?
-                            // Should handshake be started manually below
-                            // instead?
-                            let ws = tungstenite::server::accept(stream).unwrap();
-
-                            let peer_connection = PeerConnection {
-                                ws,
-                                peer_socket_addr: src,
-                                valid_until,
-                            };
-
-                            entry.insert(Some(peer_connection));
-                        },
-                        Err(err) => {
-                            if err.kind() == ErrorKind::WouldBlock {
-                                break
-                            }
-
-                            eprint!("{}", err);
-                        }
-                    }
-                }
-            } else if event.is_readable(){
-                loop {
-                    if let Some(Some(connection)) = connections.get_mut(token.0){
-                        match connection.ws.read_message(){
-                            Ok(message) => {
-                                // FIXME: parse message, send to handler
-                                // through channel (flume?)
-
-                                connection.valid_until = valid_until;
-                            },
-                            Err(tungstenite::Error::Io(err)) => {
-                                if err.kind() == ErrorKind::WouldBlock {
-                                    break
-                                }
-
-                                eprint!("{}", err);
-                            },
-                            Err(tungstenite::Error::ConnectionClosed) => {
-                                // FIXME: necessary?
-                                poll.registry().deregister(connection.ws.get_mut()).unwrap();
-
-                                connections.remove(token.0);
-                            },
-                            Err(err) => {
-                                eprint!("{}", err);
-                            }
-                        }
-                    }
-                }
-            }
-        }
-
-        let now = Instant::now();
-
-        // Close connections after some time of inactivity and write pending
-        // messages (which is required after closing anyway.)
-        //
-        // FIXME: peers need to be removed too, wherever they are stored
-        connections.retain(|_, opt_connection| {
-            if let Some(connection) = opt_connection {
-                if connection.valid_until < now {
-                    connection.ws.close(None).unwrap();
-                }
-
-                loop {
-                    match connection.ws.write_pending(){
-                        Err(tungstenite::Error::Io(err)) => {
-                            if err.kind() == ErrorKind::WouldBlock {
-                                break
-                            }
-                        },
-                        Err(tungstenite::Error::ConnectionClosed) => {
-                            // FIXME: necessary?
-                            poll.registry()
-                                .deregister(connection.ws.get_mut())
-                                .unwrap();
-
-                            return false;
-                        },
-                        _ => {}
-                    }
-                }
-            }
-
-            true
-        });
-
-        // TODO: loop through responses from channel, write them to wss or
-        // possibly register ws as writable (but this means event capacity
-        // must be adjusted accordingy and is limiting)
-
-        // How should IP's be handled? Send index and src to processing and
-        // lookup on return that entry is correct. Old ideas:
-        // Maybe use IndexMap<SocketAddr, WebSocket> and use numerical
-        // index for token? Removing element from IndexMap requires shifting
-        // or swapping indeces, so not very good.
-        for _ in 0..100 {
-            let connection_index = 1;
-            let peer_socket_addr: SocketAddr = "127.0.0.1:8080".parse().unwrap();
-            let message = Message::Text("test".to_string());
-
-            let opt_connection = connections
-                .get_mut(connection_index);
-
-            if let Some(Some(connection)) = opt_connection {
-                if connection.peer_socket_addr != peer_socket_addr {
-                    continue;
-                }
-
-                match connection.ws.write_message(message){
-                    Ok(()) => {},
-                    Err(tungstenite::Error::Io(err)) => {
-                        if err.kind() == ErrorKind::WouldBlock {
-                            continue;
-                        }
-
-                        eprint!("{}", err);
-                    },
-                    Err(tungstenite::Error::ConnectionClosed) => {
-                        // FIXME: necessary?
-                        poll.registry()
-                            .deregister(connection.ws.get_mut())
-                            .unwrap();
-
-                        connections.remove(connection_index);
-                    },
-                    Err(err) => {
-                        eprint!("{}", err);
-                    },
-                }
-            }
-        }
-    }
-}
 
 
 fn main(){
-    run_network_worker();
+    aquatic_ws::run_network_worker();
 }

aquatic_ws/src/lib/lib.rs

@@ -0,0 +1,213 @@
+//! There is not much point in doing more work until more clarity on
+//! exact protocol is achieved
+
+use std::net::{SocketAddr};
+use std::time::{Duration, Instant};
+use std::io::ErrorKind;
+use std::option::Option;
+
+use slab::Slab;
+use tungstenite::{Message, WebSocket};
+
+use mio::{Events, Poll, Interest, Token};
+use mio::net::{TcpListener, TcpStream};
+
+pub mod protocol;
+
+
+pub struct PeerConnection {
+    pub ws: WebSocket<TcpStream>,
+    pub peer_socket_addr: SocketAddr,
+    pub valid_until: Instant,
+}
+
+
+/// First thoughts on what to send to handler
+pub struct HandlerMessage<T> {
+    /// Index of socket worker that read this request. Required for sending
+    /// back response through correct channel to correct worker.
+    pub socket_worker_index: usize,
+    /// FIXME: Should this be parsed request?
+    pub message: T, 
+    /// SocketAddr of peer
+    pub peer_socket_addr: SocketAddr,
+    /// Slab index of PeerConnection
+    pub peer_connection_index: usize,
+}
+
+
+pub fn run_network_worker(){
+    let address: SocketAddr = "0.0.0.0:3000".parse().unwrap();
+
+    let mut listener = TcpListener::bind(address).unwrap();
+    let mut poll = Poll::new().expect("create poll");
+
+    poll.registry()
+        .register(&mut listener, Token(0), Interest::READABLE)
+        .unwrap();
+
+    let mut events = Events::with_capacity(1024);
+
+    let timeout = Duration::from_millis(50);
+
+    let mut connections: Slab<Option<PeerConnection>> = Slab::new();
+
+    // Insert empty first entry to prevent assignment of index 0
+    assert_eq!(connections.insert(None), 0);
+
+    loop {
+        poll.poll(&mut events, Some(timeout))
+            .expect("failed polling");
+        
+        let valid_until = Instant::now() + Duration::from_secs(600);
+
+        for event in events.iter(){
+            let token = event.token();
+
+            if token.0 == 0 {
+                loop {
+                    match listener.accept(){
+                        Ok((mut stream, src)) => {
+                            let entry = connections.vacant_entry();
+                            let token = Token(entry.key());
+
+                            poll.registry()
+                                .register(&mut stream, token, Interest::READABLE)
+                                .unwrap();
+                            
+                            // FIXME: will this cause issues due to blocking?
+                            // Should handshake be started manually below
+                            // instead?
+                            let ws = tungstenite::server::accept(stream).unwrap();
+
+                            let peer_connection = PeerConnection {
+                                ws,
+                                peer_socket_addr: src,
+                                valid_until,
+                            };
+
+                            entry.insert(Some(peer_connection));
+                        },
+                        Err(err) => {
+                            if err.kind() == ErrorKind::WouldBlock {
+                                break
+                            }
+
+                            eprint!("{}", err);
+                        }
+                    }
+                }
+            } else if event.is_readable(){
+                loop {
+                    if let Some(Some(connection)) = connections.get_mut(token.0){
+                        match connection.ws.read_message(){
+                            Ok(message) => {
+                                // FIXME: parse message, send to handler
+                                // through channel (flume?)
+
+                                connection.valid_until = valid_until;
+                            },
+                            Err(tungstenite::Error::Io(err)) => {
+                                if err.kind() == ErrorKind::WouldBlock {
+                                    break
+                                }
+
+                                eprint!("{}", err);
+                            },
+                            Err(tungstenite::Error::ConnectionClosed) => {
+                                // FIXME: necessary?
+                                poll.registry().deregister(connection.ws.get_mut()).unwrap();
+
+                                connections.remove(token.0);
+                            },
+                            Err(err) => {
+                                eprint!("{}", err);
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        let now = Instant::now();
+
+        // Close connections after some time of inactivity and write pending
+        // messages (which is required after closing anyway.)
+        //
+        // FIXME: peers need to be removed too, wherever they are stored
+        connections.retain(|_, opt_connection| {
+            if let Some(connection) = opt_connection {
+                if connection.valid_until < now {
+                    connection.ws.close(None).unwrap();
+                }
+
+                loop {
+                    match connection.ws.write_pending(){
+                        Err(tungstenite::Error::Io(err)) => {
+                            if err.kind() == ErrorKind::WouldBlock {
+                                break
+                            }
+                        },
+                        Err(tungstenite::Error::ConnectionClosed) => {
+                            // FIXME: necessary?
+                            poll.registry()
+                                .deregister(connection.ws.get_mut())
+                                .unwrap();
+
+                            return false;
+                        },
+                        _ => {}
+                    }
+                }
+            }
+
+            true
+        });
+
+        // TODO: loop through responses from channel, write them to wss or
+        // possibly register ws as writable (but this means event capacity
+        // must be adjusted accordingy and is limiting)
+
+        // How should IP's be handled? Send index and src to processing and
+        // lookup on return that entry is correct. Old ideas:
+        // Maybe use IndexMap<SocketAddr, WebSocket> and use numerical
+        // index for token? Removing element from IndexMap requires shifting
+        // or swapping indeces, so not very good.
+        for _ in 0..100 {
+            let connection_index = 1;
+            let peer_socket_addr: SocketAddr = "127.0.0.1:8080".parse().unwrap();
+            let message = Message::Text("test".to_string());
+
+            let opt_connection = connections
+                .get_mut(connection_index);
+
+            if let Some(Some(connection)) = opt_connection {
+                if connection.peer_socket_addr != peer_socket_addr {
+                    continue;
+                }
+
+                match connection.ws.write_message(message){
+                    Ok(()) => {},
+                    Err(tungstenite::Error::Io(err)) => {
+                        if err.kind() == ErrorKind::WouldBlock {
+                            continue;
+                        }
+
+                        eprint!("{}", err);
+                    },
+                    Err(tungstenite::Error::ConnectionClosed) => {
+                        // FIXME: necessary?
+                        poll.registry()
+                            .deregister(connection.ws.get_mut())
+                            .unwrap();
+
+                        connections.remove(connection_index);
+                    },
+                    Err(err) => {
+                        eprint!("{}", err);
+                    },
+                }
+            }
+        }
+    }
+}

aquatic_ws/src/lib/protocol.rs

@@ -0,0 +1,108 @@
+use std::collections::HashMap;
+use std::net::IpAddr;
+
+
+/// TODO: will need to store socket worker index and connection index
+/// for middleman activities, also save SocketAddr instead of IP and port,
+/// maybe, for comparison in socket worker
+pub struct Peer {
+    pub complete: bool, // bytes_left == 0
+    pub peer_id: [u8; 20],
+    pub ip: IpAddr, // From src socket addr
+    pub port: u16, // From src port
+}
+
+
+pub enum AnnounceEvent {
+    Started,
+    Stopped,
+    Completed,
+    Update
+}
+
+
+/// Apparently, these are sent to a number of peers when they are set
+/// in an AnnounceRequest
+/// action = "announce"
+pub struct MiddlemanOfferToPeer {
+    pub offer: (), // Gets copied from AnnounceRequestOffer
+    pub offer_id: (), // Gets copied from AnnounceRequestOffer
+    pub peer_id: [u8; 20], // Peer id of peer sending offer
+    pub info_hash: [u8; 20],
+}
+
+
+/// If announce request has answer = true, send this to peer with
+/// peer id == "to_peer_id" field
+/// Action field should be 'announce'
+pub struct MiddlemanAnswerToPeer {
+    pub answer: bool,
+    pub offer_id: (),
+    pub peer_id: [u8; 20],
+    pub info_hash: [u8; 20],
+}
+
+
+/// Element of AnnounceRequest.offers
+pub struct AnnounceRequestOffer {
+    pub offer: (), // TODO: Check client for what this is
+    pub offer_id: (), // TODO: Check client for what this is
+}
+
+
+pub struct AnnounceRequest {
+    pub info_hash: [u8; 20], // FIXME: I think these are actually really just strings with 20 len, same with peer id
+    pub peer_id: [u8; 20],
+    pub bytes_left: bool, // Just called "left" in protocol
+    pub event: AnnounceEvent, // Can be empty? Then, default is "update"
+
+    // Length of this is number of peers wanted?
+    // Only when this is an array offers are sent
+    pub offers: Option<Vec<AnnounceRequestOffer>>, 
+
+    /// If false, send response before sending offers (or possibly "skip sending update back"?)
+    /// If true, send MiddlemanAnswerToPeer to peer with "to_peer_id" as peer_id.
+    pub answer: bool, 
+    pub to_peer_id: Option<[u8; 20]>, // Only parsed to hex if answer == true, probably undefined otherwise
+}
+
+
+pub struct AnnounceResponse {
+    pub info_hash: [u8; 20],
+    pub complete: usize,
+    pub incomplete: usize,
+    // I suspect receivers don't care about this and rely on offers instead??
+    // Also, what does it contain, exacly?
+    pub peers: Vec<()>,
+    pub interval: usize, // Default 2 min probably
+
+    // Sent to "to_peer_id" peer (?? or did I put this into MiddlemanAnswerToPeer instead?)
+    pub offer_id: (),
+    pub answer: bool,
+}
+
+
+
+pub struct ScrapeRequest {
+    // If omitted, scrape for all torrents, apparently
+    // There is some kind of parsing here too which accepts a single info hash
+    // and puts it into a vector
+    pub info_hashes: Option<Vec<[u8; 20]>>,
+}
+
+
+pub struct ScrapeStatistics {
+    pub complete: usize,
+    pub incomplete: usize,
+    pub downloaded: usize,
+}
+
+
+pub struct ScrapeResponse {
+    pub files: HashMap<[u8; 20], ScrapeStatistics>, // InfoHash to Scrape stats
+    pub flags: HashMap<String, usize>,
+}
+//pub struct ScrapeResponse {
+//    pub complete: usize,
+//    pub incomplete: usize,
+//}