-
Notifications
You must be signed in to change notification settings - Fork 1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Akka.Streams:
ReuseLatest stage to repeatedly emit the most recent …
…value until a newer one is pushed (#6262) * code cleanup in Akka.Streams `Attributes` * added `RepeatPrevious{T}` stage * WIP - debugging `RepeatPreviousSpecs` * fixed tests and added documentation * fixed documentation * API approvals * fixed markdown linting * removed `SwapPrevious<T>` delegate. * renamed stage from `RepeatPrevious` to `ReuseLatest` * remove BDN results
- Loading branch information
1 parent
d914eb3
commit 76c9364
Showing
9 changed files
with
316 additions
and
8 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| //----------------------------------------------------------------------- | ||
| // <copyright file="RepeatPreviousSpec.cs" company="Akka.NET Project"> | ||
| // Copyright (C) 2013-2022 .NET Foundation <https://github.com/akkadotnet/akka.net> | ||
| // </copyright> | ||
| //----------------------------------------------------------------------- | ||
|
|
||
| using System; | ||
| using System.Linq; | ||
| using System.Threading.Tasks; | ||
| using Akka.Configuration; | ||
| using Akka.Streams.Dsl; | ||
| using Akka.TestKit; | ||
| using FluentAssertions; | ||
| using Xunit; | ||
| using Xunit.Abstractions; | ||
|
|
||
| namespace Akka.Streams.Tests.Dsl | ||
| { | ||
| public class ReuseLatestSpec : AkkaSpec | ||
| { | ||
| private ActorMaterializer Materializer { get; } | ||
|
|
||
| public ReuseLatestSpec(ITestOutputHelper testOutputHelper) : base(Config.Empty, output: testOutputHelper) | ||
| { | ||
| var settings = ActorMaterializerSettings.Create(Sys); | ||
| Materializer = ActorMaterializer.Create(Sys, settings); | ||
| } | ||
|
|
||
| [Fact] | ||
| public async Task RepeatPrevious_should_immediately_terminate_with_Empty_source() | ||
| { | ||
| var source = Source.Empty<int>(); | ||
| var result = await source.RepeatPrevious().RunWith(Sink.Seq<int>(), Materializer); | ||
| result.Should().BeEmpty(); | ||
| } | ||
|
|
||
| [Fact] | ||
| public async Task RepeatPrevious_should_complete_when_upstream_completes() | ||
| { | ||
| var source = Source.Single(1).RepeatPrevious(); | ||
| var result = await source.RunWith(Sink.Seq<int>(), Materializer); | ||
|
|
||
| // as a side-effect of RepeatPrevious' buffering process, there's going to be an extra element in the result | ||
| result.Should().BeEquivalentTo(1, 1); | ||
| } | ||
|
|
||
| [Fact] | ||
| public async Task RepeatPrevious_should_fail_when_upstream_fails() | ||
| { | ||
| Func<Task> Exec() => async () => | ||
| { | ||
| var source = Source.From(Enumerable.Range(0,9)).Where(i => | ||
| { | ||
| if (i % 5 == 0) | ||
| { | ||
| throw new ApplicationException("failed"); | ||
| } | ||
| return true; | ||
| }).RepeatPrevious(); | ||
| var result = await source.RunWith(Sink.Seq<int>(), Materializer); | ||
| }; | ||
|
|
||
| await Exec().Should().ThrowAsync<ApplicationException>(); | ||
| } | ||
|
|
||
| [Fact] | ||
| public async Task RepeatPrevious_should_repeat_when_no_newValues_available() | ||
| { | ||
| // <RepeatPrevious> | ||
| var (queue, source) = Source.Queue<int>(10, OverflowStrategy.Backpressure).PreMaterialize(Materializer); | ||
|
|
||
| // populate 1 into queue | ||
| await queue.OfferAsync(1); | ||
|
|
||
| // take 4 items from the queue | ||
| var result = await source.RepeatPrevious().Take(4).RunWith(Sink.Seq<int>(), Materializer); | ||
|
|
||
| // the most recent queue item will be repeated 3 times, plus the original element | ||
| result.Should().BeEquivalentTo(1,1,1,1); | ||
| // </RepeatPrevious> | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,96 @@ | ||
| //----------------------------------------------------------------------- | ||
| // <copyright file="RepeatPrevious.cs" company="Akka.NET Project"> | ||
| // Copyright (C) 2013-2022 .NET Foundation <https://github.com/akkadotnet/akka.net> | ||
| // </copyright> | ||
| //----------------------------------------------------------------------- | ||
|
|
||
| using System; | ||
| using Akka.Streams.Stage; | ||
| using Akka.Util; | ||
|
|
||
| namespace Akka.Streams.Dsl | ||
| { | ||
| /// <summary> | ||
| /// Reuses the latest element from upstream until it's replaced by a new value. | ||
| /// | ||
| /// This is designed to allow fan-in stages where output from one of the sources is intermittent / infrequent | ||
| /// and users just want the previous value to be reused. | ||
| /// </summary> | ||
| /// <typeparam name="T">The output type.</typeparam> | ||
| public sealed class ReuseLatest<T> : GraphStage<FlowShape<T, T>> | ||
| { | ||
| private readonly Inlet<T> _in = new Inlet<T>("RepeatPrevious.in"); | ||
| private readonly Outlet<T> _out = new Outlet<T>("RepeatPrevious.out"); | ||
|
|
||
| public override FlowShape<T, T> Shape => new FlowShape<T, T>(_in, _out); | ||
| private readonly Action<T,T> _onItemChanged; | ||
|
|
||
| /// <summary> | ||
| /// Do nothing by default | ||
| /// </summary> | ||
| private static readonly Action<T,T> DefaultSwap = (oldValue, newValue) => { }; | ||
|
|
||
| public ReuseLatest() : this(DefaultSwap) | ||
| { | ||
| } | ||
|
|
||
| public ReuseLatest(Action<T, T> onItemChanged) | ||
| { | ||
| _onItemChanged = onItemChanged; | ||
| } | ||
|
|
||
| protected override GraphStageLogic CreateLogic(Attributes inheritedAttributes) => | ||
| new Logic(this, _onItemChanged); | ||
|
|
||
| private sealed class Logic : InAndOutGraphStageLogic | ||
| { | ||
| private readonly ReuseLatest<T> _stage; | ||
| private Option<T> _last; | ||
| private readonly Action<T,T> _onItemChanged; | ||
|
|
||
| public Logic(ReuseLatest<T> stage, Action<T,T> onItemChanged) : base(stage.Shape) | ||
| { | ||
| _stage = stage; | ||
| _onItemChanged = onItemChanged; | ||
|
|
||
| SetHandler(_stage._in, this); | ||
| SetHandler(_stage._out, this); | ||
| } | ||
|
|
||
| public override void OnPush() | ||
| { | ||
| var next = Grab(_stage._in); | ||
| if (_last.HasValue) | ||
| _onItemChanged(_last.Value, next); | ||
| _last = next; | ||
|
|
||
| if (IsAvailable(_stage._out)) | ||
| { | ||
| Push(_stage._out, _last.Value); | ||
| } | ||
| } | ||
|
|
||
| public override void OnPull() | ||
| { | ||
| if (_last.HasValue) | ||
| { | ||
| if (!HasBeenPulled(_stage._in)) | ||
| { | ||
| Pull(_stage._in); | ||
| } | ||
|
|
||
| Push(_stage._out, _last.Value); | ||
| } | ||
| else | ||
| { | ||
| Pull(_stage._in); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public override string ToString() | ||
| { | ||
| return "RepeatPrevious"; | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.