From a75defa5a6f3e34fc70bab5bda2a24b143c8a738 Mon Sep 17 00:00:00 2001 From: TechnoPorg Date: Wed, 11 Feb 2026 22:36:38 +0100 Subject: [PATCH] Depend on futures-util instead of futures --- Cargo.lock | 2 +- Cargo.toml | 2 +- README.md | 6 ++-- benches/basic/execution.rs | 8 +++--- benches/basic/scaling.rs | 4 +-- benches/comparisons/deep_chain.rs | 4 +-- benches/comparisons/diamond_pattern.rs | 4 +-- benches/comparisons/etl_pipeline.rs | 4 +-- benches/comparisons/large_scale.rs | 4 +-- benches/comparisons/linear_pipeline.rs | 4 +-- benches/comparisons/wide_fanout.rs | 4 +-- benches/patterns/breakdown.rs | 6 ++-- benches/patterns/diamond.rs | 4 +-- benches/patterns/fanout.rs | 8 +++--- benches/realistic/config_broadcast.rs | 4 +-- benches/realistic/etl.rs | 4 +-- benches/realistic/memory_efficiency.rs | 4 +-- benches/realistic/mixed.rs | 4 +-- docs/COMPARISONS.md | 2 +- docs/CYCLE_PREVENTION.md | 2 +- examples/01_basic.rs | 4 +-- examples/02_fan_out.rs | 4 +-- examples/03_fan_in.rs | 4 +-- examples/04_parallel_computation.rs | 4 +-- examples/circuit_breaker.rs | 16 +++++------ examples/complex_dag.rs | 4 +-- examples/conditional_workflow.rs | 4 +-- examples/custom_types.rs | 4 +-- examples/data_pipeline.rs | 4 +-- examples/debug_tracing.rs | 6 ++-- examples/error_handling.rs | 10 +++---- examples/large_dag.rs | 12 ++++---- examples/retry_strategies.rs | 12 ++++---- examples/timeout.rs | 14 ++++----- examples/tracing_example.rs | 8 +++--- src/builder.rs | 8 +++--- src/error.rs | 2 +- src/error/tests.rs | 2 +- src/lib.rs | 29 +++++++++---------- src/runner.rs | 25 ++++++++-------- src/runner/tests.rs | 16 ++++++----- src/types.rs | 4 +-- src/types/coverage_tests.rs | 11 ++++--- tests/arc_tests.rs | 40 +++++++++++++------------- tests/boundaries/deep_chains.rs | 20 ++++++------- tests/boundaries/large_scale.rs | 12 ++++---- tests/boundaries/type_limits.rs | 18 ++++++------ tests/boundaries/zero_tasks.rs | 24 ++++++++-------- tests/custom_type_test.rs | 9 +++--- tests/cycle_prevention.rs | 3 +- tests/dependencies/resolution.rs | 10 +++---- tests/dependencies/single.rs | 12 ++++---- tests/dependencies/tuples.rs | 16 +++++------ tests/errors/error_types.rs | 19 ++++++------ tests/errors/propagation.rs | 30 +++++++++++++------ tests/errors/recovery.rs | 29 +++++++++++++------ tests/execution/basic.rs | 27 +++++++++-------- tests/execution/integration.rs | 32 ++++++++++++++------- tests/execution/results.rs | 16 ++++++----- tests/parallelism/concurrent.rs | 20 ++++++------- tests/parallelism/execution.rs | 8 +++--- tests/parallelism/spawning.rs | 5 ++-- tests/parallelism/threads.rs | 14 +++++---- tests/parallelism/timing.rs | 5 ++-- tests/patterns/butterfly.rs | 14 +++++---- tests/patterns/complex.rs | 17 +++++++---- tests/patterns/grid.rs | 13 +++++---- tests/patterns/tree.rs | 19 +++++++----- tests/performance/mod.rs | 6 ++-- tests/runtimes/compatibility.rs | 4 +-- tests/runtimes/quirky.rs | 2 +- tests/task_patterns/combined.rs | 8 +++--- tests/task_patterns/readonly.rs | 7 ++--- tests/task_patterns/stateless.rs | 17 ++++++----- tests/timing/concurrent_access.rs | 23 +++++++++------ tests/timing/interleaving.rs | 14 +++++---- tests/tracing/with_tracing.rs | 10 +++---- tests/tracing/without_tracing.rs | 5 ++-- tests/type_safety/mod.rs | 8 +++--- 79 files changed, 447 insertions(+), 380 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ce288a7..046eee9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -442,8 +442,8 @@ dependencies = [ "criterion", "dagrs", "dagx-macros", - "futures", "futures-executor", + "futures-util", "parking_lot", "pollster", "smol", diff --git a/Cargo.toml b/Cargo.toml index 1466677..7fb6d41 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,7 +28,7 @@ unexpected_cfgs = { level = "warn", check-cfg = [ ] } [dependencies] -futures = { version = "0.3", default-features = false, features = ["std"] } +futures-util = { version = "0.3", default-features = false, features = ["std"] } parking_lot = "0.12" dagx-macros = { version = "0.3.1", path = "./dagx-macros" } tracing = { version = "0.1", optional = true } diff --git a/README.md b/README.md index 16875ce..c7fe0d3 100644 --- a/README.md +++ b/README.md @@ -38,7 +38,7 @@ See [how it works](docs/CYCLE_PREVENTION.md). ```rust let sum = dag.add_task(Add).depends_on((&x, &y)); -dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; +dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; ``` That's it. No trait boilerplate, no manual channels, no node IDs. @@ -91,7 +91,7 @@ async fn main() { let sum = dag.add_task(Add).depends_on((x, &y)); // Execute with true parallelism - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); // Retrieve results assert_eq!(dag.get(sum).unwrap(), 5); @@ -223,7 +223,7 @@ dagx works with any async runtime. Provide a spawner function to `run()`: ```rust // With Tokio // The join handle result can be unwrapped because dagx catches panics internally -dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); +dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); // With async-std dag.run(|fut| async_std::task::spawn(fut)).await.unwrap(); diff --git a/benches/basic/execution.rs b/benches/basic/execution.rs index 85aa2c0..70a895c 100644 --- a/benches/basic/execution.rs +++ b/benches/basic/execution.rs @@ -2,7 +2,7 @@ use criterion::Criterion; use dagx::{task_fn, DagRunner}; -use futures::FutureExt; + pub fn bench_dag_execution(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); @@ -14,7 +14,7 @@ pub fn bench_dag_execution(c: &mut Criterion) { for i in 0..10 { dag.add_task(task_fn::<(), _, _>(move |_: ()| i)); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) @@ -28,7 +28,7 @@ pub fn bench_dag_execution(c: &mut Criterion) { for i in 0..100 { dag.add_task(task_fn::<(), _, _>(move |_: ()| i)); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) @@ -108,7 +108,7 @@ pub fn bench_dag_execution(c: &mut Criterion) { .add_task(task_fn::(|&x: &i32| x + 1)) .depends_on(t18); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) diff --git a/benches/basic/scaling.rs b/benches/basic/scaling.rs index 32199aa..61d1904 100644 --- a/benches/basic/scaling.rs +++ b/benches/basic/scaling.rs @@ -2,7 +2,7 @@ use criterion::{BenchmarkId, Criterion}; use dagx::{task_fn, DagRunner}; -use futures::FutureExt; + pub fn bench_dag_scaling(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); @@ -19,7 +19,7 @@ pub fn bench_dag_scaling(c: &mut Criterion) { for i in 0..size { dag.add_task(task_fn::<(), _, _>(move |_: ()| i)); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) diff --git a/benches/comparisons/deep_chain.rs b/benches/comparisons/deep_chain.rs index e58eb32..324c965 100644 --- a/benches/comparisons/deep_chain.rs +++ b/benches/comparisons/deep_chain.rs @@ -4,7 +4,7 @@ use criterion::Criterion; use dagx::TaskHandle; -use futures::FutureExt; + pub fn bench_deep_chain(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); @@ -29,7 +29,7 @@ pub fn bench_deep_chain(c: &mut Criterion) { .depends_on(prev); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); dag.get(prev).unwrap() diff --git a/benches/comparisons/diamond_pattern.rs b/benches/comparisons/diamond_pattern.rs index 8406457..6d92606 100644 --- a/benches/comparisons/diamond_pattern.rs +++ b/benches/comparisons/diamond_pattern.rs @@ -4,7 +4,7 @@ use criterion::Criterion; use dagx::TaskHandle; -use futures::FutureExt; + pub fn bench_diamond_pattern(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); @@ -29,7 +29,7 @@ pub fn bench_diamond_pattern(c: &mut Criterion) { .add_task(task_fn::<(i32, i32), _, _>(|(x, y): (&i32, &i32)| x + y)) .depends_on((&b, &c)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); dag.get(d).unwrap() diff --git a/benches/comparisons/etl_pipeline.rs b/benches/comparisons/etl_pipeline.rs index bf7e506..fcb92da 100644 --- a/benches/comparisons/etl_pipeline.rs +++ b/benches/comparisons/etl_pipeline.rs @@ -3,7 +3,7 @@ //! Extract → Transform → Validate → Load pattern use criterion::Criterion; -use futures::FutureExt; + pub fn bench_etl_pipeline(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); @@ -70,7 +70,7 @@ pub fn bench_etl_pipeline(c: &mut Criterion) { })) .depends_on(validate); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); dag.get(load).unwrap() diff --git a/benches/comparisons/large_scale.rs b/benches/comparisons/large_scale.rs index fbb9541..ff03076 100644 --- a/benches/comparisons/large_scale.rs +++ b/benches/comparisons/large_scale.rs @@ -3,7 +3,7 @@ //! 10,000 independent tasks to test scaling use criterion::Criterion; -use futures::FutureExt; + pub fn bench_large_scale(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); @@ -22,7 +22,7 @@ pub fn bench_large_scale(c: &mut Criterion) { dag.add_task(task_fn::<(), _, _>(move |_: ()| i * 2)); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) diff --git a/benches/comparisons/linear_pipeline.rs b/benches/comparisons/linear_pipeline.rs index bfbb7cc..543ce24 100644 --- a/benches/comparisons/linear_pipeline.rs +++ b/benches/comparisons/linear_pipeline.rs @@ -3,7 +3,7 @@ //! Simple chain of tasks: A → B → C → D → E use criterion::Criterion; -use futures::FutureExt; + pub fn bench_linear_pipeline(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); @@ -31,7 +31,7 @@ pub fn bench_linear_pipeline(c: &mut Criterion) { .add_task(task_fn::(|&x: &i32| x * 3)) .depends_on(d); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); dag.get(e).unwrap() diff --git a/benches/comparisons/wide_fanout.rs b/benches/comparisons/wide_fanout.rs index 6aa00fc..da434a2 100644 --- a/benches/comparisons/wide_fanout.rs +++ b/benches/comparisons/wide_fanout.rs @@ -4,7 +4,7 @@ use criterion::Criterion; use dagx::TaskHandle; -use futures::FutureExt; + pub fn bench_wide_fanout(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); @@ -25,7 +25,7 @@ pub fn bench_wide_fanout(c: &mut Criterion) { .depends_on(source); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) diff --git a/benches/patterns/breakdown.rs b/benches/patterns/breakdown.rs index c4f5483..401d762 100644 --- a/benches/patterns/breakdown.rs +++ b/benches/patterns/breakdown.rs @@ -2,7 +2,7 @@ use criterion::Criterion; use dagx::{task_fn, DagRunner}; -use futures::FutureExt; + pub fn bench_10k_breakdown(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); @@ -27,7 +27,7 @@ pub fn bench_10k_breakdown(c: &mut Criterion) { for i in 0..10_000 { dag.add_task(task_fn::<(), _, _>(move |_: ()| i)); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) @@ -44,7 +44,7 @@ pub fn bench_10k_breakdown(c: &mut Criterion) { b.iter(|| { rt.block_on(async { - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) diff --git a/benches/patterns/diamond.rs b/benches/patterns/diamond.rs index bb522a9..ce215ad 100644 --- a/benches/patterns/diamond.rs +++ b/benches/patterns/diamond.rs @@ -2,7 +2,7 @@ use criterion::Criterion; use dagx::{task_fn, DagRunner, TaskHandle}; -use futures::FutureExt; + pub fn bench_diamond(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); @@ -27,7 +27,7 @@ pub fn bench_diamond(c: &mut Criterion) { .depends_on((&left, &right)); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) diff --git a/benches/patterns/fanout.rs b/benches/patterns/fanout.rs index 14c61d8..c68a60f 100644 --- a/benches/patterns/fanout.rs +++ b/benches/patterns/fanout.rs @@ -2,7 +2,7 @@ use criterion::Criterion; use dagx::{task_fn, DagRunner, TaskHandle}; -use futures::FutureExt; + pub fn bench_fanout(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); @@ -19,7 +19,7 @@ pub fn bench_fanout(c: &mut Criterion) { .depends_on(source); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) @@ -46,7 +46,7 @@ pub fn bench_fanout(c: &mut Criterion) { .depends_on(source); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) @@ -72,7 +72,7 @@ pub fn bench_fanout(c: &mut Criterion) { .depends_on(source); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) diff --git a/benches/realistic/config_broadcast.rs b/benches/realistic/config_broadcast.rs index b0a5658..c011b7f 100644 --- a/benches/realistic/config_broadcast.rs +++ b/benches/realistic/config_broadcast.rs @@ -2,7 +2,7 @@ use criterion::{BenchmarkId, Criterion}; use dagx::{task_fn, DagRunner, TaskHandle}; -use futures::FutureExt; + use std::collections::HashMap; use std::hint::black_box; @@ -56,7 +56,7 @@ pub fn bench_config_broadcast(c: &mut Criterion) { .depends_on(config); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) diff --git a/benches/realistic/etl.rs b/benches/realistic/etl.rs index 9f4f853..80a9610 100644 --- a/benches/realistic/etl.rs +++ b/benches/realistic/etl.rs @@ -2,7 +2,7 @@ use criterion::{BenchmarkId, Criterion}; use dagx::{task_fn, DagRunner, TaskHandle}; -use futures::FutureExt; + pub fn bench_etl_pipeline(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); @@ -93,7 +93,7 @@ pub fn bench_etl_pipeline(c: &mut Criterion) { )) .depends_on((&validate, &scores, &names)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) diff --git a/benches/realistic/memory_efficiency.rs b/benches/realistic/memory_efficiency.rs index fa3f435..64b5e69 100644 --- a/benches/realistic/memory_efficiency.rs +++ b/benches/realistic/memory_efficiency.rs @@ -2,7 +2,7 @@ use criterion::{BenchmarkId, Criterion}; use dagx::{task_fn, DagRunner, TaskHandle}; -use futures::FutureExt; + use std::hint::black_box; pub fn bench_memory_efficiency(c: &mut Criterion) { @@ -41,7 +41,7 @@ pub fn bench_memory_efficiency(c: &mut Criterion) { .depends_on(source); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) diff --git a/benches/realistic/mixed.rs b/benches/realistic/mixed.rs index 02bf033..f0d7637 100644 --- a/benches/realistic/mixed.rs +++ b/benches/realistic/mixed.rs @@ -2,7 +2,7 @@ use criterion::Criterion; use dagx::{task_fn, DagRunner, TaskHandle}; -use futures::FutureExt; + pub fn bench_mixed_patterns(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); @@ -55,7 +55,7 @@ pub fn bench_mixed_patterns(c: &mut Criterion) { .depends_on(summary); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); }) diff --git a/docs/COMPARISONS.md b/docs/COMPARISONS.md index eec374d..108b425 100644 --- a/docs/COMPARISONS.md +++ b/docs/COMPARISONS.md @@ -19,7 +19,7 @@ let dag = DagRunner::new(); let x = dag.add_task(Value(2)); let y = dag.add_task(Value(3)); let sum = dag.add_task(Add).depends_on((x, y)); -dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; +dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; ``` **dagrs** (Complex: traits + channels + IDs + manual wiring): diff --git a/docs/CYCLE_PREVENTION.md b/docs/CYCLE_PREVENTION.md index 1e26b9c..5ee861f 100644 --- a/docs/CYCLE_PREVENTION.md +++ b/docs/CYCLE_PREVENTION.md @@ -13,7 +13,7 @@ let a = dag.add_task(TaskA); let b = dag.add_task(TaskB); a.depends_on(b); b.depends_on(a); // Compiles fine... -dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) +dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) ``` **dagx prevents this at compile time:** diff --git a/examples/01_basic.rs b/examples/01_basic.rs index 3b6d5bb..5e7ba3c 100644 --- a/examples/01_basic.rs +++ b/examples/01_basic.rs @@ -31,7 +31,7 @@ //! ``` use dagx::{task, DagRunner}; -use futures::FutureExt; + // Step 1: Define a source task (no dependencies) // @@ -106,7 +106,7 @@ async fn main() { // 1. Tasks with no dependencies run first (x and y in parallel) // 2. Once x and y complete, sum runs (it depends on both) println!("Running DAG...\n"); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/examples/02_fan_out.rs b/examples/02_fan_out.rs index d4f4723..0cc78cd 100644 --- a/examples/02_fan_out.rs +++ b/examples/02_fan_out.rs @@ -56,7 +56,7 @@ //! ``` use dagx::{task, DagRunner, TaskHandle}; -use futures::FutureExt; + // Source task with a value field struct Source { @@ -141,7 +141,7 @@ async fn main() { // Run the DAG println!("Running fan-out DAG...\n"); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/examples/03_fan_in.rs b/examples/03_fan_in.rs index b0ae2cf..cbc3761 100644 --- a/examples/03_fan_in.rs +++ b/examples/03_fan_in.rs @@ -62,7 +62,7 @@ //! ``` use dagx::{task, DagRunner}; -use futures::FutureExt; + struct FetchName; @@ -134,7 +134,7 @@ async fn main() { // Run the DAG println!("Running fan-in DAG...\n"); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/examples/04_parallel_computation.rs b/examples/04_parallel_computation.rs index 2198956..65fc54a 100644 --- a/examples/04_parallel_computation.rs +++ b/examples/04_parallel_computation.rs @@ -53,7 +53,7 @@ //! ``` use dagx::{task, DagResult, DagRunner, Task, TaskInput}; -use futures::FutureExt; + use std::time::{Duration, Instant}; use tokio::time::sleep; @@ -125,7 +125,7 @@ async fn main() -> DagResult<()> { let total = dag.add_task(Aggregate).depends_on((sum1, sum2, sum3, sum4)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; let result = dag.get(total)?; let elapsed = start.elapsed(); diff --git a/examples/circuit_breaker.rs b/examples/circuit_breaker.rs index b2c5f4c..c661448 100644 --- a/examples/circuit_breaker.rs +++ b/examples/circuit_breaker.rs @@ -151,7 +151,7 @@ //! ``` use dagx::{task, DagRunner}; -use futures::FutureExt; + use std::sync::Arc; use std::time::Instant; use tokio::sync::Mutex; @@ -390,7 +390,7 @@ async fn main() { false, // success )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -404,7 +404,7 @@ async fn main() { true, // fail )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -418,7 +418,7 @@ async fn main() { true, // fail )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -432,7 +432,7 @@ async fn main() { true, // fail )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -450,7 +450,7 @@ async fn main() { .add_task(FallbackService::new("ServiceCall-5")) .depends_on(call5); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -469,7 +469,7 @@ async fn main() { false, // success )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -483,7 +483,7 @@ async fn main() { false, // success )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/examples/complex_dag.rs b/examples/complex_dag.rs index 50810ec..b5fff8c 100644 --- a/examples/complex_dag.rs +++ b/examples/complex_dag.rs @@ -67,7 +67,7 @@ //! ``` use dagx::{task, DagResult, DagRunner}; -use futures::FutureExt; + // Extract: Load data from different sources struct LoadData { @@ -202,7 +202,7 @@ async fn main() -> DagResult<()> { // Execute the entire ETL pipeline println!("Executing pipeline...\n"); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; // Display final report println!("\n{}", dag.get(report)?); diff --git a/examples/conditional_workflow.rs b/examples/conditional_workflow.rs index f9aacfe..c47e4a6 100644 --- a/examples/conditional_workflow.rs +++ b/examples/conditional_workflow.rs @@ -56,7 +56,7 @@ //! ``` use dagx::{task, DagResult, DagRunner}; -use futures::FutureExt; + // Task that validates and processes input struct ValidateAndDouble { @@ -138,7 +138,7 @@ async fn main() -> DagResult<()> { .depends_on((&cat1, &cat2, &cat3)); // Execute - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; // Display results println!("\nFinal summary: {}", dag.get(summary)?); diff --git a/examples/custom_types.rs b/examples/custom_types.rs index c4ebb11..d61743c 100644 --- a/examples/custom_types.rs +++ b/examples/custom_types.rs @@ -35,7 +35,7 @@ //! ``` use dagx::{task, DagRunner, TaskHandle}; -use futures::FutureExt; + #[derive(Debug)] struct User { @@ -107,7 +107,7 @@ async fn main() { let validation = dag.add_task(ValidateUser).depends_on(user_task); // Execute the DAG - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/examples/data_pipeline.rs b/examples/data_pipeline.rs index 75b7d9d..c290ff1 100644 --- a/examples/data_pipeline.rs +++ b/examples/data_pipeline.rs @@ -76,7 +76,7 @@ //! ``` use dagx::{task, DagRunner}; -use futures::FutureExt; + // Data source tasks struct LoadValue1; @@ -221,7 +221,7 @@ async fn main() { // Execute the pipeline println!("Executing pipeline...\n"); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/examples/debug_tracing.rs b/examples/debug_tracing.rs index 233ecfa..9dc2d2c 100644 --- a/examples/debug_tracing.rs +++ b/examples/debug_tracing.rs @@ -115,7 +115,7 @@ //! ``` use dagx::{task, DagRunner}; -use futures::FutureExt; + use std::time::Instant; use tokio::time::{sleep, Duration}; @@ -401,7 +401,7 @@ async fn main() { .depends_on((inventory, payment, shipping)); // Run the DAG - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -441,7 +441,7 @@ async fn main() { true, // will fail )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/examples/error_handling.rs b/examples/error_handling.rs index 22cd157..3290a40 100644 --- a/examples/error_handling.rs +++ b/examples/error_handling.rs @@ -115,7 +115,7 @@ //! ``` use dagx::{task, DagRunner}; -use futures::FutureExt; + // Source task that provides a string input struct StringSource(&'static str); @@ -223,7 +223,7 @@ async fn main() { let processed = dag.add_task(ProcessData).depends_on(validated); let logged = dag.add_task(LogError).depends_on(processed); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -249,7 +249,7 @@ async fn main() { let processed = dag.add_task(ProcessData).depends_on(validated); let logged = dag.add_task(LogError).depends_on(processed); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -275,7 +275,7 @@ async fn main() { let processed = dag.add_task(ProcessData).depends_on(validated); let logged = dag.add_task(LogError).depends_on(processed); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -298,7 +298,7 @@ async fn main() { .add_task(WithFallback { fallback: 0 }) .depends_on(parsed); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/examples/large_dag.rs b/examples/large_dag.rs index d8478cc..11aac09 100644 --- a/examples/large_dag.rs +++ b/examples/large_dag.rs @@ -110,7 +110,7 @@ //! ``` use dagx::{task, DagRunner}; -use futures::FutureExt; + use std::time::Instant; // Simple computation task @@ -182,7 +182,7 @@ async fn main() { println!(" Build time: {:?}", build_time); let exec_start = Instant::now(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); let exec_time = exec_start.elapsed(); @@ -222,7 +222,7 @@ async fn main() { println!(" Build time: {:?}", build_time); let exec_start = Instant::now(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); let exec_time = exec_start.elapsed(); @@ -273,7 +273,7 @@ async fn main() { println!(" Tree depth: {}", level); let exec_start = Instant::now(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); let exec_time = exec_start.elapsed(); @@ -331,7 +331,7 @@ async fn main() { println!(" Build time: {:?}", build_time); let exec_start = Instant::now(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); let exec_time = exec_start.elapsed(); @@ -355,7 +355,7 @@ async fn main() { .map(|i| dag.add_task(FormatNumber { id: i })) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/examples/retry_strategies.rs b/examples/retry_strategies.rs index dd6f015..e04140f 100644 --- a/examples/retry_strategies.rs +++ b/examples/retry_strategies.rs @@ -153,7 +153,7 @@ //! ``` use dagx::{task, DagRunner}; -use futures::FutureExt; + use tokio::time::{sleep, Duration}; // Retry strategy configuration @@ -356,7 +356,7 @@ async fn main() { ], )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -378,7 +378,7 @@ async fn main() { ], )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -404,7 +404,7 @@ async fn main() { ], )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -423,7 +423,7 @@ async fn main() { vec![ApiError::AuthenticationFailed], )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -450,7 +450,7 @@ async fn main() { ], )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/examples/timeout.rs b/examples/timeout.rs index 6e9f9d5..68dbb2b 100644 --- a/examples/timeout.rs +++ b/examples/timeout.rs @@ -24,7 +24,7 @@ //! //! **2. DAG-Level Timeout** - Entire workflow with timeout //! ```rust -//! match timeout(dag_timeout, dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await { +//! match timeout(dag_timeout, dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await { //! Ok(result) => { /* DAG completed */ }, //! Err(_) => { /* DAG timed out */ }, //! } @@ -99,7 +99,7 @@ //! ``` use dagx::{task, DagRunner, TaskHandle}; -use futures::FutureExt; + use tokio::time::{sleep, timeout, Duration}; // Fast task that completes quickly @@ -178,7 +178,7 @@ async fn main() { timeout_ms: 500, }); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -198,7 +198,7 @@ async fn main() { let processed = dag.add_task(ProcessResult).depends_on(timed); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -225,7 +225,7 @@ async fn main() { timeout_ms: 200, }); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -251,7 +251,7 @@ async fn main() { let processed1 = dag.add_task(ProcessResult).depends_on(source); let processed2 = dag.add_task(ProcessResult).depends_on(source); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -272,7 +272,7 @@ async fn main() { let dag_timeout = Duration::from_millis(500); match timeout( dag_timeout, - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)), + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }), ) .await { diff --git a/examples/tracing_example.rs b/examples/tracing_example.rs index 96cb5b1..2fb5025 100644 --- a/examples/tracing_example.rs +++ b/examples/tracing_example.rs @@ -34,7 +34,7 @@ //! use dagx::{task, DagRunner}; -use futures::FutureExt; + use tracing_subscriber::{fmt, EnvFilter}; // Simple source task @@ -96,7 +96,7 @@ async fn main() { // Layer 2: Combine results let result = dag.add_task(Add).depends_on((&left, &right)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -117,7 +117,7 @@ async fn main() { let c = dag.add_task(Multiply).depends_on((&b, &b)); let d = dag.add_task(Add).depends_on((&c, &c)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -146,7 +146,7 @@ async fn main() { let sum_all = dag.add_task(Add).depends_on((&sum12, &sum23)); let final_result = dag.add_task(Add).depends_on((&sum_all, &sum34)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/src/builder.rs b/src/builder.rs index 1f965a3..1129442 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -22,7 +22,7 @@ use tracing::debug; /// /// ```no_run /// # use dagx::{task, DagRunner, Task}; -/// # use futures::FutureExt; +/// # /// // Using tuple struct for simple constants /// struct Constant(i32); /// @@ -57,7 +57,7 @@ use tracing::debug; /// let b = b.depends_on(a); /// // Now b is a TaskHandle /// -/// dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); +/// dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); /// assert_eq!(dag.get(b).unwrap(), 20); /// # }; /// ``` @@ -81,7 +81,7 @@ impl<'a, Tk: Task, Deps> TaskBuilder<'a, Tk, Deps> { /// /// ```no_run /// # use dagx::{task, DagRunner, Task}; - /// # use futures::FutureExt; + /// # /// // Tuple struct /// struct Value(i32); /// @@ -118,7 +118,7 @@ impl<'a, Tk: Task, Deps> TaskBuilder<'a, Tk, Deps> { /// // Multiple dependencies: tuple form /// let sum = dag.add_task(Add).depends_on((&x, y)); /// - /// dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); + /// dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); /// # }; /// ``` pub fn depends_on(self, deps: D) -> TaskHandle diff --git a/src/error.rs b/src/error.rs index 8529bec..fa9674c 100644 --- a/src/error.rs +++ b/src/error.rs @@ -63,7 +63,7 @@ impl std::fmt::Display for DagError { f, "Result not found for task #{}.\n\ \n\ - Call dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).", + Call dag.run(|fut| async move {{ tokio::spawn(fut).await.unwrap() }}).", task_id ) } diff --git a/src/error/tests.rs b/src/error/tests.rs index cb2e65a..dc2130f 100644 --- a/src/error/tests.rs +++ b/src/error/tests.rs @@ -51,7 +51,7 @@ fn test_dag_error_display_result_not_found() { assert!(display.contains("Result not found")); assert!(display.contains("task #7")); - assert!(display.contains("Call dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))")); + assert!(display.contains("Call dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })")); } #[test] diff --git a/src/lib.rs b/src/lib.rs index d1d9726..9730e5c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -24,7 +24,6 @@ //! //! ```no_run //! use dagx::{task, DagRunner, Task, TaskHandle}; -//! use futures::FutureExt; // for FutureExt::map //! //! // Source task with read-only state (tuple struct) //! struct Value(i32); @@ -53,7 +52,7 @@ //! let sum = dag.add_task(Add).depends_on((x, &y)); //! //! // Execute and retrieve results -//! dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); +//! dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); //! assert_eq!(dag.get(sum).unwrap(), 5); //! # }; //! ``` @@ -291,7 +290,7 @@ //! //! ```no_run //! use dagx::{task, DagRunner, Task}; -//! use futures::FutureExt; +//! //! //! // Source task (tuple struct) //! struct Value(i32); @@ -324,7 +323,7 @@ //! let plus1 = dag.add_task(Add(1)).depends_on(&base); //! let times2 = dag.add_task(Scale(2)).depends_on(&base); //! -//! dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); +//! dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); //! //! assert_eq!(dag.get(plus1).unwrap(), 11); //! assert_eq!(dag.get(times2).unwrap(), 20); @@ -337,7 +336,7 @@ //! //! ```no_run //! use dagx::{task, DagRunner, Task}; -//! use futures::FutureExt; +//! //! //! // Source task for String (tuple struct) //! struct Name(String); @@ -381,7 +380,7 @@ //! let active = dag.add_task(Active(true)); //! let result = dag.add_task(FormatUser).depends_on((name, age, active)); //! -//! dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); +//! dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); //! //! assert_eq!(dag.get(result).unwrap(), "User: Alice, Age: 30, Active: true"); //! # }; @@ -393,7 +392,7 @@ //! //! ```no_run //! use dagx::{task, DagRunner, Task}; -//! use futures::FutureExt; +//! //! //! // Source task (tuple struct) //! struct Value(i32); @@ -434,7 +433,7 @@ //! // Layer 3: Final result //! let total = dag.add_task(Add).depends_on((&sum_xy, &prod_yz)); // 5 + 15 = 20 //! -//! dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); +//! dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); //! //! assert_eq!(dag.get(total).unwrap(), 20); //! # }; @@ -447,7 +446,7 @@ //! //! ```no_run //! # use dagx::{task, DagRunner, Task}; -//! # use futures::FutureExt; +//! # //! //! struct User { //! name: String, @@ -476,7 +475,7 @@ //! let user = dag.add_task(CreateUser); //! let formatted = dag.add_task(FormatUser).depends_on(user); //! -//! dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); +//! dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); //! //! assert_eq!(dag.get(formatted).unwrap(), "Alice is 30 years old"); //! # }; @@ -533,7 +532,7 @@ //! //! ```no_run //! # use dagx::{task, DagRunner, Task}; -//! # use futures::FutureExt; +//! # //! # struct Value(i32); //! # #[task] //! # impl Value { @@ -543,11 +542,11 @@ //! # let dag = DagRunner::new(); //! # let node = dag.add_task(Value(42)); //! // Simple approach with .unwrap() -//! dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); +//! dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); //! let result = dag.get(node).unwrap(); //! //! // Or handle errors explicitly -//! match dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await { +//! match dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await { //! Ok(_) => println!("DAG executed successfully"), //! Err(e) => eprintln!("DAG execution failed: {}", e), //! } @@ -631,7 +630,7 @@ //! //! ``` //! # use dagx::{task, DagRunner, Task}; -//! # use futures::FutureExt; +//! # //! //! // ✅ CORRECT: Just output Vec, framework wraps in Arc internally //! struct FetchData; @@ -661,7 +660,7 @@ //! let task2 = dag.add_task(ProcessData).depends_on(&data); //! let task3 = dag.add_task(ProcessData).depends_on(&data); //! -//! dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); +//! dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); //! # }; //! ``` //! diff --git a/src/runner.rs b/src/runner.rs index 339b81e..59364da 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -9,14 +9,13 @@ use std::any::Any; use std::collections::{HashMap, HashSet, VecDeque}; use std::future::Future; use std::panic::AssertUnwindSafe; +use std::pin::Pin; use std::sync::{ atomic::{AtomicBool, Ordering}, Arc, }; -use futures::future::BoxFuture; -use futures::stream::FuturesUnordered; -use futures::{FutureExt, StreamExt, TryFutureExt}; +use futures_util::{stream::FuturesUnordered, FutureExt, StreamExt, TryFutureExt}; use parking_lot::Mutex; #[cfg(feature = "tracing")] @@ -28,6 +27,8 @@ use crate::node::{ExecutableNode, TypedNode}; use crate::task::Task; use crate::types::{NodeId, Pending, TaskHandle}; +type BoxFuture<'a, T> = Pin + Send + 'a>>; + // Guard to ensure run_lock is released even on early return or panic struct RunGuard<'a> { lock: &'a AtomicBool, @@ -56,7 +57,7 @@ impl<'a> Drop for RunGuard<'a> { /// /// ```no_run /// # use dagx::{task, DagRunner, Task}; -/// # use futures::FutureExt; +/// # /// // Task with state constructed via ::new() /// struct LoadValue { value: i32 } /// @@ -85,7 +86,7 @@ impl<'a> Drop for RunGuard<'a> { /// let y = dag.add_task(LoadValue::new(3)); /// let sum = dag.add_task(Add).depends_on((x, y)); /// -/// dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); +/// dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); /// /// assert_eq!(dag.get(sum).unwrap(), 5); /// # }; @@ -150,7 +151,7 @@ impl DagRunner { /// /// ```no_run /// # use dagx::{task, DagRunner, Task}; - /// # use futures::FutureExt; + /// # /// // Task with state - shows you construct with specific value /// struct LoadValue { /// initial: i32, @@ -192,7 +193,7 @@ impl DagRunner { /// // Construct task with offset of 1 /// let inc = dag.add_task(AddOffset::new(1)).depends_on(base); /// - /// dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); + /// dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); /// assert_eq!(dag.get(inc).unwrap(), 11); /// # }; /// ``` @@ -236,7 +237,7 @@ impl DagRunner { /// /// - `spawner`: A function that spawns futures on the async runtime /// and returns a handle to the task. This is the only way to run tasks on separate threads. Examples: - /// - Tokio: `|fut| { tokio::spawn(fut).map(Result::unwrap) }` + /// - Tokio: `|fut| { tokio::spawn(fut).await.unwrap() }` /// - Smol: `|fut| { smol::spawn(fut) }` /// - Async-std: `|fut| { async_std::task::spawn(fut) }` /// - Single-threaded: `|fut| fut` @@ -256,7 +257,7 @@ impl DagRunner { /// /// ``` /// # use dagx::{task, DagRunner, Task}; - /// # use futures::FutureExt; + /// # /// // Tuple struct /// struct Value(i32); /// @@ -280,7 +281,7 @@ impl DagRunner { /// let b = dag.add_task(Value(2)); /// let sum = dag.add_task(Add).depends_on((a, b)); /// - /// dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); // Executes all tasks + /// dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); // Executes all tasks /// # }; /// ``` #[inline] @@ -531,7 +532,7 @@ impl DagRunner { /// /// ```no_run /// # use dagx::{task, DagRunner, Task}; - /// # use futures::FutureExt; + /// # /// struct Configuration { /// setting: i32, /// } @@ -553,7 +554,7 @@ impl DagRunner { /// // Construct task with specific setting value /// let task = dag.add_task(Configuration::new(42)); /// - /// dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); + /// dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); /// /// assert_eq!(dag.get(task).unwrap(), 42); /// # }; diff --git a/src/runner/tests.rs b/src/runner/tests.rs index ddefb43..950a416 100644 --- a/src/runner/tests.rs +++ b/src/runner/tests.rs @@ -4,8 +4,6 @@ use crate::error::DagError; use crate::runner::DagRunner; use crate::types::TaskHandle; -use futures::FutureExt; - // Initialize tracing subscriber for tests (idempotent) #[cfg(feature = "tracing")] fn init_tracing() { @@ -69,7 +67,7 @@ async fn test_get_wrong_type() { let dag = DagRunner::new(); let handle = dag.add_task(TestTask { value: 42 }); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -144,12 +142,14 @@ async fn test_concurrent_run_protection() { // Start two runs concurrently let handle1 = tokio::spawn(async move { barrier1.wait().await; - dag1.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await + dag1.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await }); let handle2 = tokio::spawn(async move { barrier2.wait().await; - dag2.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await + dag2.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await }); // One should succeed, one should fail (concurrent execution not supported) @@ -197,7 +197,7 @@ async fn test_invalid_node_id_in_get() { _phantom: std::marker::PhantomData, }; - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -250,7 +250,9 @@ async fn test_task_panic_in_multi_task_layer() { let _good_task = dag.add_task(GoodTask).depends_on(source); // Run the DAG - the panic should be caught - let result = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await; + let result = dag + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await; // The run should fail due to the panic // Note: Tokio catches panics in spawned tasks, so the behavior depends on the runtime diff --git a/src/types.rs b/src/types.rs index 0fab2c1..1b2feea 100644 --- a/src/types.rs +++ b/src/types.rs @@ -23,7 +23,7 @@ pub struct NodeId(pub usize); /// /// ```no_run /// # use dagx::{task, DagRunner, Task}; -/// # use futures::FutureExt; +/// # /// # struct LoadValue { value: i32 } /// # impl LoadValue { pub fn new(v: i32) -> Self { Self { value: v } } } /// # #[task] @@ -34,7 +34,7 @@ pub struct NodeId(pub usize); /// let dag = DagRunner::new(); /// let node = dag.add_task(LoadValue::new(42)); /// -/// dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); +/// dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); /// /// assert_eq!(dag.get(node).unwrap(), 42); /// # }; diff --git a/src/types/coverage_tests.rs b/src/types/coverage_tests.rs index d5c0350..1095fa1 100644 --- a/src/types/coverage_tests.rs +++ b/src/types/coverage_tests.rs @@ -3,7 +3,6 @@ //! These tests ensure we hit line 71 in the From implementation use crate::{task, DagRunner, TaskHandle}; -use futures::FutureExt; #[tokio::test] async fn test_unit_type_taskbuilder_to_handle_conversion() { @@ -31,7 +30,7 @@ async fn test_unit_type_taskbuilder_to_handle_conversion() { // specifically covering line 71 let handle: TaskHandle<_> = builder.into(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -79,7 +78,7 @@ async fn test_unit_type_conversion_with_multiple_tasks() { let second_handle: TaskHandle<_> = second_builder.into(); let third_handle: TaskHandle = third_builder.into(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -118,7 +117,7 @@ async fn test_unit_struct_with_state() { // Explicit conversion - triggers line 71 let handle: TaskHandle<_> = builder.into(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -144,7 +143,7 @@ async fn test_taskbuilder_conversion_assignment() { // This also uses From implementation let handle: TaskHandle> = builder.into(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -181,7 +180,7 @@ async fn test_mixed_unit_and_dependent_tasks() { // Doubler needs dependencies, so we use it differently let doubler = dag.add_task(Doubler).depends_on(source_handle); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/tests/arc_tests.rs b/tests/arc_tests.rs index 95b360e..e8079aa 100644 --- a/tests/arc_tests.rs +++ b/tests/arc_tests.rs @@ -10,7 +10,7 @@ //! These tests serve as both validation and documentation for proper Arc usage. use dagx::{task, DagRunner, TaskHandle}; -use futures::FutureExt; + use std::collections::{HashMap, HashSet}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -44,7 +44,7 @@ async fn test_arc_basic_pipeline() { let producer = dag.add_task(Producer); let consumer = dag.add_task(Consumer).depends_on(producer); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -100,7 +100,7 @@ async fn test_arc_with_primitives() { .add_task(AllConsumer) .depends_on((i32_prod, f64_prod, bool_prod)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -174,7 +174,7 @@ async fn test_arc_with_collections() { .add_task(CollectionConsumer) .depends_on((vec_prod, map_prod, set_prod)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -245,7 +245,7 @@ async fn test_arc_with_custom_structs() { let age_ext = dag.add_task(AgeExtractor).depends_on(person_prod); let email_ext = dag.add_task(EmailCountExtractor).depends_on(person_prod); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -311,7 +311,7 @@ async fn test_arc_fan_out_three_consumers() { let c2 = dag.add_task(FirstConsumer).depends_on(producer); let c3 = dag.add_task(JoinConsumer).depends_on(producer); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -385,7 +385,7 @@ async fn test_arc_fan_out_five_consumers_large_data() { let c4 = dag.add_task(PatternCounter).depends_on(producer); let c5 = dag.add_task(TotalBytesComputer).depends_on(producer); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -475,7 +475,7 @@ async fn test_arc_nested_fan_out_tree() { let l2_negate = dag.add_task(Level2Negate).depends_on(l1_prod); let l2_mod = dag.add_task(Level2Mod10).depends_on(l1_prod); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -543,7 +543,7 @@ async fn test_arc_performance_vs_clone() { let start = Instant::now(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -643,7 +643,7 @@ async fn test_arc_with_result() { let consumer3 = dag.add_task(ArcResultConsumer2).depends_on(arc_result_err); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -750,7 +750,7 @@ async fn test_arc_with_option() { let c3 = dag.add_task(OptionArcConsumer1).depends_on(opt_arc_some); let c4 = dag.add_task(OptionArcConsumer2).depends_on(opt_arc_none); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -805,7 +805,7 @@ async fn test_arc_diamond_dependency() { let right = dag.add_task(RightBranch).depends_on(top); let bottom = dag.add_task(Bottom).depends_on((&left, &right)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -873,7 +873,7 @@ async fn test_arc_empty_collections() { .add_task(EmptyConsumer) .depends_on((vec_task, string_task, map_task)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -944,7 +944,7 @@ async fn test_arc_very_large_data() { let last = dag.add_task(LastSampler).depends_on(producer); let middle = dag.add_task(MiddleSampler).depends_on(producer); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -981,7 +981,7 @@ async fn test_arc_of_arc() { let producer = dag.add_task(NestedProducer); let consumer = dag.add_task(NestedConsumer).depends_on(producer); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -1045,7 +1045,7 @@ async fn test_arc_mixed_with_non_arc() { let unwrap = dag.add_task(Unwrapper).depends_on(arc_prod); let wrap = dag.add_task(Wrapper).depends_on(non_arc_prod); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -1113,7 +1113,7 @@ async fn test_arc_parallel_execution() { let start = Instant::now(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -1162,7 +1162,7 @@ async fn test_arc_reference_counting() { let producer = dag.add_task(RefCountProducer); let checker = dag.add_task(RefCountChecker).depends_on(producer); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -1240,7 +1240,7 @@ async fn test_arc_massive_fanout_performance_comparison() { let start_without_arc = Instant::now(); dag_without_arc - .run(|fut| tokio::spawn(fut).map(Result::unwrap)) + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); let duration_without_arc = start_without_arc.elapsed(); @@ -1294,7 +1294,7 @@ async fn test_arc_massive_fanout_performance_comparison() { let start_with_arc = Instant::now(); dag_with_arc - .run(|fut| tokio::spawn(fut).map(Result::unwrap)) + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); let duration_with_arc = start_with_arc.elapsed(); diff --git a/tests/boundaries/deep_chains.rs b/tests/boundaries/deep_chains.rs index 8db76fb..1a4c9fe 100644 --- a/tests/boundaries/deep_chains.rs +++ b/tests/boundaries/deep_chains.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{DagResult, DagRunner}; -use futures::FutureExt; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; @@ -20,7 +20,7 @@ async fn test_1000_level_deep_chain() -> DagResult<()> { .depends_on(current); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(current)?, 999); Ok(()) @@ -50,7 +50,7 @@ async fn test_deep_chain_with_branches() -> DagResult<()> { } } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(main_chain)?, 499); @@ -108,7 +108,7 @@ async fn test_multiple_parallel_deep_chains() -> DagResult<()> { &chain_ends[7], )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; // Each chain: start + 500 // Chain 0: 0 + 500 = 500 @@ -149,7 +149,7 @@ async fn test_deep_chain_execution_order() -> DagResult<()> { .depends_on(current); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(order.load(Ordering::SeqCst), 100); assert_eq!(dag.get(current)?, 99); @@ -179,7 +179,7 @@ async fn test_deep_binary_tree() -> DagResult<()> { let root = build_tree(&dag, 10, 1); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; // Sum of all values in a complete binary tree with values 1024..2047 // This is sum from 2^10 to 2^11-1 = 1024 * 1023 / 2 + 1024 * 512 = 1047552 @@ -209,7 +209,7 @@ async fn test_fibonacci_chain() -> DagResult<()> { prev1 = next; } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; // 50th Fibonacci number (0-indexed) = 7778742049 assert_eq!(dag.get(prev1)?, 7778742049); @@ -232,7 +232,7 @@ async fn test_deep_chain_with_accumulator() -> DagResult<()> { .depends_on(current); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; let result = dag.get(current)?; assert_eq!(result, 20100); // sum of 1..=200 @@ -270,7 +270,7 @@ async fn test_zigzag_dependencies() -> DagResult<()> { .add_task(task_fn::<(i32, i32), _, _>(|(a, b): (&i32, &i32)| a + b)) .depends_on((&chain_a, &chain_b)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; let result = dag.get(final_task)?; assert!(result > 100); @@ -293,7 +293,7 @@ async fn test_10000_level_chain_stress() -> DagResult<()> { .depends_on(current); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(current)?, 9_999); Ok(()) diff --git a/tests/boundaries/large_scale.rs b/tests/boundaries/large_scale.rs index 542077c..ab75d8f 100644 --- a/tests/boundaries/large_scale.rs +++ b/tests/boundaries/large_scale.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{DagResult, DagRunner}; -use futures::FutureExt; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; @@ -23,7 +23,7 @@ async fn test_10000_independent_tasks() -> DagResult<()> { }) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; // Verify all tasks completed assert_eq!(counter.load(Ordering::Relaxed), 10_000); @@ -100,7 +100,7 @@ async fn test_5000_node_pyramid() -> DagResult<()> { .add_task(task_fn::<(i32, i32), _, _>(|(a, b): (&i32, &i32)| a + b)) .depends_on((¤t_layer[0], ¤t_layer[1])); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; // We created 1024 ones and 1024 zeros at the bottom // The sum should propagate up to 1024 @@ -209,7 +209,7 @@ async fn test_wide_dag_1000_sources_1000_sinks() -> DagResult<()> { }) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; // Verify some sinks completed correctly let result0 = dag.get(sinks[0])?; @@ -258,7 +258,7 @@ async fn test_10000_node_linear_chain_segments() -> DagResult<()> { chain_ends.push(current); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; // All 10,000 tasks should have executed assert_eq!(counter.load(Ordering::Relaxed), 10_000); @@ -297,7 +297,7 @@ async fn test_100000_nodes_stress() -> DagResult<()> { .collect(); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(completed.load(Ordering::Relaxed), 100_000); Ok(()) diff --git a/tests/boundaries/type_limits.rs b/tests/boundaries/type_limits.rs index 4665506..1e3a544 100644 --- a/tests/boundaries/type_limits.rs +++ b/tests/boundaries/type_limits.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{DagResult, DagRunner, TaskHandle}; -use futures::FutureExt; + use std::marker::PhantomData; #[tokio::test] @@ -37,7 +37,7 @@ async fn test_zero_sized_types_throughout() -> DagResult<()> { })) .depends_on(t2); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(t1)?, Empty); assert_eq!(dag.get(t2)?, 42); @@ -72,7 +72,7 @@ async fn test_large_value_types() -> DagResult<()> { })) .depends_on(t1); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(t2)?, 1042); @@ -95,7 +95,7 @@ async fn test_unit_type_chains() -> DagResult<()> { .add_task(task_fn::<(), _, _>(|_: ()| "done")) .depends_on((t3,)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; dag.get(t1)?; // Verify no error dag.get(t2)?; @@ -145,7 +145,7 @@ async fn test_string_types_variety() -> DagResult<()> { )) .depends_on((t1, t2, t3, t4, t5, t6)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; let result = dag.get(combined)?; assert!(result.contains("static str")); @@ -178,7 +178,7 @@ async fn test_generic_type_constraints() -> DagResult<()> { )) .depends_on((t1, t2, t3)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; let result = dag.get(combined)?; assert!(result.contains("Wrapper(42)")); @@ -204,7 +204,7 @@ async fn test_reference_wrapper_types() -> DagResult<()> { .add_task(task_fn::(|&val: &i32| val * 2)) .depends_on(t1); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(t2)?, 200); @@ -231,7 +231,7 @@ async fn test_array_types() -> DagResult<()> { )) .depends_on((t1, t2)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(t3)?, 55); @@ -270,7 +270,7 @@ async fn test_enum_variants() -> DagResult<()> { )) .depends_on((t1, t2, t3, t4)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert!(dag.get(combined)?); diff --git a/tests/boundaries/zero_tasks.rs b/tests/boundaries/zero_tasks.rs index dafc44c..f884d8d 100644 --- a/tests/boundaries/zero_tasks.rs +++ b/tests/boundaries/zero_tasks.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{task, DagResult, DagRunner}; -use futures::FutureExt; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use tokio::task::yield_now; @@ -12,7 +12,7 @@ async fn test_empty_dag_execution() -> DagResult<()> { // Test that an empty DAG can be created and run without issues let dag = DagRunner::new(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; // Should complete successfully with no tasks Ok(()) @@ -25,7 +25,7 @@ async fn test_single_task_no_deps() -> DagResult<()> { let task = dag.add_task(task_fn::<(), _, _>(|_: ()| 42)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(task)?, 42); Ok(()) @@ -44,7 +44,7 @@ async fn test_single_task_unit_input_output() -> DagResult<()> { exec.fetch_add(1, Ordering::SeqCst); })); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; dag.get(task)?; // Just verify it doesn't error assert_eq!(executed.load(Ordering::SeqCst), 1); @@ -61,7 +61,7 @@ async fn test_zero_sized_type_task() -> DagResult<()> { let task = dag.add_task(task_fn::<(), _, _>(|_: ()| ZeroSized)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(task)?, ZeroSized); Ok(()) @@ -73,7 +73,7 @@ async fn test_empty_dag_multiple_runs() -> DagResult<()> { let dag = DagRunner::new(); for _ in 0..10 { - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; } Ok(()) @@ -88,7 +88,7 @@ async fn test_dag_with_only_source_tasks() -> DagResult<()> { .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i))) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; for (i, task) in tasks.into_iter().enumerate() { assert_eq!(dag.get(task)?, i); @@ -114,7 +114,7 @@ async fn test_single_self_contained_stateful_task() -> DagResult<()> { let dag = DagRunner::new(); let task = dag.add_task(Counter { value: 21 }); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(task)?, 42); Ok(()) @@ -128,11 +128,11 @@ async fn test_alternating_empty_and_filled_dag_runs() -> DagResult<()> { if i % 2 == 0 { // Empty DAG - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; } else { // DAG with a task let task = dag.add_task(task_fn::<(), _, _>(move |_: ()| i)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(task)?, i); } } @@ -152,7 +152,7 @@ async fn test_dag_with_never_type_simulation() -> DagResult<()> { // This task completes immediately but returns a Result containing "never" type let task = dag.add_task(task_fn::<(), _, _>(|_: ()| Result::::Ok(42))); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(task)?, Ok(42)); Ok(()) @@ -175,7 +175,7 @@ async fn test_minimal_task_with_minimal_async_work() -> DagResult<()> { let task = dag.add_task(YieldTask); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(task)?, 1337); Ok(()) diff --git a/tests/custom_type_test.rs b/tests/custom_type_test.rs index 50e4c66..7031744 100644 --- a/tests/custom_type_test.rs +++ b/tests/custom_type_test.rs @@ -1,6 +1,5 @@ // Test that custom types work without any trait implementations use dagx::{task, DagRunner, TaskHandle}; -use futures::FutureExt; #[derive(Clone, Debug, PartialEq)] struct Person { @@ -57,7 +56,7 @@ async fn test_custom_type_single_dependency() { let fetch = dag.add_task(FetchPerson); let process = dag.add_task(ProcessPerson).depends_on(fetch); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -73,7 +72,7 @@ async fn test_custom_type_passthrough() { let transform = dag.add_task(PersonToCompany).depends_on(fetch); let process = dag.add_task(ProcessPerson).depends_on(fetch); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -96,7 +95,7 @@ async fn test_custom_type_diamond_pattern() { let path_a = dag.add_task(ProcessPerson).depends_on(fetch); let path_b = dag.add_task(PersonToCompany).depends_on(fetch); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -116,7 +115,7 @@ async fn test_custom_type_fan_out() { let process2 = dag.add_task(ProcessPerson).depends_on(fetch); let process3 = dag.add_task(ProcessPerson).depends_on(fetch); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/tests/cycle_prevention.rs b/tests/cycle_prevention.rs index a6e2a41..7365115 100644 --- a/tests/cycle_prevention.rs +++ b/tests/cycle_prevention.rs @@ -4,7 +4,6 @@ //! prevents it through the type-state pattern. use dagx::{task, DagRunner, TaskHandle}; -use futures::FutureExt; /// Demonstrates that TaskBuilder is consumed by depends_on, /// preventing you from wiring the same task multiple times @@ -271,7 +270,7 @@ async fn test_type_safety_prevents_invalid_graphs() { let result = dag.add_task(Aggregate).depends_on((&t1, &t2)); // Execute and verify - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/tests/dependencies/resolution.rs b/tests/dependencies/resolution.rs index 8f4f6af..5f70e5c 100644 --- a/tests/dependencies/resolution.rs +++ b/tests/dependencies/resolution.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{task, DagResult, DagRunner, TaskHandle}; -use futures::FutureExt; + use std::sync::{Arc, Mutex}; #[tokio::test] @@ -79,7 +79,7 @@ async fn test_dependency_resolution_order() -> DagResult<()> { .depends_on(d) }; - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(e)?, 5); @@ -113,7 +113,7 @@ async fn test_transitive_dependencies() -> DagResult<()> { .add_task(task_fn::(|&x: &i32| x + 5)) .depends_on(b); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(a)?, 10); assert_eq!(dag.get(b)?, 20); @@ -136,7 +136,7 @@ async fn test_deep_chain_50_levels() -> DagResult<()> { handles.push(next); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(handles.last().unwrap())?, 50); Ok(()) @@ -155,7 +155,7 @@ async fn test_deep_chain_200_levels() -> DagResult<()> { .depends_on(current); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(current)?, 201); Ok(()) diff --git a/tests/dependencies/single.rs b/tests/dependencies/single.rs index 070d406..290e88f 100644 --- a/tests/dependencies/single.rs +++ b/tests/dependencies/single.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{DagResult, DagRunner, TaskHandle}; -use futures::FutureExt; + #[tokio::test] async fn test_single_dependency() -> DagResult<()> { @@ -13,7 +13,7 @@ async fn test_single_dependency() -> DagResult<()> { .add_task(task_fn::(|&x: &i32| x + 1)) .depends_on(source); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(dependent)?, 43); Ok(()) @@ -44,7 +44,7 @@ async fn test_dependency_data_flow() -> DagResult<()> { .add_task(task_fn::<(i32, i32), _, _>(|(s, p): (&i32, &i32)| s + p)) .depends_on((&sum, &product)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; // Note: sum and product are not sinks (final_result depends on them) // Only final_result is a sink and can be retrieved @@ -68,7 +68,7 @@ async fn test_dependencies_with_different_types() -> DagResult<()> { )) .depends_on((int_source, string_source, bool_source)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(combined)?, "hello: 42 (true)"); Ok(()) @@ -89,7 +89,7 @@ async fn test_shared_dependencies() -> DagResult<()> { }) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; // Verify all dependents got the correct value from shared for (i, handle) in dependents.iter().enumerate() { @@ -108,7 +108,7 @@ async fn test_multiple_source_nodes() -> DagResult<()> { .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i * 10))) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; for (i, source) in sources.into_iter().enumerate() { assert_eq!(dag.get(source)?, i as i32 * 10); diff --git a/tests/dependencies/tuples.rs b/tests/dependencies/tuples.rs index 3f78291..9d40635 100644 --- a/tests/dependencies/tuples.rs +++ b/tests/dependencies/tuples.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{DagResult, DagRunner}; -use futures::FutureExt; + #[tokio::test] async fn test_two_dependencies() -> DagResult<()> { @@ -14,7 +14,7 @@ async fn test_two_dependencies() -> DagResult<()> { .add_task(task_fn::<(i32, i32), _, _>(|(x, y): (&i32, &i32)| x + y)) .depends_on((a, b)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(sum)?, 30); Ok(()) @@ -33,7 +33,7 @@ async fn test_three_dependencies() -> DagResult<()> { )) .depends_on((a, b, c)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(sum)?, 6); Ok(()) @@ -53,7 +53,7 @@ async fn test_four_dependencies() -> DagResult<()> { )) .depends_on((&deps[0], &deps[1], &deps[2], &deps[3])); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(sum)?, 10); // 1+2+3+4 Ok(()) @@ -73,7 +73,7 @@ async fn test_five_dependencies() -> DagResult<()> { )) .depends_on((&deps[0], &deps[1], &deps[2], &deps[3], &deps[4])); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(sum)?, 15); // 1+2+3+4+5 Ok(()) @@ -93,7 +93,7 @@ async fn test_six_dependencies() -> DagResult<()> { )) .depends_on((&deps[0], &deps[1], &deps[2], &deps[3], &deps[4], &deps[5])); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(sum)?, 21); // 1+2+3+4+5+6 Ok(()) @@ -117,7 +117,7 @@ async fn test_seven_dependencies() -> DagResult<()> { &deps[0], &deps[1], &deps[2], &deps[3], &deps[4], &deps[5], &deps[6], )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(sum)?, 28); // 1+2+3+4+5+6+7 Ok(()) @@ -141,7 +141,7 @@ async fn test_eight_dependencies() -> DagResult<()> { &deps[0], &deps[1], &deps[2], &deps[3], &deps[4], &deps[5], &deps[6], &deps[7], )); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(sum)?, 36); // Sum of 1..=8 Ok(()) diff --git a/tests/errors/error_types.rs b/tests/errors/error_types.rs index 5c8a880..153d3ea 100644 --- a/tests/errors/error_types.rs +++ b/tests/errors/error_types.rs @@ -2,7 +2,6 @@ use crate::common::task_fn; use dagx::{DagError, DagRunner, TaskHandle}; -use futures::FutureExt; #[tokio::test] async fn test_result_not_found_error() { @@ -15,7 +14,7 @@ async fn test_result_not_found_error() { assert!(matches!(result, Err(DagError::ResultNotFound { .. }))); // Run and verify it works after - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -33,7 +32,7 @@ async fn test_nested_result_error_handling() { let task2 = dag.add_task(task_fn::<(), _, _>(|_: ()| Result::::Ok(42))); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -48,7 +47,7 @@ async fn test_option_none_not_error() { let task = dag.add_task(task_fn::<(), _, _>(|_: ()| None::)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -77,7 +76,7 @@ async fn test_custom_error_types() { }) })); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -106,7 +105,9 @@ async fn test_error_in_tuple_dependency() { .add_task(task_fn::<(i32, i32), _, _>(|(a, b): (&i32, &i32)| a + b)) .depends_on((ok_task, err_task)); - let result = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await; + let result = dag + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await; assert!(result.is_err()); assert!(dag.get(dependent).is_err()); @@ -124,7 +125,9 @@ async fn test_string_panic_vs_structured_panic() { panic!("Formatted panic: value={}, code={}", 42, "ERROR"); })); - let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await; + let _ = dag + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await; // Both should result in errors assert!(dag.get(string_panic).is_err()); @@ -144,7 +147,7 @@ async fn test_anyhow_like_error_chain() { outer })); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/tests/errors/propagation.rs b/tests/errors/propagation.rs index 060b308..4008f8d 100644 --- a/tests/errors/propagation.rs +++ b/tests/errors/propagation.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{task, DagRunner, TaskHandle}; -use futures::FutureExt; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -25,7 +25,9 @@ async fn test_linear_error_propagation() { .add_task(task_fn::(|&x: &i32| x * 2)) .depends_on(t3); - let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await; + let _ = dag + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await; // t1 succeeds assert_eq!(dag.get(t1).unwrap(), 1); @@ -64,7 +66,9 @@ async fn test_diamond_error_propagation() { .add_task(task_fn::<(i32, i32), _, _>(|(l, r): (&i32, &i32)| l + r)) .depends_on((&left, &right)); - let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await; + let _ = dag + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await; assert_eq!(dag.get(source).unwrap(), 100); assert_eq!(dag.get(left).unwrap(), 50); @@ -110,7 +114,9 @@ async fn test_error_stops_at_boundary() { } })); - let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await; + let _ = dag + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await; // Error branch: source runs but dependent doesn't assert!(dag.get(error_source).is_err()); @@ -149,7 +155,9 @@ async fn test_multiple_error_sources_convergence() { )) .depends_on((&error1, &error2, &success)); - let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await; + let _ = dag + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await; // All error sources fail assert!(dag.get(error1).is_err()); @@ -180,7 +188,9 @@ async fn test_error_in_wide_fanout() { }) .collect(); - let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await; + let _ = dag + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await; // Source fails assert!(dag.get(source).is_err()); @@ -225,7 +235,7 @@ async fn test_selective_error_propagation() { })) .depends_on(handler); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -291,7 +301,9 @@ async fn test_error_propagation_timing() { })) .depends_on(t2); - let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await; + let _ = dag + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await; let order = propagation_order.lock().clone(); @@ -364,7 +376,7 @@ async fn test_partial_branch_failure_propagation() { )) .depends_on((&a3, &b3, &c3)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .ok(); diff --git a/tests/errors/recovery.rs b/tests/errors/recovery.rs index 50bd630..588bd3a 100644 --- a/tests/errors/recovery.rs +++ b/tests/errors/recovery.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{DagResult, DagRunner}; -use futures::FutureExt; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; @@ -24,7 +24,9 @@ async fn test_fallback_on_error() -> DagResult<()> { let _selector = dag.add_task(task_fn::<(), _, _>(|_: ()| {})); // Run DAG - let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await; + let _ = dag + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await; // Primary should fail assert!(dag.get(primary).is_err()); @@ -58,7 +60,8 @@ async fn test_retry_pattern_simulation() -> DagResult<()> { } })); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Task completes with error on first "attempt" assert_eq!(dag.get(task)?, Err("Simulated failure")); @@ -99,7 +102,8 @@ async fn test_circuit_breaker_pattern() -> DagResult<()> { }) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // First few tasks fail and open circuit for task in tasks.into_iter().take(4) { @@ -131,7 +135,9 @@ async fn test_partial_recovery_with_defaults() -> DagResult<()> { vec![0, 10, 20, 0, 40, 50, 0, 70, 80, 0] // 0 for failed tasks })); - let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await; + let _ = dag + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await; let result = dag.get(aggregator)?; assert_eq!(result.len(), 10); @@ -159,7 +165,8 @@ async fn test_error_accumulation_pattern() -> DagResult<()> { }) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Check accumulated errors let error_list = errors.lock().clone(); @@ -197,7 +204,8 @@ async fn test_graceful_degradation() -> DagResult<()> { })) .depends_on(critical); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Critical path works assert_eq!(dag.get(final_task)?, 15); @@ -228,7 +236,8 @@ async fn test_error_boundary_isolation() -> DagResult<()> { .add_task(task_fn::(|&x: &i32| x + 5)) .depends_on(b2); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Group A fails (returns Err) assert_eq!(dag.get(a2)?, Err("Group A error")); @@ -261,7 +270,9 @@ async fn test_compensation_action() -> DagResult<()> { } })); - let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await; + let _ = dag + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await; // Action failed assert!(dag.get(action).is_err()); diff --git a/tests/execution/basic.rs b/tests/execution/basic.rs index a21a7e2..acbcb66 100644 --- a/tests/execution/basic.rs +++ b/tests/execution/basic.rs @@ -1,13 +1,12 @@ // Basic execution tests for DagRunner use crate::common::task_fn; -use futures::FutureExt; use dagx::*; #[tokio::test] async fn test_empty_dag() { let dag = DagRunner::new(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); // Should complete without error } @@ -16,7 +15,7 @@ async fn test_empty_dag() { async fn test_single_task() { let dag = DagRunner::new(); let task = dag.add_task(task_fn::<(), _, _>(|_: ()| 42)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); assert_eq!(dag.get(task).unwrap(), 42); @@ -40,7 +39,7 @@ async fn test_deep_chain() { .add_task(task_fn::(|&x: &i32| x + 1)) .depends_on(d); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -57,7 +56,7 @@ async fn test_wide_parallel() { .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i * 2))) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -81,7 +80,7 @@ async fn test_diamond_dependency() { .add_task(task_fn::<(i32, i32), _, _>(|(x, y): (&i32, &i32)| x + y)) .depends_on((&b, &c)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -107,7 +106,7 @@ async fn test_different_output_types() { age: 30, })); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -140,7 +139,7 @@ async fn test_multiple_sinks() { .add_task(task_fn::(|&x: &i32| x * 3)) .depends_on(branch2); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); // Should wait for both sinks @@ -162,7 +161,7 @@ async fn test_single_task_inline_path() { .add_task(task_fn::(|&x: &i32| x + 5)) .depends_on(t2); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -187,7 +186,7 @@ async fn test_multi_task_spawn_path() { .add_task(task_fn::(|&x: &i32| x + 3)) .depends_on(source); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -213,7 +212,7 @@ async fn test_producer_consumer_channels() { })) .depends_on(producer); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -238,7 +237,7 @@ async fn test_multi_consumer_fanout() { .add_task(task_fn::(|&x: &i32| x - 5)) .depends_on(producer); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -266,7 +265,7 @@ async fn test_compute_layers_with_dependents_tracking() { .add_task(task_fn::<(i32, i32), _, _>(|(l, r): (&i32, &i32)| l + r)) .depends_on((&left, &right)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -308,7 +307,7 @@ async fn test_layer_execution_order() { })) .depends_on(t2); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/tests/execution/integration.rs b/tests/execution/integration.rs index 524197c..0c808d7 100644 --- a/tests/execution/integration.rs +++ b/tests/execution/integration.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{DagResult, DagRunner, Task}; -use futures::FutureExt; + use std::time::Instant; use tokio::time::{sleep, Duration}; @@ -68,7 +68,8 @@ async fn integration_test_etl_pipeline() -> DagResult<()> { let final_result = dag.add_task(Add).depends_on((&combine12, &transform3)); // Execute pipeline - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Verify let result = dag.get(final_result)?; @@ -98,7 +99,8 @@ async fn integration_test_multi_layer_dag() -> DagResult<()> { // Layer 4: Final result let final_result = dag.add_task(Add).depends_on((&double_ab, &double_bc)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let result = dag.get(final_result)?; // (1+2)*2 + (2+3)*2 = 6 + 10 = 16 @@ -119,7 +121,8 @@ async fn integration_test_multiple_sinks() -> DagResult<()> { let branch2 = dag.add_task(Multiply::new(3)).depends_on(source); let branch3 = dag.add_task(Multiply::new(4)).depends_on(source); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // All branches should complete assert_eq!(dag.get(branch1)?, 20); @@ -145,7 +148,8 @@ async fn integration_test_diamond_cascade() -> DagResult<()> { let right2 = dag.add_task(Multiply::new(3)).depends_on(sink1); let sink2 = dag.add_task(Add).depends_on((&left2, &right2)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let result = dag.get(sink2)?; // Diamond 1: (10*2) + (10*3) = 50 @@ -176,7 +180,8 @@ async fn integration_test_wide_and_deep() -> DagResult<()> { final_nodes.push(step3); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Verify each chain for (i, node) in final_nodes.iter().enumerate() { @@ -195,7 +200,8 @@ async fn integration_test_error_recovery() -> DagResult<()> { let source = dag.add_task(LoadValue::new(42)); let transform = dag.add_task(Multiply::new(2)).depends_on(source); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Should be able to access results assert_eq!(dag.get(transform)?, 84); @@ -231,7 +237,8 @@ async fn integration_test_large_scale() -> DagResult<()> { }) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Verify some results assert_eq!(dag.get(aggregations[0])?, 1); // First two @@ -255,7 +262,8 @@ async fn integration_test_reused_values() -> DagResult<()> { // Combine some of the results let combined = dag.add_task(Add).depends_on((&double, &triple)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(double)?, 200); assert_eq!(dag.get(triple)?, 300); @@ -284,7 +292,8 @@ async fn integration_test_parallel_execution_timing() -> DagResult<()> { .collect(); let start = Instant::now(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let elapsed = start.elapsed(); // Verify all tasks completed @@ -345,7 +354,8 @@ async fn integration_test_parallel_vs_sequential_timing() -> DagResult<()> { .depends_on(seq2); let start = Instant::now(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let elapsed = start.elapsed(); // Verify results diff --git a/tests/execution/results.rs b/tests/execution/results.rs index fe3a81e..bf234a4 100644 --- a/tests/execution/results.rs +++ b/tests/execution/results.rs @@ -3,7 +3,6 @@ use crate::common::task_fn; use crate::common::tasks::Counter; use dagx::*; -use futures::FutureExt; #[tokio::test] async fn test_result_type() { @@ -14,7 +13,7 @@ async fn test_result_type() { Err::("error".to_string()) })); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -29,7 +28,8 @@ async fn test_dag_result_type_works() { // Test that DagResult<()> works with ? operator pattern async fn run_dag(dag: &DagRunner) -> DagResult<()> { - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; Ok(()) } @@ -64,7 +64,7 @@ async fn test_error_messages_are_helpful() { let err_msg = format!("{}", result.unwrap_err()); assert!(err_msg.contains("Result not found")); assert!(err_msg.contains("task #0")); // Task ID - assert!(err_msg.contains("Call dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))")); + assert!(err_msg.contains("Call dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })")); } #[tokio::test] @@ -73,7 +73,9 @@ async fn test_successful_execution_returns_ok() { let task = dag.add_task(task_fn::<(), _, _>(|_: ()| 42)); // Should return Ok(()) - let result = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await; + let result = dag + .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await; assert!(result.is_ok()); // Should be able to get value @@ -87,7 +89,7 @@ async fn test_stateful_tasks() { let input = dag.add_task(task_fn::<(), _, _>(|_: ()| 5)); let counter = dag.add_task(Counter::new(10)).depends_on(input); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -107,7 +109,7 @@ async fn test_task_fn_with_captured_state() { .add_task(task_fn::(move |&x: &i32| x * multiplier)) .depends_on(base); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/tests/parallelism/concurrent.rs b/tests/parallelism/concurrent.rs index 1aa6d28..64b0cc4 100644 --- a/tests/parallelism/concurrent.rs +++ b/tests/parallelism/concurrent.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{task, DagRunner}; -use futures::FutureExt; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -31,7 +31,7 @@ async fn test_concurrent_run_calls() { for _ in 0..5 { let dag = Arc::clone(&dag); - handles.spawn(async move { dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await }); + handles.spawn(async move { dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await }); } // Only one run should succeed, others should fail with concurrent execution error @@ -66,7 +66,7 @@ async fn test_concurrent_execution_and_building() { .collect(); // Execute the initial DAG - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -95,7 +95,7 @@ async fn test_concurrent_execution_and_building() { } // Execute again with the new tasks - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); } @@ -139,7 +139,7 @@ async fn test_dag_builder_thread_safety() { assert_eq!(counter.load(Ordering::SeqCst), 10); // Execute the combined DAG - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -161,7 +161,7 @@ async fn test_concurrent_get_operations() { .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i))) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -213,7 +213,7 @@ async fn test_no_race_in_dependency_tracking() { } // Execute - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -273,7 +273,7 @@ async fn test_high_concurrency_stress() { assert_eq!(completed.load(Ordering::SeqCst), 1000); // The DAG should still be executable - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); } @@ -363,7 +363,7 @@ async fn test_atomic_dag_operations() { assert_eq!(deps.len(), 50); // Execute and verify - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -416,7 +416,7 @@ async fn test_no_deadlock_on_circular_waiting() { assert!(result.is_ok(), "Deadlock detected - operation timed out"); // Should be able to execute the DAG - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); } diff --git a/tests/parallelism/execution.rs b/tests/parallelism/execution.rs index b78b3b9..6c25331 100644 --- a/tests/parallelism/execution.rs +++ b/tests/parallelism/execution.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{task, DagResult, DagRunner, TaskHandle}; -use futures::FutureExt; + use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use tokio::time::sleep; @@ -87,7 +87,7 @@ async fn test_layers_execute_in_parallel() -> DagResult<()> { .depends_on((&layer1[0], &layer1[1], &layer1[2])) }; - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(final_task)?, 33); // (0+1) + (10+1) + (20+1) @@ -213,7 +213,7 @@ async fn test_completion_order_matches_dependencies() -> DagResult<()> { .depends_on(c) }; - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(d)?, 6); // (1 + 2) * 2 @@ -313,7 +313,7 @@ async fn test_dependent_tasks_respect_ordering() -> DagResult<()> { .depends_on(c) }; - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; // Verify results assert_eq!(dag.get(a)?, 1); diff --git a/tests/parallelism/spawning.rs b/tests/parallelism/spawning.rs index 9e918dd..39a9a38 100644 --- a/tests/parallelism/spawning.rs +++ b/tests/parallelism/spawning.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{DagResult, DagRunner}; -use futures::FutureExt; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; @@ -27,7 +27,8 @@ async fn test_spawner_actually_spawns_tasks() -> DagResult<()> { .collect(); // Custom spawner that counts invocations - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Verify all tasks were spawned for (i, task) in tasks.into_iter().enumerate() { diff --git a/tests/parallelism/threads.rs b/tests/parallelism/threads.rs index 7845144..f424c69 100644 --- a/tests/parallelism/threads.rs +++ b/tests/parallelism/threads.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{task, DagResult, DagRunner, TaskHandle}; -use futures::FutureExt; + use std::collections::HashSet; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; @@ -52,7 +52,8 @@ async fn test_tasks_run_on_different_threads() -> DagResult<()> { }) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Verify all tasks completed for task in tasks.into_iter() { @@ -129,7 +130,8 @@ async fn test_concurrent_execution_with_atomic_counter() -> DagResult<()> { }) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Verify all tasks completed for (i, task) in tasks.into_iter().enumerate() { @@ -212,7 +214,8 @@ async fn test_diamond_parallel_execution() -> DagResult<()> { .add_task(task_fn::<(i32, i32), _, _>(|(a, b): (&i32, &i32)| a + b)) .depends_on((¶llel1, ¶llel2)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(sink)?, 500); // (100 * 2) + (100 * 3) @@ -279,7 +282,8 @@ async fn test_massive_parallel_fanout() -> DagResult<()> { .collect(); let start = Instant::now(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let elapsed = start.elapsed(); // All should complete diff --git a/tests/parallelism/timing.rs b/tests/parallelism/timing.rs index 76b1b8d..5d1aaa5 100644 --- a/tests/parallelism/timing.rs +++ b/tests/parallelism/timing.rs @@ -1,7 +1,7 @@ //! Tests for timing and speedup verification use dagx::{task, DagResult, DagRunner}; -use futures::FutureExt; + use std::time::{Duration, Instant}; use tokio::time::sleep; @@ -29,7 +29,8 @@ async fn test_parallel_execution_speedup() -> DagResult<()> { .collect(); let start = Instant::now(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let elapsed = start.elapsed(); // Verify all tasks completed diff --git a/tests/patterns/butterfly.rs b/tests/patterns/butterfly.rs index 1d90275..2b53112 100644 --- a/tests/patterns/butterfly.rs +++ b/tests/patterns/butterfly.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{DagResult, DagRunner, TaskHandle}; -use futures::FutureExt; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; @@ -52,7 +52,8 @@ async fn test_butterfly_network() -> DagResult<()> { .depends_on((&layer1[3], &layer1[7])), ]; - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Verify some outputs exist for task in &layer2 { @@ -107,7 +108,8 @@ async fn test_recursive_butterfly() -> DagResult<()> { let outputs = build_butterfly(&dag, inputs); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(outputs.len(), 4); for output in outputs { @@ -158,7 +160,8 @@ async fn test_benes_network() -> DagResult<()> { .depends_on((&stage1[1], &stage1[3])), ]; - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; for output in outputs { let _ = dag.get(output)?; @@ -212,7 +215,8 @@ async fn test_shuffle_exchange_network() -> DagResult<()> { }) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(counter.load(Ordering::Relaxed), 8); diff --git a/tests/patterns/complex.rs b/tests/patterns/complex.rs index bf00063..4fee8b2 100644 --- a/tests/patterns/complex.rs +++ b/tests/patterns/complex.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{DagResult, DagRunner, TaskHandle}; -use futures::FutureExt; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; @@ -60,7 +60,8 @@ async fn test_workflow_with_conditional_paths() -> DagResult<()> { })) .depends_on(combine); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(process)?, 84); let result = dag.get(final_result)?; @@ -147,7 +148,8 @@ async fn test_map_reduce_pattern() -> DagResult<()> { )) .depends_on((&reducers[0], &reducers[1], &reducers[2], &reducers[3])); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let result = dag.get(final_task)?; assert!(result > 0); @@ -198,7 +200,8 @@ async fn test_pipeline_with_feedback() -> DagResult<()> { })) .depends_on(stage3); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(stage1)?, 2); assert_eq!(dag.get(stage3)?, 12); @@ -244,7 +247,8 @@ async fn test_scatter_gather_pattern() -> DagResult<()> { )) .depends_on((&workers[0], &workers[1], &workers[2], &workers[3])); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Sum of 10+20+30+40+50+60+70+80 = 360 assert_eq!(dag.get(gather)?, 360); @@ -323,7 +327,8 @@ async fn test_fork_join_with_barriers() -> DagResult<()> { })) .depends_on(barrier); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let _ = dag.get(join)?; assert_eq!(phase_counter.load(Ordering::SeqCst), 2); diff --git a/tests/patterns/grid.rs b/tests/patterns/grid.rs index 40a2a8c..95c8a29 100644 --- a/tests/patterns/grid.rs +++ b/tests/patterns/grid.rs @@ -2,7 +2,6 @@ use crate::common::task_fn; use dagx::{DagResult, DagRunner}; -use futures::FutureExt; #[tokio::test] async fn test_2d_grid_pattern() -> DagResult<()> { @@ -37,7 +36,8 @@ async fn test_2d_grid_pattern() -> DagResult<()> { } } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Check corner values let bottom_right = dag.get(grid[3][3].as_ref().unwrap())?; @@ -102,7 +102,8 @@ async fn test_3d_grid_pattern() -> DagResult<()> { } } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Check that all cells have values for row in &grid { @@ -176,7 +177,8 @@ async fn test_hexagonal_grid() -> DagResult<()> { hex_grid.insert((q, r), task); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Verify center and some ring values assert_eq!(dag.get(hex_grid.get(&(0, 0)).unwrap())?, 100); @@ -225,7 +227,8 @@ async fn test_toroidal_grid() -> DagResult<()> { } } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Verify all torus tasks completed for task in torus_tasks { diff --git a/tests/patterns/tree.rs b/tests/patterns/tree.rs index 2346697..ef713e9 100644 --- a/tests/patterns/tree.rs +++ b/tests/patterns/tree.rs @@ -2,7 +2,6 @@ use crate::common::task_fn; use dagx::{DagResult, DagRunner, TaskHandle}; -use futures::FutureExt; #[tokio::test] async fn test_binary_tree_reduction() -> DagResult<()> { @@ -36,7 +35,8 @@ async fn test_binary_tree_reduction() -> DagResult<()> { let root = current_level[0]; - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Sum of powers of 2 from 0 to 7: 1+2+4+8+16+32+64+128 = 255 assert_eq!(dag.get(root)?, 255); @@ -67,7 +67,8 @@ async fn test_n_ary_tree() -> DagResult<()> { let root = build_tree(&dag, 3, 1); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let result = dag.get(root)?; assert!(result > 0); @@ -114,7 +115,8 @@ async fn test_unbalanced_tree() -> DagResult<()> { .add_task(task_fn::<(i32, i32), _, _>(|(a, b): (&i32, &i32)| a * b)) .depends_on((&a, &b)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(f)?, 1); assert_eq!(dag.get(c)?, 2); @@ -166,7 +168,8 @@ async fn test_trie_like_structure() -> DagResult<()> { .add_task(task_fn::(|v: &String| format!("{}/log", v))) .depends_on(var); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(usr_bin)?, "root/usr/bin"); assert_eq!(dag.get(usr_lib)?, "root/usr/lib"); @@ -221,7 +224,8 @@ async fn test_balanced_k_ary_tree() -> DagResult<()> { let root = create_balanced_tree(&dag, DEPTH, K, 0); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let result = dag.get(root)?; assert!(result > 0); @@ -268,7 +272,8 @@ async fn test_merkle_tree_pattern() -> DagResult<()> { let root_hash = current_level[0]; - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let result = dag.get(root_hash)?; assert!(result.starts_with("hash")); diff --git a/tests/performance/mod.rs b/tests/performance/mod.rs index f402a53..8a73a60 100644 --- a/tests/performance/mod.rs +++ b/tests/performance/mod.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::DagRunner; -use futures::FutureExt; + use std::sync::Arc; use std::time::{Duration, Instant}; @@ -27,7 +27,7 @@ async fn test_deep_chain_100_levels() { // Execute let start = Instant::now(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); let exec_time = start.elapsed(); @@ -88,7 +88,7 @@ async fn test_stress_mixed_patterns() { // Execute let start = Instant::now(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); let exec_time = start.elapsed(); diff --git a/tests/runtimes/compatibility.rs b/tests/runtimes/compatibility.rs index dd206af..5073235 100644 --- a/tests/runtimes/compatibility.rs +++ b/tests/runtimes/compatibility.rs @@ -1,7 +1,7 @@ //! Runtime integration tests - verify dagx works correctly across different runtimes use dagx::{DagResult, DagRunner}; -use futures::FutureExt; + struct Value(i32); #[dagx::task] @@ -27,7 +27,7 @@ async fn runtime_integration_tokio() -> DagResult<()> { let b = dag.add_task(Value(20)); let sum = dag.add_task(Add).depends_on((a, b)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; assert_eq!(dag.get(sum)?, 30); Ok(()) diff --git a/tests/runtimes/quirky.rs b/tests/runtimes/quirky.rs index 74e6f47..bad3970 100644 --- a/tests/runtimes/quirky.rs +++ b/tests/runtimes/quirky.rs @@ -15,8 +15,8 @@ use crate::common::task_fn; use dagx::{DagRunner, TaskHandle}; -use futures::task::SpawnExt; use futures_executor::ThreadPool; +use futures_util::task::SpawnExt; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; diff --git a/tests/task_patterns/combined.rs b/tests/task_patterns/combined.rs index d29b81f..fa15c53 100644 --- a/tests/task_patterns/combined.rs +++ b/tests/task_patterns/combined.rs @@ -3,7 +3,7 @@ use crate::common::task_fn; // Comprehensive test showing all three patterns working together use dagx::*; -use futures::FutureExt; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; @@ -53,7 +53,7 @@ async fn test_all_three_task_patterns() { // Pattern 3: Mutable accumulation let accumulated = dag.add_task(MutableAccumulator(10)).depends_on(doubled); // 10 + 16 = 26 - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -90,7 +90,7 @@ async fn test_parallel_execution() { 3 })); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -111,7 +111,7 @@ async fn test_task_fn_with_captured_state() { .add_task(task_fn::(move |&x: &i32| x * multiplier)) .depends_on(base); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/tests/task_patterns/readonly.rs b/tests/task_patterns/readonly.rs index eb076e0..f2b1c70 100644 --- a/tests/task_patterns/readonly.rs +++ b/tests/task_patterns/readonly.rs @@ -3,7 +3,6 @@ use crate::common::task_fn; // These tests verify the read-only state pattern using &self use dagx::*; -use futures::FutureExt; // Read-only state tasks struct ReadOnlyValue(i32); @@ -43,7 +42,7 @@ async fn test_readonly_state_basic() { let value = dag.add_task(ReadOnlyValue(42)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); assert_eq!(dag.get(value).unwrap(), 42); @@ -57,7 +56,7 @@ async fn test_readonly_state_with_input() { let input = dag.add_task(task_fn::<(), _, _>(|_: ()| 5)); let result = dag.add_task(ReadOnlyMultiplier(7)).depends_on(input); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); assert_eq!(dag.get(result).unwrap(), 35); @@ -84,7 +83,7 @@ async fn test_readonly_state_multiple_uses() { }; let result2 = dag.add_task(config2).depends_on(input2); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); assert_eq!(dag.get(result1).unwrap(), 35); // (10 * 3) + 5 diff --git a/tests/task_patterns/stateless.rs b/tests/task_patterns/stateless.rs index 3b4d890..b48baa5 100644 --- a/tests/task_patterns/stateless.rs +++ b/tests/task_patterns/stateless.rs @@ -1,6 +1,5 @@ // Stateless task pattern tests use crate::common::task_fn; -use futures::FutureExt; use dagx::*; @@ -86,7 +85,7 @@ async fn test_stateless_basic() { let y = dag.add_task(task_fn::<(), _, _>(|_: ()| 20)); let sum = dag.add_task(StatelessAdd).depends_on((x, y)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); assert_eq!(dag.get(sum).unwrap(), 30); @@ -100,7 +99,7 @@ async fn test_stateless_single_input() { let input = dag.add_task(task_fn::<(), _, _>(|_: ()| 21)); let doubled = dag.add_task(StatelessDouble).depends_on(input); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); assert_eq!(dag.get(doubled).unwrap(), 42); @@ -113,7 +112,7 @@ async fn test_stateless_no_input() { let source = dag.add_task(StatelessSource); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); assert_eq!(dag.get(source).unwrap(), 42); @@ -130,7 +129,7 @@ async fn test_stateless_multiple_inputs() { let result = dag.add_task(StatelessFormat).depends_on((num, text, flag)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); assert_eq!(dag.get(result).unwrap(), "test: 42 (true)"); @@ -169,7 +168,7 @@ async fn test_stateless_mixed_with_stateful() { // Stateful counter let counted = dag.add_task(Counter::new(10)).depends_on(product); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -187,7 +186,7 @@ async fn test_stateless_chain() { let step2 = dag.add_task(StatelessInc).depends_on(step1); let step3 = dag.add_task(StatelessInc).depends_on(step2); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -208,7 +207,7 @@ async fn test_stateless_parallel() { }) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -229,7 +228,7 @@ async fn test_stateless_with_string_types() { let greeting = dag.add_task(StatelessConcat).depends_on((hello, world)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/tests/timing/concurrent_access.rs b/tests/timing/concurrent_access.rs index 07207f8..0477d85 100644 --- a/tests/timing/concurrent_access.rs +++ b/tests/timing/concurrent_access.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{task, DagResult, DagRunner, TaskHandle}; -use futures::FutureExt; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -32,7 +32,8 @@ async fn test_concurrent_result_retrieval() -> DagResult<()> { .collect(); // Execute the DAG - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Now try to get results concurrently from multiple threads let mut get_handles = Vec::new(); @@ -83,7 +84,8 @@ async fn test_simultaneous_dag_building_and_execution() -> DagResult<()> { tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; // Execute the DAG - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Get the additional tasks let additional_tasks = build_handle.await.unwrap(); @@ -97,7 +99,8 @@ async fn test_simultaneous_dag_building_and_execution() -> DagResult<()> { // (they need another run cycle) if building_complete.load(Ordering::SeqCst) { // Run again for the new tasks - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Now they should have results for (i, task) in additional_tasks.iter().enumerate() { @@ -139,7 +142,8 @@ async fn test_concurrent_dependency_resolution() -> DagResult<()> { }) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Check all completed assert_eq!(resolution_counter.load(Ordering::SeqCst), 101); @@ -212,7 +216,8 @@ async fn test_read_write_race_patterns() -> DagResult<()> { }) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Verify readers saw their writer's data for reader in &readers { @@ -265,7 +270,8 @@ async fn test_atomic_counter_races() -> DagResult<()> { }) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Final counter should be 200 * 100 = 20000 assert_eq!(counter.load(Ordering::SeqCst), 20000); @@ -299,7 +305,8 @@ async fn test_multiple_dag_runners_interaction() -> DagResult<()> { }) .collect(); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Collect results let mut results = Vec::new(); diff --git a/tests/timing/interleaving.rs b/tests/timing/interleaving.rs index 5b4c31a..c47e691 100644 --- a/tests/timing/interleaving.rs +++ b/tests/timing/interleaving.rs @@ -2,7 +2,7 @@ use crate::common::task_fn; use dagx::{task, DagResult, DagRunner, TaskHandle}; -use futures::FutureExt; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use tokio::sync::Barrier; @@ -58,7 +58,8 @@ async fn test_forced_interleaving_with_barriers() -> DagResult<()> { )) .depends_on((t1, t2, t3)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(collector)?, 6); @@ -186,7 +187,8 @@ async fn test_alternating_execution_pattern() -> DagResult<()> { .depends_on(chain_b); } - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Check alternating pattern let event_list = events.lock().clone(); @@ -262,7 +264,8 @@ async fn test_layered_interleaving() -> DagResult<()> { }) .depends_on((&layer2[0], &layer2[1])); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Sum of 0+1+2+3 = 6 assert_eq!(dag.get(layer3)?, 6); @@ -368,7 +371,8 @@ async fn test_cross_branch_synchronization() -> DagResult<()> { .add_task(task_fn::<(i32, i32), _, _>(|(a, b): (&i32, &i32)| a + b)) .depends_on((&a2, &b2)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(merge)?, 13); // 2 + 11 diff --git a/tests/tracing/with_tracing.rs b/tests/tracing/with_tracing.rs index 1ef3d88..fc0f77b 100644 --- a/tests/tracing/with_tracing.rs +++ b/tests/tracing/with_tracing.rs @@ -1,7 +1,7 @@ //! Tests with tracing feature enabled use dagx::{task, DagRunner}; -use futures::FutureExt; + use tracing_subscriber::{fmt, EnvFilter}; struct Value(i32); @@ -45,7 +45,7 @@ async fn test_tracing_with_subscriber() { let b = dag.add_task(Value(3)); let sum = dag.add_task(Add).depends_on((a, b)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -68,7 +68,7 @@ async fn test_tracing_with_complex_dag() { let right = dag.add_task(Multiply).depends_on((&source, &source)); let sink = dag.add_task(Add).depends_on((&left, &right)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -94,7 +94,7 @@ async fn test_tracing_inline_execution() { let b = dag.add_task(Add).depends_on((&a, &a)); let c = dag.add_task(Multiply).depends_on((&b, &b)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -128,7 +128,7 @@ async fn test_tracing_multiple_layers() { // Layer 2 let final_sum = dag.add_task(Add).depends_on((&sum12, &mul23)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/tests/tracing/without_tracing.rs b/tests/tracing/without_tracing.rs index cf05260..7b2ff3c 100644 --- a/tests/tracing/without_tracing.rs +++ b/tests/tracing/without_tracing.rs @@ -1,7 +1,6 @@ //! Tests that the library works correctly without tracing feature use dagx::{task, DagRunner}; -use futures::FutureExt; struct Value(i32); @@ -29,7 +28,7 @@ async fn test_basic_dag_without_tracing() { let b = dag.add_task(Value(3)); let sum = dag.add_task(Add).depends_on((a, b)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -50,7 +49,7 @@ async fn test_complex_dag_without_tracing() { let sum34 = dag.add_task(Add).depends_on((v3, v4)); let final_sum = dag.add_task(Add).depends_on((&sum12, &sum34)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); diff --git a/tests/type_safety/mod.rs b/tests/type_safety/mod.rs index 55465dc..49504cd 100644 --- a/tests/type_safety/mod.rs +++ b/tests/type_safety/mod.rs @@ -1,6 +1,6 @@ // Type safety tests for TaskHandle use crate::common::task_fn; -use futures::FutureExt; + // These tests verify dagx's compile-time type safety guarantees. // Tests cover both runtime behavior (wrong types return None) and // compile-time checking (type mismatches fail to compile). @@ -16,7 +16,7 @@ async fn test_get_with_wrong_type_returns_none() { let int_task = dag.add_task(task_fn::<(), _, _>(|_: ()| 42i32)); // Run DAG - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -37,7 +37,7 @@ async fn test_handle_preserves_type() { let string_task = dag.add_task(task_fn::<(), _, _>(|_: ()| "hello".to_string())); let int_task = dag.add_task(task_fn::<(), _, _>(|_: ()| 42i32)); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); @@ -63,7 +63,7 @@ async fn test_type_safety_handle_reuse() { .add_task(task_fn::(|&x: &i32| x * 3)) .depends_on(source); - dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)) + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap();