Skip to content

Commit

Permalink
Fix tests
Browse files Browse the repository at this point in the history
  • Loading branch information
olegnn committed Mar 20, 2023
1 parent d94765b commit 0505536
Show file tree
Hide file tree
Showing 2 changed files with 40 additions and 31 deletions.
7 changes: 7 additions & 0 deletions futures/tests/no-std/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3,28 +3,35 @@

#[cfg(feature = "futures-core-alloc")]
#[cfg(target_has_atomic = "ptr")]
#[allow(useless_anonymous_reexport)]
pub use futures_core::task::__internal::AtomicWaker as _;

#[cfg(feature = "futures-task-alloc")]
#[cfg(target_has_atomic = "ptr")]
#[allow(useless_anonymous_reexport)]
pub use futures_task::ArcWake as _;

#[cfg(feature = "futures-channel-alloc")]
#[cfg(target_has_atomic = "ptr")]
#[allow(useless_anonymous_reexport)]
pub use futures_channel::oneshot as _;

#[cfg(any(feature = "futures", feature = "futures-alloc"))]
#[cfg(target_has_atomic = "ptr")]
#[allow(useless_anonymous_reexport)]
pub use futures::task::AtomicWaker as _;

#[cfg(feature = "futures-alloc")]
#[cfg(target_has_atomic = "ptr")]
#[allow(useless_anonymous_reexport)]
pub use futures::stream::FuturesOrdered as _;

#[cfg(any(feature = "futures-util", feature = "futures-util-alloc"))]
#[cfg(target_has_atomic = "ptr")]
#[allow(useless_anonymous_reexport)]
pub use futures_util::task::AtomicWaker as _;

#[cfg(feature = "futures-util-alloc")]
#[cfg(target_has_atomic = "ptr")]
#[allow(useless_anonymous_reexport)]
pub use futures_util::stream::FuturesOrdered as _;
64 changes: 33 additions & 31 deletions futures/tests/stream.rs
Original file line number Diff line number Diff line change
Expand Up @@ -344,40 +344,42 @@ fn flatten_unordered() {

// nested `flatten_unordered`
let te = ThreadPool::new().unwrap();
let handle = te.spawn_with_handle(async move {
let inner = stream::iter(0..10)
.then(|_| {
let task = Arc::new(AtomicBool::new(false));
let mut spawned = false;

future::poll_fn(move |cx| {
if !spawned {
let waker = cx.waker().clone();
let task = task.clone();

std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(500));
task.store(true, Ordering::Release);

waker.wake_by_ref()
});
spawned = true;
}

if task.load(Ordering::Acquire) {
Poll::Ready(Some(()))
} else {
Poll::Pending
}
let handle = te
.spawn_with_handle(async move {
let inner = stream::iter(0..10)
.then(|_| {
let task = Arc::new(AtomicBool::new(false));
let mut spawned = false;

future::poll_fn(move |cx| {
if !spawned {
let waker = cx.waker().clone();
let task = task.clone();

std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(500));
task.store(true, Ordering::Release);

waker.wake_by_ref()
});
spawned = true;
}

if task.load(Ordering::Acquire) {
Poll::Ready(Some(()))
} else {
Poll::Pending
}
})
})
})
.map(|_| stream::once(future::ready(())))
.flatten_unordered(None);
.map(|_| stream::once(future::ready(())))
.flatten_unordered(None);

let stream = stream::once(future::ready(inner)).flatten_unordered(None);
let stream = stream::once(future::ready(inner)).flatten_unordered(None);

assert_eq!(stream.count().await, 10);
}).unwrap();
assert_eq!(stream.count().await, 10);
})
.unwrap();

block_on(handle);
}
Expand Down

0 comments on commit 0505536

Please sign in to comment.