-
Notifications
You must be signed in to change notification settings - Fork 9
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
0b316fa
commit d182f34
Showing
14 changed files
with
383 additions
and
3 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
58 changes: 58 additions & 0 deletions
58
src/Be.Vlaanderen.Basisregisters.ProjectionHandling.Syndication/FeedProjector.cs
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,58 @@ | ||
| namespace Be.Vlaanderen.Basisregisters.ProjectionHandling.Syndication | ||
| { | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Linq; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using Autofac.Features.OwnedInstances; | ||
| using Microsoft.Extensions.Logging; | ||
| using Runner; | ||
|
|
||
| public class FeedProjector<TContext> | ||
| where TContext: RunnerDbContext<TContext> | ||
| { | ||
| private readonly Func<Owned<TContext>> _contextFactory; | ||
| private readonly List<IFeedProjectionRunner<TContext>> _feedProjectionRunners; | ||
| private readonly ILogger _logger; | ||
|
|
||
| public FeedProjector( | ||
| Func<Owned<TContext>> contextFactory, | ||
| ILoggerFactory loggerFactory, | ||
| IEnumerable<IFeedProjectionRunner<TContext>> configuredFeedProjectionRunners) | ||
| { | ||
| _contextFactory = contextFactory; | ||
| _feedProjectionRunners = configuredFeedProjectionRunners?.ToList() ?? new List<IFeedProjectionRunner<TContext>>(); | ||
| _logger = loggerFactory?.CreateLogger<FeedProjector<TContext>>() ?? throw new ArgumentNullException(nameof(loggerFactory)); | ||
| } | ||
|
|
||
| public FeedProjector<TContext> Register(params IFeedProjectionRunner<TContext>[] feedProjectionRunners) | ||
| => Register((IEnumerable<IFeedProjectionRunner<TContext>>)feedProjectionRunners); | ||
|
|
||
| public FeedProjector<TContext> Register(IEnumerable<IFeedProjectionRunner<TContext>> feedProjectionRunners) | ||
| { | ||
| if (feedProjectionRunners != null) | ||
| _feedProjectionRunners.AddRange(feedProjectionRunners); | ||
|
|
||
| return this; | ||
| } | ||
|
|
||
| public async Task Start(CancellationToken cancellationToken) | ||
| { | ||
| var catchupRunners = _feedProjectionRunners.Select(runner => CreateCatchUpTask(runner, cancellationToken)); | ||
| await Task.WhenAll(catchupRunners); | ||
| } | ||
|
|
||
| private async Task CreateCatchUpTask(IFeedProjectionRunner<TContext> feed, CancellationToken cancellationToken) | ||
| { | ||
| try | ||
| { | ||
| await feed.CatchUpAsync(_contextFactory, cancellationToken); | ||
| } | ||
| catch (Exception exception) | ||
| { | ||
| _logger.LogError(exception, "FeedProjectionRunner failed"); | ||
| } | ||
| } | ||
| } | ||
| } |
9 changes: 9 additions & 0 deletions
9
...yndication.Tests/Be.Vlaanderen.Basisregisters.ProjectionHandling.Syndication.Tests.csproj
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,9 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
| <Import Project="..\..\packages\Be.Vlaanderen.Basisregisters.Build.Pipeline\Content\Be.Vlaanderen.Basisregisters.Build.Pipeline.Settings.Test.props" /> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\..\src\Be.Vlaanderen.Basisregisters.ProjectionHandling.Syndication\Be.Vlaanderen.Basisregisters.ProjectionHandling.Syndication.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| <Import Project="..\..\.paket\Paket.Restore.targets" /> | ||
| </Project> |
29 changes: 29 additions & 0 deletions
29
...nderen.Basisregisters.ProjectionHandling.Syndication.Tests/FeedProjector/FailingRunner.cs
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,29 @@ | ||
| namespace Be.Vlaanderen.Basisregisters.ProjectionHandling.Syndication.Tests.FeedProjector | ||
| { | ||
| using System; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using Autofac.Features.OwnedInstances; | ||
|
|
||
| public class FailingRunner : IFeedProjectionRunner<FeedProjectorTestContext> | ||
| { | ||
| private readonly Exception _exception; | ||
| private readonly TimeSpan _delay; | ||
|
|
||
| public FailingRunner(Exception exception) | ||
| : this(exception, TimeSpan.Zero) | ||
| { } | ||
|
|
||
| public FailingRunner(Exception exception, TimeSpan delay) | ||
| { | ||
| _exception = exception; | ||
| _delay = delay; | ||
| } | ||
|
|
||
| public async Task CatchUpAsync(Func<Owned<FeedProjectorTestContext>> context, CancellationToken cancellationToken) | ||
| { | ||
| await Task.Delay(_delay, cancellationToken); | ||
| throw _exception; | ||
| } | ||
| } | ||
| } |
9 changes: 9 additions & 0 deletions
9
...isregisters.ProjectionHandling.Syndication.Tests/FeedProjector/FeedProjectionException.cs
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,9 @@ | ||
| namespace Be.Vlaanderen.Basisregisters.ProjectionHandling.Syndication.Tests.FeedProjector | ||
| { | ||
| using System; | ||
|
|
||
| public class FeedProjectionException : Exception | ||
| { | ||
| public FeedProjectionException(string message) : base(message) { } | ||
| } | ||
| } |
9 changes: 9 additions & 0 deletions
9
...sregisters.ProjectionHandling.Syndication.Tests/FeedProjector/FeedProjectorTestContext.cs
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,9 @@ | ||
| namespace Be.Vlaanderen.Basisregisters.ProjectionHandling.Syndication.Tests.FeedProjector | ||
| { | ||
| using Runner; | ||
|
|
||
| public class FeedProjectorTestContext : RunnerDbContext<FeedProjectorTestContext> | ||
| { | ||
| public override string ProjectionStateSchema => "FakeSchema"; | ||
| } | ||
| } |
133 changes: 133 additions & 0 deletions
133
...n.Basisregisters.ProjectionHandling.Syndication.Tests/FeedProjector/FeedProjectorTests.cs
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,133 @@ | ||
| namespace Be.Vlaanderen.Basisregisters.ProjectionHandling.Syndication.Tests.FeedProjector | ||
| { | ||
| using System; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using Autofac.Features.OwnedInstances; | ||
| using FluentAssertions; | ||
| using Microsoft.Extensions.Logging; | ||
| using Moq; | ||
| using Xunit; | ||
|
|
||
| public class When_starting_the_projector_with_a_projection_that_throws_an_exception | ||
| { | ||
| private readonly FeedProjector<FeedProjectorTestContext> _sut; | ||
| private readonly LoggerFactoryMock _loggerFactoryMock; | ||
| private readonly FeedProjectionException _expectedException; | ||
|
|
||
| public When_starting_the_projector_with_a_projection_that_throws_an_exception() | ||
| { | ||
| _loggerFactoryMock = new LoggerFactoryMock(); | ||
| _expectedException = new FeedProjectionException("error1"); | ||
|
|
||
| _sut = new FeedProjector<FeedProjectorTestContext>( | ||
| () => new Mock<Owned<FeedProjectorTestContext>>().Object, | ||
| _loggerFactoryMock, | ||
| new []{ new FailingRunner(_expectedException) }); | ||
| } | ||
|
|
||
| private Func<Task> StartProjections => | ||
| () => _sut.Start(CancellationToken.None); | ||
|
|
||
| [Fact] | ||
| public void Then_the_projector_does_not_throw_an_exception() | ||
| { | ||
| StartProjections.Should().NotThrow(); | ||
| } | ||
|
|
||
| [Fact] | ||
| public async Task Then_the_projector_logs_the_thrown_exception() | ||
| { | ||
| await StartProjections(); | ||
|
|
||
| _loggerFactoryMock | ||
| .ResolveLogger<FeedProjector<FeedProjectorTestContext>>() | ||
| .Verify( | ||
| logger => logger.Log( | ||
| LogLevel.Error, | ||
| It.IsAny<EventId>(), | ||
| It.Is<It.IsAnyType>((v, t) => v.ToString() == "FeedProjectionRunner failed"), | ||
| _expectedException, | ||
| (Func<It.IsAnyType, Exception, string>)It.IsAny<object>()), | ||
| Times.Once); | ||
| } | ||
| } | ||
|
|
||
| public class When_starting_the_projector_with_a_multiple_failing_projections_and_an_infinte_projection | ||
| { | ||
| private readonly FeedProjector<FeedProjectorTestContext> _sut; | ||
| private readonly LoggerFactoryMock _loggerFactoryMock; | ||
| private readonly FeedProjectionException _firstExpectedException, _secondExpectedException; | ||
| private readonly CancellationTokenSource _cancellationTokenSource; | ||
|
|
||
| public When_starting_the_projector_with_a_multiple_failing_projections_and_an_infinte_projection() | ||
| { | ||
| _cancellationTokenSource = new CancellationTokenSource(); | ||
| _loggerFactoryMock = new LoggerFactoryMock(); | ||
| _firstExpectedException = new FeedProjectionException("error1"); | ||
| _secondExpectedException = new FeedProjectionException("error2"); | ||
|
|
||
| _sut = new FeedProjector<FeedProjectorTestContext>( | ||
| () => new Mock<Owned<FeedProjectorTestContext>>().Object, | ||
| _loggerFactoryMock, | ||
| new IFeedProjectionRunner<FeedProjectorTestContext>[] | ||
| { | ||
| new FailingRunner(_secondExpectedException, TimeSpan.FromMilliseconds(300)), | ||
| new InfiniteRunner(), | ||
| new FailingRunner(_firstExpectedException, TimeSpan.FromMilliseconds(50)), | ||
| }); | ||
| } | ||
|
|
||
| private Func<Task> StartProjections => | ||
| () => _sut.Start(_cancellationTokenSource.Token); | ||
|
|
||
| private void StopProjections() | ||
| => _cancellationTokenSource.Cancel(); | ||
|
|
||
| [Fact] | ||
| public async Task Then_the_projector_logs_the_first_thrown_exception_is_logged_while_the_projector_is_still_running() | ||
| { | ||
| //don't await starting the projections | ||
| StartProjections(); | ||
|
|
||
| // wait for exception to be thrown | ||
| await Task.Delay(55); | ||
|
|
||
| _loggerFactoryMock | ||
| .ResolveLogger<FeedProjector<FeedProjectorTestContext>>() | ||
| .Verify( | ||
| logger => logger.Log( | ||
| LogLevel.Error, | ||
| It.IsAny<EventId>(), | ||
| It.Is<It.IsAnyType>((v, t) => v.ToString() == "FeedProjectionRunner failed"), | ||
| _firstExpectedException, | ||
| (Func<It.IsAnyType, Exception, string>)It.IsAny<object>()), | ||
| Times.Once); | ||
|
|
||
| StopProjections(); | ||
| } | ||
|
|
||
| [Fact] | ||
| public async Task Then_the_projector_logs_the_second_thrown_exception_is_logged_while_the_projector_is_still_running() | ||
| { | ||
| //don't await starting the projections | ||
| StartProjections(); | ||
|
|
||
| // wait for exception to be thrown | ||
| await Task.Delay(305); | ||
|
|
||
| _loggerFactoryMock | ||
| .ResolveLogger<FeedProjector<FeedProjectorTestContext>>() | ||
| .Verify( | ||
| logger => logger.Log( | ||
| LogLevel.Error, | ||
| It.IsAny<EventId>(), | ||
| It.Is<It.IsAnyType>((v, t) => v.ToString() == "FeedProjectionRunner failed"), | ||
| _secondExpectedException, | ||
| (Func<It.IsAnyType, Exception, string>)It.IsAny<object>()), | ||
| Times.Once); | ||
|
|
||
| StopProjections(); | ||
| } | ||
| } | ||
| } |
18 changes: 18 additions & 0 deletions
18
...deren.Basisregisters.ProjectionHandling.Syndication.Tests/FeedProjector/InfiniteRunner.cs
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,18 @@ | ||
| namespace Be.Vlaanderen.Basisregisters.ProjectionHandling.Syndication.Tests.FeedProjector | ||
| { | ||
| using System; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using Autofac.Features.OwnedInstances; | ||
|
|
||
| public class InfiniteRunner : IFeedProjectionRunner<FeedProjectorTestContext> | ||
| { | ||
| public async Task CatchUpAsync(Func<Owned<FeedProjectorTestContext>> context, CancellationToken cancellationToken) | ||
| { | ||
| while (!cancellationToken.IsCancellationRequested) | ||
| { | ||
| await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken); | ||
| } | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.