Skip to content

Commit a1d663f

Browse files
eabaAarononthewebArkatufus
authored
[Obsolete][CS0618] AwaitResult > Use ShouldCompleteWithin instead (#6498)
* [Obsolete][CS0618] `AwaitResult` > `Use ShouldCompleteWithin instead` * fix * more fixed * [test] Within `2.Seconds()` * [test][FileSinkSpec] change to `Async`s * Fixes * [revert] `ValveSpec` seq.Invoking* * [change] `await this.AssertAllStagesStoppedAsync(async() => { }, materializer);` * improves `FlowAggregateAsyncSpec`, `FlowSelectAsyncSpec` * fixed --------- Co-authored-by: Aaron Stannard <[email protected]> Co-authored-by: Gregorius Soedharmo <[email protected]>
1 parent 7ca22af commit a1d663f

24 files changed

+448
-303
lines changed

‎src/core/Akka.Streams.Tests/Dsl/AttributesSpec.cs

+8-5
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,9 @@
1111
using Akka.Streams.Implementation;
1212
using Akka.Streams.TestKit;
1313
using Akka.TestKit;
14+
using Akka.TestKit.Extensions;
1415
using FluentAssertions;
16+
using FluentAssertions.Extensions;
1517
using Xunit;
1618

1719
namespace Akka.Streams.Tests.Dsl
@@ -30,26 +32,27 @@ public AttributesSpec()
3032
Attributes.CreateName("a").And(Attributes.CreateName("b")).And(Attributes.CreateInputBuffer(1, 2));
3133

3234
[Fact]
33-
public void Attributes_must_be_overridable_on_a_module_basis()
35+
public async Task Attributes_must_be_overridable_on_a_module_basis()
3436
{
3537
var runnable =
3638
Source.Empty<NotUsed>()
3739
.ToMaterialized(AttributesSink.Create().WithAttributes(Attributes.CreateName("new-name")),
3840
Keep.Right);
3941
var task = runnable.Run(Materializer);
4042

41-
task.AwaitResult().GetAttribute<Attributes.Name>().Value.Should().Contain("new-name");
43+
var complete = await task.ShouldCompleteWithin(3.Seconds());
44+
complete.GetAttribute<Attributes.Name>().Value.Should().Contain("new-name");
4245
}
4346

4447
[Fact]
45-
public void Attributes_must_keep_the_outermost_attribute_as_the_least_specific()
48+
public async Task Attributes_must_keep_the_outermost_attribute_as_the_least_specific()
4649
{
4750
var task = Source.Empty<NotUsed>()
4851
.ToMaterialized(AttributesSink.Create(), Keep.Right)
4952
.WithAttributes(Attributes.CreateName("new-name"))
5053
.Run(Materializer);
51-
52-
task.AwaitResult().GetAttribute<Attributes.Name>().Value.Should().Contain("attributesSink");
54+
var complete = await task.ShouldCompleteWithin(3.Seconds());
55+
complete.GetAttribute<Attributes.Name>().Value.Should().Contain("attributesSink");
5356
}
5457

5558
[Fact]

‎src/core/Akka.Streams.Tests/Dsl/FlowAggregateAsyncSpec.cs

+24-17
Original file line numberDiff line numberDiff line change
@@ -56,43 +56,47 @@ private static Sink<int, Task<int>> AggregateSink
5656

5757

5858
[Fact]
59-
public void A_AggregateAsync_must_work_when_using_Source_AggregateAsync()
59+
public async Task A_AggregateAsync_must_work_when_using_Source_AggregateAsync()
6060
{
61-
this.AssertAllStagesStopped(() =>
61+
await this.AssertAllStagesStoppedAsync(async() =>
6262
{
6363
var task = AggregateSource.RunWith(Sink.First<int>(), Materializer);
64-
task.AwaitResult().Should().Be(Expected);
64+
var complete = await task.ShouldCompleteWithin(3.Seconds());
65+
complete.Should().Be(Expected);
6566
}, Materializer);
6667
}
6768

6869
[Fact]
6970
public void A_AggregateAsync_must_work_when_using_Sink_AggregateAsync()
7071
{
71-
this.AssertAllStagesStopped(() =>
72+
this.AssertAllStagesStopped(async() =>
7273
{
7374
var task = InputSource.RunWith(AggregateSink, Materializer);
74-
task.AwaitResult().Should().Be(Expected);
75+
var complete = await task.ShouldCompleteWithin(3.Seconds());
76+
complete.Should().Be(Expected);
7577
}, Materializer);
7678
}
7779

7880
[LocalFact(SkipLocal = "Racy on Azure DevOps")]
7981
public void A_AggregateAsync_must_work_when_using_Flow_AggregateAsync()
8082
{
8183
var flowTimeout = TimeSpan.FromMilliseconds(FlowDelayInMs*Input.Count()) + TimeSpan.FromSeconds(3);
82-
this.AssertAllStagesStopped(() =>
84+
this.AssertAllStagesStopped(async() =>
8385
{
8486
var task = InputSource.Via(AggregateFlow).RunWith(Sink.First<int>(), Materializer);
85-
task.AwaitResult(flowTimeout).Should().Be(Expected);
87+
var complete = await task.ShouldCompleteWithin(flowTimeout);
88+
complete.Should().Be(Expected);
8689
}, Materializer);
8790
}
8891

8992
[Fact]
9093
public void A_AggregateAsync_must_work_when_using_Source_AggregateAsync_and_Flow_AggregateAsync_and_Sink_AggregateAsync()
9194
{
92-
this.AssertAllStagesStopped(() =>
95+
this.AssertAllStagesStopped(async() =>
9396
{
9497
var task = AggregateSource.Via(AggregateFlow).RunWith(AggregateSink, Materializer);
95-
task.AwaitResult().Should().Be(Expected);
98+
var complete = await task.ShouldCompleteWithin(3.Seconds());
99+
complete.Should().Be(Expected);
96100
}, Materializer);
97101
}
98102

@@ -277,11 +281,11 @@ await this.AssertAllStagesStoppedAsync(async () =>
277281
}
278282

279283
[Fact]
280-
public void A_AggregateAsync_must_finish_after_task_failure()
284+
public async Task A_AggregateAsync_must_finish_after_task_failure()
281285
{
282-
this.AssertAllStagesStopped(() =>
286+
await this.AssertAllStagesStoppedAsync(async() =>
283287
{
284-
Source.From(Enumerable.Range(1, 3)).AggregateAsync(1, (_, n) => Task.Run(() =>
288+
var complete = await Source.From(Enumerable.Range(1, 3)).AggregateAsync(1, (_, n) => Task.Run(() =>
285289
{
286290
if (n == 3)
287291
throw new Exception("err3b");
@@ -290,7 +294,8 @@ public void A_AggregateAsync_must_finish_after_task_failure()
290294
.WithAttributes(ActorAttributes.CreateSupervisionStrategy(Deciders.ResumingDecider))
291295
.Grouped(10)
292296
.RunWith(Sink.First<IEnumerable<int>>(), Materializer)
293-
.AwaitResult().Should().BeEquivalentTo(2);
297+
.ShouldCompleteWithin(3.Seconds());
298+
complete.Should().BeEquivalentTo(2);
294299
}, Materializer);
295300
}
296301

@@ -417,22 +422,24 @@ public void A_AggregateAsync_must_handle_cancel_properly()
417422
[Fact]
418423
public void A_AggregateAsync_must_complete_task_and_return_zero_given_an_empty_stream()
419424
{
420-
this.AssertAllStagesStopped(() =>
425+
this.AssertAllStagesStopped(async() =>
421426
{
422427
var task = Source.From(Enumerable.Empty<int>())
423428
.RunAggregateAsync(0, (acc, element) => Task.FromResult(acc + element), Materializer);
424-
task.AwaitResult(RemainingOrDefault).ShouldBe(0);
429+
var complete = await task.ShouldCompleteWithin(RemainingOrDefault);
430+
complete.ShouldBe(0);
425431
}, Materializer);
426432
}
427433

428434
[Fact]
429435
public void A_AggregateAsync_must_complete_task_and_return_zero_and_item_given_a_stream_of_one_item()
430436
{
431-
this.AssertAllStagesStopped(() =>
437+
this.AssertAllStagesStopped(async() =>
432438
{
433439
var task = Source.Single(100)
434440
.RunAggregateAsync(5, (acc, element) => Task.FromResult(acc + element), Materializer);
435-
task.AwaitResult(RemainingOrDefault).ShouldBe(105);
441+
var complete = await task.ShouldCompleteWithin(RemainingOrDefault);
442+
complete.ShouldBe(105);
436443
}, Materializer);
437444
}
438445
}

‎src/core/Akka.Streams.Tests/Dsl/FlowAggregateSpec.cs

+26-16
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,10 @@
1414
using Akka.Streams.TestKit;
1515
using Akka.TestKit;
1616
using FluentAssertions;
17+
using Akka.TestKit.Extensions;
1718
using Xunit;
1819
using Xunit.Abstractions;
20+
using FluentAssertions.Extensions;
1921

2022
namespace Akka.Streams.Tests.Dsl
2123
{
@@ -38,50 +40,55 @@ public FlowAggregateSpec(ITestOutputHelper helper) : base(helper)
3840
[Fact]
3941
public void A_Aggregate_must_work_when_using_Source_RunAggregate()
4042
{
41-
this.AssertAllStagesStopped(() =>
43+
this.AssertAllStagesStopped(async() =>
4244
{
4345
var task = InputSource.RunAggregate(0, (sum, i) => sum + i, Materializer);
44-
task.AwaitResult().Should().Be(Expected);
46+
var complete = await task.ShouldCompleteWithin(3.Seconds());
47+
complete.Should().Be(Expected);
4548
}, Materializer);
4649
}
4750

4851
[Fact]
4952
public void A_Aggregate_must_work_when_using_Source_Aggregate()
5053
{
51-
this.AssertAllStagesStopped(() =>
54+
this.AssertAllStagesStopped(async() =>
5255
{
5356
var task = AggregateSource.RunWith(Sink.First<int>(), Materializer);
54-
task.AwaitResult().Should().Be(Expected);
57+
var complete = await task.ShouldCompleteWithin(3.Seconds());
58+
complete.Should().Be(Expected);
5559
}, Materializer);
5660
}
5761

5862
[Fact]
5963
public void A_Aggregate_must_work_when_using_Sink_Aggregate()
6064
{
61-
this.AssertAllStagesStopped(() =>
65+
this.AssertAllStagesStopped(async() =>
6266
{
6367
var task = InputSource.RunWith(AggregateSink, Materializer);
64-
task.AwaitResult().Should().Be(Expected);
68+
var complete = await task.ShouldCompleteWithin(3.Seconds());
69+
complete.Should().Be(Expected);
6570
}, Materializer);
6671
}
6772

6873
[Fact]
6974
public void A_Aggregate_must_work_when_using_Flow_Aggregate()
7075
{
71-
this.AssertAllStagesStopped(() =>
76+
this.AssertAllStagesStopped(async() =>
7277
{
7378
var task = InputSource.Via(AggregateFlow).RunWith(Sink.First<int>(), Materializer);
74-
task.AwaitResult().Should().Be(Expected);
79+
var complete = await task.ShouldCompleteWithin(3.Seconds());
80+
complete.Should().Be(Expected);
7581
}, Materializer);
7682
}
7783

7884
[Fact]
7985
public void A_Aggregate_must_work_when_using_Source_Aggregate_and_Flow_Aggregate_and_Sink_Aggregate()
8086
{
81-
this.AssertAllStagesStopped(() =>
87+
this.AssertAllStagesStopped(async() =>
8288
{
8389
var task = AggregateSource.Via(AggregateFlow).RunWith(AggregateSink, Materializer);
84-
task.AwaitResult().Should().Be(Expected);
90+
var complete = await task.ShouldCompleteWithin(3.Seconds());
91+
complete.Should().Be(Expected);
8592
}, Materializer);
8693
}
8794

@@ -129,7 +136,7 @@ public void
129136
[Fact]
130137
public void A_Aggregate_must_resume_with_the_accumulated_state_when_the_aggregating_funtion_throws_and_the_supervisor_strategy_decides_to_resume()
131138
{
132-
this.AssertAllStagesStopped(() =>
139+
this.AssertAllStagesStopped(async() =>
133140
{
134141
var error = new Exception("boom");
135142
var aggregate = Sink.Aggregate(0, (int x, int y) =>
@@ -142,14 +149,15 @@ public void A_Aggregate_must_resume_with_the_accumulated_state_when_the_aggregat
142149
var task = InputSource.RunWith(
143150
aggregate.WithAttributes(ActorAttributes.CreateSupervisionStrategy(Deciders.ResumingDecider)),
144151
Materializer);
145-
task.AwaitResult().Should().Be(Expected - 50);
152+
var complete = await task.ShouldCompleteWithin(3.Seconds());
153+
complete.Should().Be(Expected - 50);
146154
}, Materializer);
147155
}
148156

149157
[Fact]
150158
public void A_Aggregate_must_resume_and_reset_the_state_when_the_aggregating_funtion_throws_and_the_supervisor_strategy_decides_to_restart()
151159
{
152-
this.AssertAllStagesStopped(() =>
160+
this.AssertAllStagesStopped(async() =>
153161
{
154162
var error = new Exception("boom");
155163
var aggregate = Sink.Aggregate(0, (int x, int y) =>
@@ -162,18 +170,20 @@ public void A_Aggregate_must_resume_and_reset_the_state_when_the_aggregating_fun
162170
var task = InputSource.RunWith(
163171
aggregate.WithAttributes(ActorAttributes.CreateSupervisionStrategy(Deciders.RestartingDecider)),
164172
Materializer);
165-
task.AwaitResult().Should().Be(Enumerable.Range(51, 50).Sum());
173+
var complete = await task.ShouldCompleteWithin(3.Seconds());
174+
complete.Should().Be(Enumerable.Range(51, 50).Sum());
166175
}, Materializer);
167176
}
168177

169178
[Fact]
170179
public void A_Aggregate_must_complete_task_and_return_zero_given_an_empty_stream()
171180
{
172-
this.AssertAllStagesStopped(() =>
181+
this.AssertAllStagesStopped(async() =>
173182
{
174183
var task = Source.From(Enumerable.Empty<int>())
175184
.RunAggregate(0, (acc, element) => acc + element, Materializer);
176-
task.AwaitResult().ShouldBe(0);
185+
var complete = await task.ShouldCompleteWithin(3.Seconds());
186+
complete.Should().Be(0);
177187
}, Materializer);
178188
}
179189
}

‎src/core/Akka.Streams.Tests/Dsl/FlowOrElseSpec.cs

+9-4
Original file line numberDiff line numberDiff line change
@@ -9,10 +9,13 @@
99
using System;
1010
using Akka.Streams.Dsl;
1111
using Akka.Streams.TestKit;
12+
using Akka.TestKit.Extensions;
1213
using Akka.TestKit;
1314
using FluentAssertions;
1415
using Xunit;
1516
using Xunit.Abstractions;
17+
using System.Threading.Tasks;
18+
using FluentAssertions.Extensions;
1619

1720
namespace Akka.Streams.Tests.Dsl
1821
{
@@ -28,25 +31,27 @@ public FlowOrElseSpec(ITestOutputHelper helper) : base(helper)
2831
private ActorMaterializer Materializer { get; }
2932

3033
[Fact]
31-
public void An_OrElse_flow_should_pass_elements_from_the_first_input()
34+
public async Task An_OrElse_flow_should_pass_elements_from_the_first_input()
3235
{
3336
var source1 = Source.From(new[] {1, 2, 3});
3437
var source2 = Source.From(new[] {4, 5, 6});
3538

3639
var sink = Sink.Seq<int>();
3740

38-
source1.OrElse(source2).RunWith(sink, Materializer).AwaitResult().Should().BeEquivalentTo(new[] {1, 2, 3});
41+
var complete = await source1.OrElse(source2).RunWith(sink, Materializer).ShouldCompleteWithin(3.Seconds());
42+
complete.Should().BeEquivalentTo(new[] { 1, 2, 3 });
3943
}
4044

4145
[Fact]
42-
public void An_OrElse_flow_should_pass_elements_from_the_second_input_if_the_first_completes_with_no_elements_emitted()
46+
public async Task An_OrElse_flow_should_pass_elements_from_the_second_input_if_the_first_completes_with_no_elements_emitted()
4347
{
4448
var source1 = Source.Empty<int>();
4549
var source2 = Source.From(new[] { 4, 5, 6 });
4650

4751
var sink = Sink.Seq<int>();
4852

49-
source1.OrElse(source2).RunWith(sink, Materializer).AwaitResult().Should().BeEquivalentTo(new[] { 4, 5, 6 });
53+
var complete = await source1.OrElse(source2).RunWith(sink, Materializer).ShouldCompleteWithin(3.Seconds());
54+
complete.Should().BeEquivalentTo(new[] { 4, 5, 6 });
5055
}
5156

5257
[Fact]

‎src/core/Akka.Streams.Tests/Dsl/FlowScanAsyncSpec.cs

+5-2
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,12 @@
1616
using Akka.Streams.TestKit;
1717
using Akka.TestKit;
1818
using Akka.TestKit.Xunit2.Attributes;
19+
using Akka.TestKit.Extensions;
1920
using FluentAssertions;
2021
using Xunit;
2122
using Xunit.Abstractions;
2223
using Decider = Akka.Streams.Supervision.Decider;
24+
using FluentAssertions.Extensions;
2325

2426
namespace Akka.Streams.Tests.Dsl
2527
{
@@ -59,14 +61,15 @@ public void A_ScanAsync_must_work_with_a_single_source()
5961
}
6062

6163
[Fact]
62-
public void A_ScanAsync_must_work_with_a_large_source()
64+
public async Task A_ScanAsync_must_work_with_a_large_source()
6365
{
6466
var elements = Enumerable.Range(1, 100000).Select(i => (long)i).ToList();
6567
var expectedSum = elements.Sum();
6668
var eventualActual = Source.From(elements)
6769
.ScanAsync(0L, (l, l1) => Task.FromResult(l + l1))
6870
.RunWith(Sink.Last<long>(), Materializer);
69-
eventualActual.AwaitResult().ShouldBe(expectedSum);
71+
var complete = await eventualActual.ShouldCompleteWithin(3.Seconds());
72+
complete.ShouldBe(expectedSum);
7073
}
7174

7275
[LocalFact(SkipLocal = "Racy on Azure DevOps")]

0 commit comments

Comments
 (0)