Skip to content

Commit 0ae9228

Browse files
eabaeaba
authored
[52-74] HeadSinkSpec (#6599)
* [52-74] `HeadSinkSpec` * fix * Changes to `async/await`
1 parent 4e6817d commit 0ae9228

File tree

1 file changed

+43
-44
lines changed

1 file changed

+43
-44
lines changed

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

+43-44
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
//-----------------------------------------------------------------------
77

88
using System;
9+
using System.Threading.Tasks;
910
using Akka.Streams.Dsl;
1011
using Akka.Streams.TestKit;
1112
using FluentAssertions;
@@ -25,23 +26,22 @@ public HeadSinkSpec(ITestOutputHelper helper):base(helper)
2526
}
2627

2728
[Fact]
28-
public void A_FLow_with_a_Sink_Head_must_yield_the_first_value()
29+
public async Task A_FLow_with_a_Sink_Head_must_yield_the_first_value()
2930
{
30-
this.AssertAllStagesStopped(() =>
31-
{
31+
await this.AssertAllStagesStoppedAsync(async() => {
3232
var p = this.CreateManualPublisherProbe<int>();
33-
var task = Source.FromPublisher(p).Select(x=>x).RunWith(Sink.First<int>(), Materializer);
34-
var proc = p.ExpectSubscription();
35-
proc.ExpectRequest();
33+
var task = Source.FromPublisher(p).Select(x => x).RunWith(Sink.First<int>(), Materializer);
34+
var proc = await p.ExpectSubscriptionAsync();
35+
await proc.ExpectRequestAsync();
3636
proc.SendNext(42);
3737
task.Wait(100);
3838
task.Result.Should().Be(42);
39-
proc.ExpectCancellation();
39+
await proc.ExpectCancellationAsync();
4040
}, Materializer);
4141
}
4242

4343
[Fact]
44-
public void A_FLow_with_a_Sink_Head_must_yield_the_first_value_when_actively_constructing()
44+
public async Task A_FLow_with_a_Sink_Head_must_yield_the_first_value_when_actively_constructing()
4545
{
4646
var p = this.CreateManualPublisherProbe<int>();
4747
var f = Sink.First<int>();
@@ -51,79 +51,78 @@ public void A_FLow_with_a_Sink_Head_must_yield_the_first_value_when_actively_con
5151
var future = t.Item2;
5252

5353
p.Subscribe(subscriber);
54-
var proc = p.ExpectSubscription();
55-
proc.ExpectRequest();
54+
var proc = await p.ExpectSubscriptionAsync();
55+
await proc.ExpectRequestAsync();
5656
proc.SendNext(42);
5757
future.Wait(100);
5858
future.Result.Should().Be(42);
59-
proc.ExpectCancellation();
59+
await proc.ExpectCancellationAsync();
6060
}
6161

6262
[Fact]
63-
public void A_FLow_with_a_Sink_Head_must_yield_the_first_error()
63+
public async Task A_FLow_with_a_Sink_Head_must_yield_the_first_error()
6464
{
65-
this.AssertAllStagesStopped(() =>
66-
{
67-
Source.Failed<int>(new Exception("ex"))
68-
.Invoking(s => s.RunWith(Sink.First<int>(), Materializer).Wait(TimeSpan.FromSeconds(1)))
69-
.Should().Throw<AggregateException>()
70-
.WithInnerException<Exception>()
71-
.WithMessage("ex");
65+
await this.AssertAllStagesStoppedAsync(() => {
66+
Source.Failed<int>(new Exception("ex"))
67+
.Invoking(s => s.RunWith(Sink.First<int>(), Materializer).Wait(TimeSpan.FromSeconds(1)))
68+
.Should().Throw<AggregateException>()
69+
.WithInnerException<Exception>()
70+
.WithMessage("ex");
71+
return Task.CompletedTask;
7272
}, Materializer);
7373
}
7474

7575
[Fact]
76-
public void A_FLow_with_a_Sink_Head_must_yield_NoSuchElementException_for_empty_stream()
76+
public async Task A_FLow_with_a_Sink_Head_must_yield_NoSuchElementException_for_empty_stream()
7777
{
78-
this.AssertAllStagesStopped(() =>
79-
{
80-
Source.Empty<int>()
81-
.Invoking(s => s.RunWith(Sink.First<int>(), Materializer).Wait(TimeSpan.FromSeconds(1)))
82-
.Should().Throw<AggregateException>()
83-
.WithInnerException<NoSuchElementException>()
84-
.WithMessage("First of empty stream");
78+
await this.AssertAllStagesStoppedAsync(() => {
79+
Source.Empty<int>()
80+
.Invoking(s => s.RunWith(Sink.First<int>(), Materializer).Wait(TimeSpan.FromSeconds(1)))
81+
.Should().Throw<AggregateException>()
82+
.WithInnerException<NoSuchElementException>()
83+
.WithMessage("First of empty stream");
84+
return Task.CompletedTask;
8585
}, Materializer);
8686
}
8787

8888

8989

9090
[Fact]
91-
public void A_FLow_with_a_Sink_HeadOption_must_yield_the_first_value()
91+
public async Task A_FLow_with_a_Sink_HeadOption_must_yield_the_first_value()
9292
{
93-
this.AssertAllStagesStopped(() =>
94-
{
93+
await this.AssertAllStagesStoppedAsync(async() => {
9594
var p = this.CreateManualPublisherProbe<int>();
9695
var task = Source.FromPublisher(p).Select(x => x).RunWith(Sink.FirstOrDefault<int>(), Materializer);
97-
var proc = p.ExpectSubscription();
98-
proc.ExpectRequest();
96+
var proc = await p.ExpectSubscriptionAsync();
97+
await proc.ExpectRequestAsync();
9998
proc.SendNext(42);
10099
task.Wait(100);
101100
task.Result.Should().Be(42);
102-
proc.ExpectCancellation();
101+
await proc.ExpectCancellationAsync();
103102
}, Materializer);
104103
}
105104

106105
[Fact]
107-
public void A_FLow_with_a_Sink_HeadOption_must_yield_the_first_error()
106+
public async Task A_FLow_with_a_Sink_HeadOption_must_yield_the_first_error()
108107
{
109-
this.AssertAllStagesStopped(() =>
110-
{
111-
Source.Failed<int>(new Exception("ex"))
112-
.Invoking(s => s.RunWith(Sink.FirstOrDefault<int>(), Materializer).Wait(TimeSpan.FromSeconds(1)))
113-
.Should().Throw<AggregateException>()
114-
.WithInnerException<Exception>()
115-
.WithMessage("ex");
108+
await this.AssertAllStagesStoppedAsync(() => {
109+
Source.Failed<int>(new Exception("ex"))
110+
.Invoking(s => s.RunWith(Sink.FirstOrDefault<int>(), Materializer).Wait(TimeSpan.FromSeconds(1)))
111+
.Should().Throw<AggregateException>()
112+
.WithInnerException<Exception>()
113+
.WithMessage("ex");
114+
return Task.CompletedTask;
116115
}, Materializer);
117116
}
118117

119118
[Fact]
120-
public void A_FLow_with_a_Sink_HeadOption_must_yield_default_for_empty_stream()
119+
public async Task A_FLow_with_a_Sink_HeadOption_must_yield_default_for_empty_stream()
121120
{
122-
this.AssertAllStagesStopped(() =>
123-
{
121+
await this.AssertAllStagesStoppedAsync(() => {
124122
var task = Source.Empty<int>().RunWith(Sink.FirstOrDefault<int>(), Materializer);
125123
task.Wait(TimeSpan.FromSeconds(1)).Should().BeTrue();
126124
task.Result.Should().Be(0);
125+
return Task.CompletedTask;
127126
}, Materializer);
128127
}
129128

0 commit comments

Comments
 (0)