Skip to content

[66-74] SubstreamSubscriptionTimeoutSpec #6614

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Mar 30, 2023
Merged

[66-74] SubstreamSubscriptionTimeoutSpec #6614

Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fae058b
[66-74] `SubstreamSubscriptionTimeoutSpec`
eaba Mar 26, 2023
d3845da
Merge branch 'dev' into SubstreamSubscriptionTimeoutSpec
eaba Mar 28, 2023
b4b476d
Changes to `async`
eaba Mar 29, 2023
7fcee24
Merge branch 'dev' into SubstreamSubscriptionTimeoutSpec
eaba Mar 29, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 10 additions & 9 deletions src/core/Akka.Streams.Tests/Dsl/SubstreamSubscriptionTimeoutSpec.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

using System;
using System.Threading;
using System.Threading.Tasks;
using Akka.Streams.Dsl;
using Akka.Streams.Implementation;
using Akka.Streams.TestKit;
Expand Down Expand Up @@ -40,15 +41,14 @@ public SubstreamSubscriptionTimeoutSpec(ITestOutputHelper helper) : base(Config,
}

[LocalFact(SkipLocal = "Racy on Azure DevOps")]
public void GroupBy_and_SplitWhen_must_timeout_and_cancel_substream_publisher_when_no_one_subscribes_to_them_after_some_time()
public async Task GroupBy_and_SplitWhen_must_timeout_and_cancel_substream_publisher_when_no_one_subscribes_to_them_after_some_time()
{
this.AssertAllStagesStopped(() =>
{
await this.AssertAllStagesStoppedAsync(() => {
var subscriber = this.CreateManualSubscriberProbe<(int, Source<int, NotUsed>)>();
var publisherProbe = this.CreatePublisherProbe<int>();
Source.FromPublisher(publisherProbe)
.GroupBy(3, x => x%3)
.Lift(x => x%3)
.GroupBy(3, x => x % 3)
.Lift(x => x % 3)
.RunWith(Sink.FromSubscriber(subscriber), Materializer);

var downstreamSubscription = subscriber.ExpectSubscription();
Expand All @@ -57,7 +57,7 @@ public void GroupBy_and_SplitWhen_must_timeout_and_cancel_substream_publisher_wh
publisherProbe.SendNext(1);
publisherProbe.SendNext(2);
publisherProbe.SendNext(3);

/*
* Why this spec is skipped: in the event that subscriber.ExpectSubscription() or (subscriber.ExpectNext()
* + s1SubscriberProbe.ExpectSubscription()) exceeds 300ms, the next call to subscriber.ExpectNext will
Expand Down Expand Up @@ -91,14 +91,14 @@ public void GroupBy_and_SplitWhen_must_timeout_and_cancel_substream_publisher_wh
action.Should().Throw<SubscriptionTimeoutException>();
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

 // sleep long enough for it to be cleaned up
                Thread.Sleep(1500);

Change to Task.Delay


publisherProbe.SendComplete();
return Task.CompletedTask;
}, Materializer);
}

[Fact]
public void GroupBy_and_SplitWhen_must_timeout_and_stop_groupBy_parent_actor_if_none_of_the_substreams_are_actually_consumed()
public async Task GroupBy_and_SplitWhen_must_timeout_and_stop_groupBy_parent_actor_if_none_of_the_substreams_are_actually_consumed()
{
this.AssertAllStagesStopped(() =>
{
await this.AssertAllStagesStoppedAsync(() => {
var subscriber = this.CreateManualSubscriberProbe<(int, Source<int, NotUsed>)>();
var publisherProbe = this.CreatePublisherProbe<int>();
Source.FromPublisher(publisherProbe)
Expand All @@ -116,6 +116,7 @@ public void GroupBy_and_SplitWhen_must_timeout_and_stop_groupBy_parent_actor_if_

subscriber.ExpectNext();
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Use async methods

subscriber.ExpectNext();
return Task.CompletedTask;
}, Materializer);
}

Expand Down