Merge pull request #677 from swimos/circ-fix

Fixes faulty test logic for the circular buffer channel.

Author: horned-sphere

swimos_utilities/swimos_sync/Cargo.toml

@@ -16,4 +16,4 @@ swimos_num = { workspace = true }
 
 [dev-dependencies]
 futures = { workspace = true }
-tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros", "sync"] }

swimos_utilities/swimos_sync/src/circular_buffer/tests.rs

@@ -14,6 +14,7 @@
 
 use crate::circular_buffer::error::{RecvError, SendError};
 use crate::circular_buffer::{InternalQueue, OneItemQueue, LARGE_BOUNDARY};
+use futures::future::join;
 use futures::task::ArcWake;
 use futures::StreamExt;
 use std::future::Future;
@@ -183,6 +184,7 @@ async fn receive_several_stream(n: usize) {
         for i in 0..n {
             assert!(tx.try_send(i).is_ok());
         }
+        tx
     };
 
     let recv_task = async move {
@@ -194,8 +196,9 @@ async fn receive_several_stream(n: usize) {
     let send_handle = tokio::spawn(send_task);
     let recv_handle = tokio::spawn(recv_task);
 
-    assert!(send_handle.await.is_ok());
-    assert!(recv_handle.await.is_ok());
+    let (send_result, recv_result) = join(send_handle, recv_handle).await;
+    assert!(send_result.is_ok());
+    assert!(recv_result.is_ok());
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -204,8 +207,6 @@ async fn small_receive_several_stream() {
 }
 
 #[tokio::test(flavor = "multi_thread")]
-#[ignore]
-// todo: spuriously failing on CI
 async fn large_receive_several_stream() {
     receive_several_stream(LARGE_BOUNDARY + 1).await;
 }