1
1
//-----------------------------------------------------------------------
2
2
// <copyright file="FlowInitialDelaySpec.cs" company="Akka.NET Project">
3
- // Copyright (C) 2009-2023 Lightbend Inc. <http://www.lightbend.com>
4
- // Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
3
+ // Copyright (C) 2009-2022 Lightbend Inc. <http://www.lightbend.com>
4
+ // Copyright (C) 2013-2022 .NET Foundation <https://github.com/akkadotnet/akka.net>
5
5
// </copyright>
6
6
//-----------------------------------------------------------------------
7
7
8
8
using System ;
9
9
using System . Collections . Generic ;
10
10
using System . Linq ;
11
+ using System . Threading . Tasks ;
11
12
using Akka . Streams . Dsl ;
12
13
using Akka . Streams . TestKit ;
13
14
using Akka . TestKit ;
@@ -30,48 +31,47 @@ public FlowInitialDelaySpec(ITestOutputHelper helper) : base(helper)
30
31
}
31
32
32
33
[ Fact ]
33
- public void Flow_InitialDelay_must_work_with_zero_delay ( )
34
+ public async Task Flow_InitialDelay_must_work_with_zero_delay ( )
34
35
{
35
- this . AssertAllStagesStopped ( ( ) =>
36
- {
36
+ await this . AssertAllStagesStoppedAsync ( ( ) => {
37
37
var task = Source . From ( Enumerable . Range ( 1 , 10 ) )
38
- . InitialDelay ( TimeSpan . Zero )
39
- . Grouped ( 100 )
40
- . RunWith ( Sink . First < IEnumerable < int > > ( ) , Materializer ) ;
38
+ . InitialDelay ( TimeSpan . Zero )
39
+ . Grouped ( 100 )
40
+ . RunWith ( Sink . First < IEnumerable < int > > ( ) , Materializer ) ;
41
41
task . Wait ( TimeSpan . FromSeconds ( 1 ) ) . Should ( ) . BeTrue ( ) ;
42
- task . Result . Should ( ) . BeEquivalentTo ( Enumerable . Range ( 1 , 10 ) ) ;
42
+ task . Result . Should ( ) . BeEquivalentTo ( Enumerable . Range ( 1 , 10 ) ) ;
43
+ return Task . CompletedTask ;
43
44
} , Materializer ) ;
44
45
}
45
46
46
47
[ Fact ]
47
- public void Flow_InitialDelay_must_delay_elements_by_the_specified_time_but_not_more ( )
48
+ public async Task Flow_InitialDelay_must_delay_elements_by_the_specified_time_but_not_more ( )
48
49
{
49
- this . AssertAllStagesStopped ( ( ) =>
50
- {
50
+ await this . AssertAllStagesStoppedAsync ( ( ) => {
51
51
var task = Source . From ( Enumerable . Range ( 1 , 10 ) )
52
- . InitialDelay ( TimeSpan . FromSeconds ( 2 ) )
53
- . InitialTimeout ( TimeSpan . FromSeconds ( 1 ) )
54
- . RunWith ( Sink . Ignore < int > ( ) , Materializer ) ;
52
+ . InitialDelay ( TimeSpan . FromSeconds ( 2 ) )
53
+ . InitialTimeout ( TimeSpan . FromSeconds ( 1 ) )
54
+ . RunWith ( Sink . Ignore < int > ( ) , Materializer ) ;
55
55
task . Invoking ( t => t . Wait ( TimeSpan . FromSeconds ( 2 ) ) ) . Should ( ) . Throw < TimeoutException > ( ) ;
56
+ return Task . CompletedTask ;
56
57
} , Materializer ) ;
57
58
}
58
59
59
60
[ Fact ]
60
- public void Flow_InitialDelay_must_properly_ignore_timer_while_backpressured ( )
61
+ public async Task Flow_InitialDelay_must_properly_ignore_timer_while_backpressured ( )
61
62
{
62
- this . AssertAllStagesStopped ( ( ) =>
63
- {
63
+ await this . AssertAllStagesStoppedAsync ( async ( ) => {
64
64
var probe = this . CreateSubscriberProbe < int > ( ) ;
65
65
Source . From ( Enumerable . Range ( 1 , 10 ) )
66
66
. InitialDelay ( TimeSpan . FromSeconds ( 0.5 ) )
67
67
. RunWith ( Sink . FromSubscriber ( probe ) , Materializer ) ;
68
68
69
- probe . EnsureSubscription ( ) ;
70
- probe . ExpectNoMsg ( TimeSpan . FromSeconds ( 1.5 ) ) ;
71
- probe . Request ( 20 ) ;
72
- probe . ExpectNextN ( Enumerable . Range ( 1 , 10 ) ) ;
69
+ await probe . EnsureSubscriptionAsync ( ) ;
70
+ await probe . ExpectNoMsgAsync ( TimeSpan . FromSeconds ( 1.5 ) ) ;
71
+ await probe . RequestAsync ( 20 ) ;
72
+ await probe . ExpectNextNAsync ( Enumerable . Range ( 1 , 10 ) ) ;
73
73
74
- probe . ExpectComplete ( ) ;
74
+ await probe . ExpectCompleteAsync ( ) ;
75
75
} , Materializer ) ;
76
76
}
77
77
}
0 commit comments