fix tokio conflict (#168)

* fix tokio conflict * fix tests Co-authored-by: Tim Wilson <[email protected]>
tjwilson90 · Mar 8, 2021 · c2fcc91 · c2fcc91
1 parent e90dbcc
commit c2fcc91
Show file tree

Hide file tree

Showing 7 changed files with 49 additions and 66 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/server/Cargo.toml b/server/Cargo.toml
@@ -7,6 +7,7 @@ edition = "2018"
 [dependencies]
 base64 = "*"
 env_logger = { version = "*", default-features = false, features = ["humantime"] }
+futures-util = "*"
 http = "*"
 log = "*"
 once_cell = "*"
@@ -20,7 +21,7 @@ rusqlite = { version = "*", features = ["bundled"] }
 serde = { version = "*", features = ["derive"] }
 serde_json = "*"
 thiserror = "*"
-tokio = { version = "*", features = ["blocking", "macros", "parking_lot", "rt-threaded", "stream", "sync", "time"] }
+tokio = { version = "1", features = ["macros", "parking_lot", "rt-multi-thread", "sync", "time"] }
 tokio-stream = "*"
 turbo-hearts-api = { path = "../api" }
 turbo-hearts-bot = { path = "../bot" }

diff --git a/server/src/bot.rs b/server/src/bot.rs
@@ -1,13 +1,10 @@
 use crate::{CardsError, Games};
+use futures_util::FutureExt;
 use log::debug;
 use rand::distributions::Distribution;
 use rand_distr::Gamma;
 use std::time::Instant;
-use tokio::{
-    sync::mpsc::{error::TryRecvError, UnboundedReceiver},
-    time,
-    time::Duration,
-};
+use tokio::{sync::mpsc::UnboundedReceiver, time, time::Duration};
 use turbo_hearts_api::{
     can_claim, BotState, BotStrategy, Card, Cards, GameEvent, GameId, GameState, Seat, UserId,
 };
@@ -58,12 +55,12 @@ impl BotRunner {
         loop {
             let now = Instant::now();
             loop {
-                match rx.try_recv() {
-                    Ok((event, _)) => {
+                match rx.recv().now_or_never() {
+                    Some(Some((event, _))) => {
                         action = self.handle(event);
                     }
-                    Err(TryRecvError::Empty) => break,
-                    Err(TryRecvError::Closed) => return Ok(()),
+                    Some(None) => return Ok(()),
+                    None => break,
                 }
             }
             let delay =
@@ -165,7 +162,7 @@ impl BotRunner {
     async fn delay(delay: Option<Duration>, start: Instant) {
         let delay = delay.and_then(|delay| delay.checked_sub(start.elapsed()));
         if let Some(delay) = delay {
-            time::delay_for(delay).await;
+            time::sleep(delay).await;
         }
     }
 

diff --git a/server/src/game_endpoints.rs b/server/src/game_endpoints.rs
@@ -1,6 +1,6 @@
 use crate::{auth_redirect, Games, Lobby};
 use tokio::sync::mpsc::UnboundedReceiver;
-use tokio_stream::{Stream, StreamExt};
+use tokio_stream::{wrappers::UnboundedReceiverStream, Stream, StreamExt};
 use turbo_hearts_api::{
     AcceptClaimRequest, ChargeRequest, ClaimRequest, GameChatRequest, GameEvent, GameId,
     PassRequest, PlayRequest, RejectClaimRequest, UserId,
@@ -49,6 +49,7 @@ fn subscribe<'a>(games: infallible!(&'a Games), user_id: rejection!(UserId)) ->
     fn stream(
         rx: UnboundedReceiver<(GameEvent, usize)>,
     ) -> impl Stream<Item = Result<Event, warp::Error>> {
+        let rx = UnboundedReceiverStream::new(rx);
         rx.map(|(event, id)| {
             if event.is_ping() {
                 return Ok(Event::default().comment(String::new()));

diff --git a/server/src/lobby_endpoints.rs b/server/src/lobby_endpoints.rs
@@ -1,6 +1,6 @@
 use crate::{auth_redirect, Games, Lobby};
 use tokio::sync::mpsc::UnboundedReceiver;
-use tokio_stream::{Stream, StreamExt};
+use tokio_stream::{wrappers::UnboundedReceiverStream, Stream, StreamExt};
 use turbo_hearts_api::{
     AddBotRequest, JoinGameRequest, LeaveGameRequest, LobbyChatRequest, LobbyEvent, NewGameRequest,
     Player, PlayerWithOptions, RemovePlayerRequest, StartGameRequest, UserId,
@@ -42,6 +42,7 @@ fn subscribe<'a>(lobby: infallible!(&'a Lobby), user_id: rejection!(UserId)) ->
     }
 
     fn stream(rx: UnboundedReceiver<LobbyEvent>) -> impl Stream<Item = Result<Event, warp::Error>> {
+        let rx = UnboundedReceiverStream::new(rx);
         rx.map(|event| {
             Ok(if event.is_ping() {
                 Event::default().comment(String::new())

diff --git a/server/src/main.rs b/server/src/main.rs
@@ -2,7 +2,6 @@ use crate::CardsError;
 use http::header;
 use reqwest::Client;
 use tokio::{time, time::Duration};
-use tokio_stream::StreamExt;
 use turbo_hearts_api::UserId;
 use warp::{Filter, Rejection};
 
@@ -49,8 +48,9 @@ fn user_id<'a>(users: infallible!(&'a Users)) -> rejection!(UserId) {
 
 fn start_stale_game_cleanup(lobby: &'static Lobby) {
     tokio::task::spawn(async move {
-        let mut stream = time::interval(Duration::from_secs(60 * 60));
-        while let Some(_) = stream.next().await {
+        let mut ticker = time::interval(Duration::from_secs(60 * 60));
+        loop {
+            ticker.tick().await;
             if let Err(e) = lobby.delete_stale_games().await {
                 log::error!("Failed to delete stale games {:?}", e);
             }
@@ -60,8 +60,9 @@ fn start_stale_game_cleanup(lobby: &'static Lobby) {
 
 fn start_background_pings(lobby: &'static Lobby, games: &'static Games) {
     tokio::task::spawn(async move {
-        let mut stream = time::interval(Duration::from_secs(15));
-        while let Some(_) = stream.next().await {
+        let mut ticker = time::interval(Duration::from_secs(15));
+        loop {
+            ticker.tick().await;
             lobby.ping().await;
             games.ping().await;
         }

diff --git a/server/src/test.rs b/server/src/test.rs
@@ -79,7 +79,7 @@ impl TestRunner {
     }
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_lobby() -> Result<(), CardsError> {
     async fn test(_: &Database, lobby: &Lobby, _: &Games) -> Result<(), CardsError> {
         let mut twilson = lobby.subscribe(*TWILSON).await?;
@@ -183,7 +183,7 @@ async fn test_lobby() -> Result<(), CardsError> {
     TestRunner::new().run(test).await
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_new_game() -> Result<(), CardsError> {
     async fn test(_: &Database, lobby: &Lobby, games: &Games) -> Result<(), CardsError> {
         let game_id = lobby
@@ -252,7 +252,7 @@ async fn test_new_game() -> Result<(), CardsError> {
     TestRunner::new().run(test).await
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_pass() -> Result<(), CardsError> {
     async fn test(db: &Database, _lobby: &Lobby, games: &Games) -> Result<(), CardsError> {
         let game_id = GameId::new();
@@ -314,7 +314,7 @@ async fn test_pass() -> Result<(), CardsError> {
     TestRunner::new().run(test).await
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_seeded_game() -> Result<(), CardsError> {
     async fn test(_db: &Database, lobby: &Lobby, games: &Games) -> Result<(), CardsError> {
         let game_id = lobby
@@ -429,7 +429,7 @@ async fn test_seeded_game() -> Result<(), CardsError> {
     TestRunner::new().run(test).await
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_bot_game() -> Result<(), CardsError> {
     async fn test(_: &Database, lobby: &Lobby, games: &Games) -> Result<(), CardsError> {
         let game_id = lobby