initial tokio 1.0 commit

6 files changed, 201 insertions, 82 deletions

diff --git a/Cargo.lock b/Cargo.lock
index bdd8329..76b92c1 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -38,6 +38,27 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f8ebf5827e4ac4fd5946560e6a99776ea73b596d80898f357007317a7141e47"
 
 [[package]]
+name = "async-stream"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3670df70cbc01729f901f94c887814b3c68db038aad1329a418bae178bc5295c"
+dependencies = [
+ "async-stream-impl",
+ "futures-core",
+]
+
+[[package]]
+name = "async-stream-impl"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a3548b8efc9f8e8a5a0a2808c5bd8451a9031b9e5b879a79590304ae928b0a70"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
 name = "atty"
 version = "0.2.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -72,7 +93,7 @@ checksum = "99de13bb6361e01e493b3db7928085dcc474b7ba4f5481818e53a89d76b8393f"
 dependencies = [
  "bitflags",
  "cexpr",
- "cfg-if",
+ "cfg-if 0.1.10",
  "clang-sys",
  "lazy_static",
  "lazycell",
@@ -104,9 +125,9 @@ checksum = "08c48aae112d48ed9f069b33538ea9e3e90aa263cfa3d1c24309612b1f7472de"
 
 [[package]]
 name = "bytes"
-version = "0.5.6"
+version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0e4cec68f03f32e44924783795810fa50a7035d8c8ebe78580ad7e6c703fba38"
+checksum = "ad1f8e949d755f9d79112b5bb46938e0ef9d3804a0b16dfab13aafcaa5f0fa72"
 
 [[package]]
 name = "cc"
@@ -130,6 +151,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4785bdd1c96b2a846b2bd7cc02e86b6b3dbf14e7e53446c4f54c92a361040822"
 
 [[package]]
+name = "cfg-if"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
+
+[[package]]
 name = "chrono"
 version = "0.4.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -267,7 +294,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c3c7c73a2d1e9fc0886a08b93e98eb643461230d5f1925e4036204d5f2e261a8"
 dependencies = [
  "autocfg",
- "cfg-if",
+ "cfg-if 0.1.10",
  "lazy_static",
 ]
 
@@ -537,7 +564,7 @@ version = "0.1.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fc587bc0ec293155d5bfa6b9891ec18a1e330c234f896ea47fbada4cadbe47e6"
 dependencies = [
- "cfg-if",
+ "cfg-if 0.1.10",
  "libc",
  "wasi 0.9.0+wasi-snapshot-preview1",
 ]
@@ -632,7 +659,7 @@ dependencies = [
  "fnv",
  "lazy_static",
  "libc",
- "mio",
+ "mio 0.6.22",
  "rand",
  "serde",
  "tempfile",
@@ -729,7 +756,7 @@ version = "0.4.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4fabed175da42fed1fa0746b0ea71f412aa9d35e76e95e59b192c64b9dc2bf8b"
 dependencies = [
- "cfg-if",
+ "cfg-if 0.1.10",
 ]
 
 [[package]]
@@ -759,20 +786,33 @@ version = "0.6.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fce347092656428bc8eaf6201042cb551b8d67855af7374542a92a0fbfcac430"
 dependencies = [
- "cfg-if",
+ "cfg-if 0.1.10",
  "fuchsia-zircon",
  "fuchsia-zircon-sys",
  "iovec",
  "kernel32-sys",
  "libc",
  "log",
- "miow",
+ "miow 0.2.1",
  "net2",
  "slab",
  "winapi 0.2.8",
 ]
 
 [[package]]
+name = "mio"
+version = "0.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f33bc887064ef1fd66020c9adfc45bb9f33d75a42096c81e7c56c65b75dd1a8b"
+dependencies = [
+ "libc",
+ "log",
+ "miow 0.3.6",
+ "ntapi",
+ "winapi 0.3.9",
+]
+
+[[package]]
 name = "miow"
 version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -785,10 +825,18 @@ dependencies = [
 ]
 
 [[package]]
+name = "miow"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a33c1b55807fbed163481b5ba66db4b2fa6cde694a5027be10fb724206c5897"
+dependencies = [
+ "socket2",
+ "winapi 0.3.9",
+]
+
+[[package]]
 name = "mumble-protocol"
 version = "0.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4954adf2418ebc052d723997cf3e090729132d59a795113db931aa5811e46542"
 dependencies = [
  "byteorder",
  "bytes",
@@ -831,7 +879,7 @@ dependencies = [
  "opus",
  "serde",
  "tokio",
- "tokio-tls",
+ "tokio-native-tls",
  "tokio-util",
 ]
 
@@ -870,7 +918,7 @@ version = "0.2.35"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3ebc3ec692ed7c9a255596c67808dee269f64655d8baf7b4f0638e51ba1d6853"
 dependencies = [
- "cfg-if",
+ "cfg-if 0.1.10",
  "libc",
  "winapi 0.3.9",
 ]
@@ -883,7 +931,7 @@ checksum = "3b2e0b4f3320ed72aaedb9a5ac838690a8047c7b275da22711fddff4f8a14229"
 dependencies = [
  "bitflags",
  "cc",
- "cfg-if",
+ "cfg-if 0.1.10",
  "libc",
  "void",
 ]
@@ -899,6 +947,15 @@ dependencies = [
 ]
 
 [[package]]
+name = "ntapi"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3f6bb902e437b6d86e03cce10a7e2af662292c5dfef23b65899ea3ac9354ad44"
+dependencies = [
+ "winapi 0.3.9",
+]
+
+[[package]]
 name = "num-integer"
 version = "0.1.43"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -918,6 +975,16 @@ dependencies = [
 ]
 
 [[package]]
+name = "num_cpus"
+version = "1.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "05499f3756671c15885fee9034446956fff3f243d6077b91e5767df161f766b3"
+dependencies = [
+ "hermit-abi",
+ "libc",
+]
+
+[[package]]
 name = "once_cell"
 version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -930,7 +997,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8d575eff3665419f9b83678ff2815858ad9d11567e082f5ac1814baba4e2bcb4"
 dependencies = [
  "bitflags",
- "cfg-if",
+ "cfg-if 0.1.10",
  "foreign-types",
  "lazy_static",
  "libc",
@@ -993,7 +1060,7 @@ version = "0.6.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b876b1b9e7ac6e1a74a6da34d25c42e17e8862aa409cbbbdcfc8d86c6f3bc62b"
 dependencies = [
- "cfg-if",
+ "cfg-if 0.1.10",
  "cloudabi",
  "libc",
  "redox_syscall",
@@ -1030,9 +1097,9 @@ dependencies = [
 
 [[package]]
 name = "pin-project-lite"
-version = "0.1.11"
+version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c917123afa01924fc84bb20c4c03f004d9c38e5127e3c039bbf7f4b9c76a2f6b"
+checksum = "6b063f57ec186e6140e2b8b6921e5f1bd89c7356dda5b33acc5401203ca6131c"
 
 [[package]]
 name = "pin-utils"
@@ -1289,6 +1356,17 @@ dependencies = [
 ]
 
 [[package]]
+name = "socket2"
+version = "0.3.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "122e570113d28d773067fab24266b66753f6ea915758651696b6e35e49f88d6e"
+dependencies = [
+ "cfg-if 1.0.0",
+ "libc",
+ "winapi 0.3.9",
+]
+
+[[package]]
 name = "stdweb"
 version = "0.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1317,7 +1395,7 @@ version = "3.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7a6e24d9338a0a5be79593e2fa15a648add6138caa803e2d5bc782c371732ca9"
 dependencies = [
- "cfg-if",
+ "cfg-if 0.1.10",
  "libc",
  "rand",
  "redox_syscall",
@@ -1367,26 +1445,23 @@ dependencies = [
 
 [[package]]
 name = "tokio"
-version = "0.2.22"
+version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5d34ca54d84bf2b5b4d7d31e901a8464f7b60ac145a284fba25ceb801f2ddccd"
+checksum = "9f4bfdcbd00fa893ac0549b38aa27080636a0104b0d0c38475a99439405e1df8"
 dependencies = [
- "bytes",
- "fnv",
- "futures-core",
- "iovec",
- "lazy_static",
- "mio",
+ "autocfg",
+ "libc",
+ "mio 0.7.6",
+ "num_cpus",
  "pin-project-lite",
- "slab",
  "tokio-macros",
 ]
 
 [[package]]
 name = "tokio-macros"
-version = "0.2.5"
+version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f0c3acc6aa564495a0f2e1d59fab677cd7f81a19994cfc7f3ad0e64301560389"
+checksum = "42517d2975ca3114b22a16192634e8241dc5cc1f130be194645970cc1c371494"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1394,20 +1469,32 @@ dependencies = [
 ]
 
 [[package]]
-name = "tokio-tls"
-version = "0.3.1"
+name = "tokio-native-tls"
+version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9a70f4fcd7b3b24fb194f837560168208f669ca8cb70d0c4b862944452396343"
+checksum = "f7d995660bd2b7f8c1568414c1126076c13fbb725c40112dc0120b78eb9b717b"
 dependencies = [
  "native-tls",
  "tokio",
 ]
 
 [[package]]
+name = "tokio-stream"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3f3be913b74b13210c8fe04b17ab833f5a124f45b93d0f99f59fff621f64392a"
+dependencies = [
+ "async-stream",
+ "futures-core",
+ "pin-project-lite",
+ "tokio",
+]
+
+[[package]]
 name = "tokio-util"
-version = "0.3.1"
+version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "be8242891f2b6cbef26a2d7e8605133c2c554cd35b3e4948ea892d6d68436499"
+checksum = "36135b7e7da911f5f8b9331209f7fab4cc13498f3fff52f72a710c78187e3148"
 dependencies = [
  "bytes",
  "futures-core",
@@ -1415,6 +1502,7 @@ dependencies = [
  "log",
  "pin-project-lite",
  "tokio",
+ "tokio-stream",
 ]
 
 [[package]]
@@ -1489,7 +1577,7 @@ version = "0.2.68"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1ac64ead5ea5f05873d7c12b545865ca2b8d28adfc50a49b84770a3a97265d42"
 dependencies = [
- "cfg-if",
+ "cfg-if 0.1.10",
  "wasm-bindgen-macro",
 ]
 
diff --git a/mumd/Cargo.toml b/mumd/Cargo.toml
index 39b9d26..08ebfca 100644
--- a/mumd/Cargo.toml
+++ b/mumd/Cargo.toml
@@ -17,7 +17,7 @@ notifications = ["libnotify"]
 mumlib = { path = "../mumlib" }
 
 argparse = "0.2"
-bytes = "0.5"
+bytes = "1.0"
 cpal = { git = "https://github.com/RustAudio/cpal" }
 dasp_interpolate = { version = "0.11", features = ["linear"] }
 dasp_signal = "0.11"
@@ -26,14 +26,14 @@ futures-util = "0.3"
 hound = "3.4"
 ipc-channel = "0.14"
 log = "0.4"
-mumble-protocol = "0.3.1"
+mumble-protocol = { path="../../rust-mumble-protocol" }
 native-tls = "0.2"
 openssl = { version = "0.10" }
 opus = "0.2"
 serde = { version = "1.0", features = ["derive"] }
-tokio = { version = "0.2", features = ["blocking", "macros", "rt-core", "sync", "tcp", "time"] }
-tokio-tls = "0.3"
-tokio-util = { version = "0.3", features = ["codec", "udp"] }
+tokio = { version = "1.0", features = ["macros", "rt", "rt-multi-thread", "sync", "net", "time"] }
+tokio-native-tls = "0.3"
+tokio-util = { version = "0.6", features = ["codec", "net"] }
 
 libnotify = { version = "1.0", optional = true }
 
diff --git a/mumd/src/audio.rs b/mumd/src/audio.rs
index 812bb4c..9e8bd6e 100644
--- a/mumd/src/audio.rs
+++ b/mumd/src/audio.rs
@@ -305,11 +305,11 @@ impl Audio {
     }
 
     pub fn set_input_volume(&self, input_volume: f32) {
-        self.input_volume_sender.broadcast(input_volume).unwrap();
+        self.input_volume_sender.send(input_volume).unwrap();
     }
 
     pub fn set_output_volume(&self, output_volume: f32) {
-        self.output_volume_sender.broadcast(output_volume).unwrap();
+        self.output_volume_sender.send(output_volume).unwrap();
     }
 
     pub fn set_user_volume(&self, id: u32, volume: f32) {
diff --git a/mumd/src/network/tcp.rs b/mumd/src/network/tcp.rs
index b1743c8..9024ef3 100644
--- a/mumd/src/network/tcp.rs
+++ b/mumd/src/network/tcp.rs
@@ -17,8 +17,9 @@ use std::sync::{Arc, Mutex};
 use tokio::net::TcpStream;
 use tokio::sync::{mpsc, oneshot, watch};
 use tokio::time::{self, Duration};
-use tokio_tls::{TlsConnector, TlsStream};
+use tokio_native_tls::{TlsConnector, TlsStream};
 use tokio_util::codec::{Decoder, Framed};
+// use tokio_util::codec::decoder::Decoder;
 
 type TcpSender = SplitSink<
     Framed<TlsStream<TcpStream>, ControlCodec<Serverbound, Clientbound>>,
@@ -48,16 +49,22 @@ pub async fn handle(
     mut tcp_event_register_receiver: mpsc::UnboundedReceiver<(TcpEvent, TcpEventCallback)>,
 ) {
     loop {
-        let connection_info = loop {
-            match connection_info_receiver.recv().await {
-                None => {
-                    return;
-                }
-                Some(None) => {}
-                Some(Some(connection_info)) => {
-                    break connection_info;
+        let connection_info = 'data: loop {
+            while let Ok(()) = connection_info_receiver.changed().await {
+                if let Some(data) = connection_info_receiver.borrow().clone() {
+                    break 'data data;
                 }
             }
+            return;
+            // match connection_info_receiver.changed().await {
+            //     None => {
+            //         return;
+            //     }
+            //     Some(None) => {}
+            //     Some(Some(connection_info)) => {
+            //         break connection_info;
+            //     }
+            // }
         };
         let (mut sink, stream) = connect(
             connection_info.socket_addr,
@@ -165,7 +172,7 @@ async fn send_packets(
         },
         || async {
             //clears queue of remaining packets
-            while packet_receiver.borrow_mut().try_recv().is_ok() {}
+            // while packet_receiver.borrow_mut().try_recv().is_ok() {}
 
             sink.borrow_mut().close().await.unwrap();
         },
@@ -323,10 +330,16 @@ async fn run_until_disconnection<T, F, G, H>(
 {
     let (tx, rx) = oneshot::channel();
     let phase_transition_block = async {
-        while !matches!(
-            phase_watcher.recv().await.unwrap(),
-            StatePhase::Disconnected
-        ) {}
+        loop {
+            phase_watcher.changed().await.unwrap();
+            if matches!(*phase_watcher.borrow(), StatePhase::Disconnected) {
+                break;
+            }
+        }
+        // while !matches!(
+        //     phase_watcher.recv().await.unwrap(),
+        //     StatePhase::Disconnected
+        // ) {}
         tx.send(true).unwrap();
     };
 
diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs
index b1c202a..55a47a3 100644
--- a/mumd/src/network/udp.rs
+++ b/mumd/src/network/udp.rs
@@ -29,16 +29,22 @@ pub async fn handle(
     let mut receiver = state.lock().unwrap().audio_mut().take_receiver().unwrap();
 
     loop {
-        let connection_info = loop {
-            match connection_info_receiver.recv().await {
-                None => {
-                    return;
-                }
-                Some(None) => {}
-                Some(Some(connection_info)) => {
-                    break connection_info;
+        let connection_info = 'data: loop {
+            while let Ok(()) = connection_info_receiver.changed().await {
+                if let Some(data) = connection_info_receiver.borrow().clone() {
+                    break 'data data;
                 }
             }
+            return;
+            // match connection_info_receiver.recv().await {
+            //     None => {
+            //         return;
+            //     }
+            //     Some(None) => {}
+            //     Some(Some(connection_info)) => {
+            //         break connection_info;
+            //     }
+            // }
         };
         let (mut sink, source) = connect(&mut crypt_state_receiver).await;
 
@@ -114,10 +120,16 @@ async fn listen(
 ) {
     let (tx, rx) = oneshot::channel();
     let phase_transition_block = async {
-        while !matches!(
-            phase_watcher.recv().await.unwrap(),
-            StatePhase::Disconnected
-        ) {}
+        loop {
+            phase_watcher.changed().await.unwrap();
+            if matches!(*phase_watcher.borrow(), StatePhase::Disconnected) {
+                break;
+            }
+        }
+        // while !matches!(
+        //     phase_watcher.recv().await.unwrap(),
+        //     StatePhase::Disconnected
+        // ) {}
         tx.send(true).unwrap();
     };
 
@@ -203,10 +215,16 @@ async fn send_voice(
 ) {
     let (tx, rx) = oneshot::channel();
     let phase_transition_block = async {
-        while !matches!(
-            phase_watcher.recv().await.unwrap(),
-            StatePhase::Disconnected
-        ) {}
+        loop {
+            phase_watcher.changed().await.unwrap();
+            if matches!(*phase_watcher.borrow(), StatePhase::Disconnected) {
+                break;
+            }
+        }
+        // while !matches!(
+        //     phase_watcher.recv().await.unwrap(),
+        //     StatePhase::Disconnected
+        // ) {}
         tx.send(true).unwrap();
     };
 
@@ -261,11 +279,11 @@ pub async fn handle_pings(
         Box<dyn FnOnce(PongPacket)>,
     )>,
 ) {
-    let udp_socket = UdpSocket::bind((Ipv6Addr::from(0u128), 0u16))
+    let udp_socket = Arc::new(UdpSocket::bind((Ipv6Addr::from(0u128), 0u16))
         .await
-        .expect("Failed to bind UDP socket");
+        .expect("Failed to bind UDP socket"));
 
-    let (mut receiver, mut sender) = udp_socket.split();
+    // let (mut receiver, mut sender) = udp_socket.split();
 
     let pending = Rc::new(Mutex::new(HashMap::new()));
 
@@ -273,14 +291,14 @@ pub async fn handle_pings(
         while let Some((id, socket_addr, handle)) = ping_request_receiver.recv().await {
             let packet = PingPacket { id };
             let packet: [u8; 12] = packet.into();
-            sender.send_to(&packet, &socket_addr).await.unwrap();
+            udp_socket.send_to(&packet, &socket_addr).await.unwrap();
             pending.lock().unwrap().insert(id, handle);
         }
     };
 
     let receiver_handle = async {
         let mut buf = vec![0; 24];
-        while let Ok(read) = receiver.recv(&mut buf).await {
+        while let Ok(read) = udp_socket.recv(&mut buf).await {
             assert_eq!(read, 24);
 
             let packet = PongPacket::try_from(buf.as_slice()).unwrap();
diff --git a/mumd/src/state.rs b/mumd/src/state.rs
index aba0931..85e5449 100644
--- a/mumd/src/state.rs
+++ b/mumd/src/state.rs
@@ -321,7 +321,7 @@ impl State {
                 self.server = Some(server);
                 self.phase_watcher
                     .0
-                    .broadcast(StatePhase::Connecting)
+                    .send(StatePhase::Connecting)
                     .unwrap();
 
                 let socket_addr = match (host.as_ref(), port)
@@ -335,7 +335,7 @@ impl State {
                     }
                 };
                 self.connection_info_sender
-                    .broadcast(Some(ConnectionInfo::new(
+                    .send(Some(ConnectionInfo::new(
                         socket_addr,
                         host,
                         accept_invalid_cert,
@@ -366,7 +366,7 @@ impl State {
 
                 self.phase_watcher
                     .0
-                    .broadcast(StatePhase::Disconnected)
+                    .send(StatePhase::Disconnected)
                     .unwrap();
                 self.audio.play_effect(NotificationEvents::ServerDisconnect);
                 now!(Ok(None))
@@ -581,7 +581,7 @@ impl State {
     pub fn initialized(&self) {
         self.phase_watcher
             .0
-            .broadcast(StatePhase::Connected)
+            .send(StatePhase::Connected)
             .unwrap();
         self.audio.play_effect(NotificationEvents::ServerConnect);
     }