-
Notifications
You must be signed in to change notification settings - Fork 11
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
added initial Akka.Persistence.Hosting implementation for Azure (#223)
* added initial Akka.Persistence.Hosting implementation for Azure close #222 * fixed build issues * added sanity check for Akka.Persistence.Azure.Hosting * close #224 fixed malformed log message * fixed Akka.Hosting test setup
- Loading branch information
1 parent
3111200
commit f3344af
Showing
7 changed files
with
264 additions
and
6 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
17 changes: 17 additions & 0 deletions
17
src/Akka.Persistence.Azure.Hosting/Akka.Persistence.Azure.Hosting.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,17 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
| <Import Project="..\common.props" /> | ||
| <PropertyGroup> | ||
| <TargetFramework>$(NetStandardLibVersion)</TargetFramework> | ||
| <Description>Akka.Hosting support for Akka.Persistence.Azure.</Description> | ||
| <LangVersion>8.0</LangVersion> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\Akka.Persistence.Azure\Akka.Persistence.Azure.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| <ItemGroup> | ||
| <PackageReference Include="Akka.Persistence.Hosting" Version="0.4.0" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
102 changes: 102 additions & 0 deletions
102
src/Akka.Persistence.Azure.Hosting/AzurePersistenceExtensions.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,102 @@ | ||
| using System; | ||
| using Akka.Actor; | ||
| using Akka.Configuration; | ||
| using Akka.Hosting; | ||
| using Akka.Persistence.Azure.Query; | ||
| using Akka.Persistence.Hosting; | ||
|
|
||
| namespace Akka.Persistence.Azure.Hosting | ||
| { | ||
| /// <summary> | ||
| /// Extension methods for Akka.Hosting and Akka.Azure.Persistence | ||
| /// </summary> | ||
| public static class AzurePersistenceExtensions | ||
| { | ||
| public const string DefaultTableName = "AkkaPersistenceDefaultTable"; | ||
| public const string DefaultBlobContainerName = "akka-persistence-default-container"; | ||
|
|
||
| private static string ToHocon(bool b) | ||
| { | ||
| return b ? "on" : "off"; | ||
| } | ||
|
|
||
|
|
||
| public static AkkaConfigurationBuilder WithAzureTableJournal(this AkkaConfigurationBuilder builder, | ||
| string connectionString, bool autoInitialize = true, string tableName = DefaultTableName, Action<AkkaPersistenceJournalBuilder> configurator = null) | ||
| { | ||
| Config journalConfiguration = @$" | ||
| akka.persistence {{ | ||
| journal {{ | ||
| plugin = ""akka.persistence.journal.azure-table"" | ||
| azure-table {{ | ||
| class = ""Akka.Persistence.Azure.Journal.AzureTableStorageJournal, Akka.Persistence.Azure"" | ||
| connection-string = ""{connectionString}"" | ||
| # the name of the Windows Azure Table used to persist journal events | ||
| table-name = ""{tableName}"" | ||
| auto-initialize = {ToHocon(autoInitialize)} | ||
| }} | ||
| }} | ||
| }}"; | ||
|
|
||
| var finalConfig = journalConfiguration; | ||
| builder.AddHocon(finalConfig, HoconAddMode.Prepend); | ||
|
|
||
| // PUSH DEFAULT CONFIG TO END | ||
| builder.AddHocon(AzurePersistence.DefaultConfig, HoconAddMode.Append); | ||
|
|
||
| if (configurator != null) // configure event adapters | ||
| { | ||
| builder.WithJournal("azure-table", configurator); | ||
| } | ||
|
|
||
| return builder; | ||
| } | ||
|
|
||
| public static AkkaConfigurationBuilder WithAzureBlobsSnapshotStore(this AkkaConfigurationBuilder builder, | ||
| string connectionString, bool autoInitialize = true, string containerName = DefaultBlobContainerName) | ||
| { | ||
| Config journalConfiguration = @$" | ||
| akka.persistence {{ | ||
| snapshot-store {{ | ||
| plugin = ""akka.persistence.snapshot-store.azure-blob-store"" | ||
| azure-blob-store {{ | ||
| class = ""Akka.Persistence.Azure.Snapshot.AzureBlobSnapshotStore, Akka.Persistence.Azure"" | ||
| connection-string = ""{connectionString}"" | ||
| # the name of the Windows Azure Table used to persist journal events | ||
| container-name = ""{containerName}"" | ||
| auto-initialize = {ToHocon(autoInitialize)} | ||
| }} | ||
| }} | ||
| }}"; | ||
|
|
||
| var finalConfig = journalConfiguration; | ||
| builder.AddHocon(finalConfig, HoconAddMode.Prepend); | ||
|
|
||
| // PUSH DEFAULT CONFIG TO END | ||
| builder.AddHocon(AzurePersistence.DefaultConfig, HoconAddMode.Append); | ||
|
|
||
| return builder; | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Adds both AzureTableStorage journal and AzureBlobStorage snapshot-store as the default Akka.Persistence | ||
| /// implementations for a given <see cref="ActorSystem"/>. | ||
| /// </summary> | ||
| /// <param name="builder"></param> | ||
| /// <param name="connectionString"></param> | ||
| /// <param name="autoInitialize"></param> | ||
| /// <param name="containerName"></param> | ||
| /// <param name="tableName"></param> | ||
| /// <param name="configurator"></param> | ||
| /// <returns></returns> | ||
| public static AkkaConfigurationBuilder WithAzurePersistence(this AkkaConfigurationBuilder builder, | ||
| string connectionString, bool autoInitialize = true, string containerName = DefaultBlobContainerName, | ||
| string tableName = DefaultTableName, Action<AkkaPersistenceJournalBuilder> configurator = null) | ||
| { | ||
| builder.WithAzureTableJournal(connectionString, autoInitialize, tableName, configurator); | ||
| builder.WithAzureBlobsSnapshotStore(connectionString, autoInitialize, containerName); | ||
|
|
||
| return builder; | ||
| } | ||
| } | ||
| } |
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
127 changes: 127 additions & 0 deletions
127
src/Akka.Persistence.Azure.Tests/Hosting/AzurePersistenceHostingSanityCheck.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,127 @@ | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Threading.Tasks; | ||
| using Akka.Actor; | ||
| using Akka.Event; | ||
| using Akka.Hosting; | ||
| using Akka.Persistence.Azure.Hosting; | ||
| using Akka.TestKit.Xunit2.Internals; | ||
| using FluentAssertions; | ||
| using Microsoft.Extensions.DependencyInjection; | ||
| using Microsoft.Extensions.Hosting; | ||
| using Xunit; | ||
| using Xunit.Abstractions; | ||
|
|
||
| namespace Akka.Persistence.Azure.Tests.Hosting | ||
| { | ||
| public class AzurePersistenceHostingSanityCheck | ||
| { | ||
| public static async Task<IHost> StartHost(Action<AkkaConfigurationBuilder> testSetup) | ||
| { | ||
| var conn = Environment.GetEnvironmentVariable("AZURE_CONNECTION_STR"); | ||
| var host = new HostBuilder() | ||
| .ConfigureServices(collection => | ||
| { | ||
| collection.AddAkka("MyActorSys", builder => | ||
| { | ||
| builder.WithAzurePersistence(conn); | ||
| testSetup(builder); | ||
| }); | ||
| }).Build(); | ||
|
|
||
| await host.StartAsync(); | ||
| return host; | ||
| } | ||
|
|
||
| public sealed class MyPersistenceActor : ReceivePersistentActor | ||
| { | ||
| private List<int> _values = new List<int>(); | ||
|
|
||
| public MyPersistenceActor(string persistenceId) | ||
| { | ||
| PersistenceId = persistenceId; | ||
|
|
||
| Recover<SnapshotOffer>(offer => | ||
| { | ||
| if (offer.Snapshot is IEnumerable<int> ints) | ||
| { | ||
| _values = new List<int>(ints); | ||
| } | ||
| }); | ||
|
|
||
| Recover<int>(i => { _values.Add(i); }); | ||
|
|
||
| Command<int>(i => | ||
| { | ||
| Persist(i, i1 => | ||
| { | ||
| _values.Add(i); | ||
| if (LastSequenceNr % 2 == 0) | ||
| { | ||
| SaveSnapshot(_values); | ||
| } | ||
| Sender.Tell("ACK"); | ||
| }); | ||
| }); | ||
|
|
||
| Command<string>(str => str.Equals("getall"), s => { Sender.Tell(_values.ToArray()); }); | ||
|
|
||
| Command<SaveSnapshotSuccess>(s => { }); | ||
| } | ||
|
|
||
| public override string PersistenceId { get; } | ||
| } | ||
|
|
||
| private readonly ITestOutputHelper _output; | ||
|
|
||
| public AzurePersistenceHostingSanityCheck(ITestOutputHelper output) | ||
| { | ||
| _output = output; | ||
| } | ||
|
|
||
| [Fact] | ||
| public async Task ShouldLaunchAzurePersistence() | ||
| { | ||
| // arrange | ||
| using var host = await StartHost(builder => { | ||
| builder.StartActors((system, registry) => | ||
| { | ||
| var myActor = system.ActorOf(Props.Create(() => new MyPersistenceActor("ac1")), "actor1"); | ||
| registry.Register<MyPersistenceActor>(myActor); | ||
| }) | ||
| .WithActors((system, registry) => | ||
| { | ||
| var extSystem = (ExtendedActorSystem)system; | ||
| var logger = extSystem.SystemActorOf(Props.Create(() => new TestOutputLogger(_output)), "log-test"); | ||
| logger.Tell(new InitializeLogger(system.EventStream)); | ||
| }); | ||
| }); | ||
|
|
||
| var actorSystem = host.Services.GetRequiredService<ActorSystem>(); | ||
| var actorRegistry = host.Services.GetRequiredService<ActorRegistry>(); | ||
| var myPersistentActor = actorRegistry.Get<MyPersistenceActor>(); | ||
|
|
||
| // act | ||
| var resp1 = await myPersistentActor.Ask<string>(1, TimeSpan.FromSeconds(3)); | ||
| var resp2 = await myPersistentActor.Ask<string>(2, TimeSpan.FromSeconds(3)); | ||
| var snapshot = await myPersistentActor.Ask<int[]>("getall", TimeSpan.FromSeconds(3)); | ||
|
|
||
| // assert | ||
| snapshot.Should().BeEquivalentTo(new[] {1, 2}); | ||
|
|
||
| // kill + recreate actor with same PersistentId | ||
| await myPersistentActor.GracefulStop(TimeSpan.FromSeconds(3)); | ||
| var myPersistentActor2 = actorSystem.ActorOf(Props.Create(() => new MyPersistenceActor("ac1")), "actor1a"); | ||
|
|
||
| var snapshot2 = await myPersistentActor2.Ask<int[]>("getall", TimeSpan.FromSeconds(3)); | ||
| snapshot2.Should().BeEquivalentTo(new[] {1, 2}); | ||
|
|
||
| // validate configs | ||
| var config = actorSystem.Settings.Config; | ||
| config.GetString("akka.persistence.journal.plugin").Should().Be("akka.persistence.journal.azure-table"); | ||
| config.GetString("akka.persistence.snapshot-store.plugin").Should().Be("akka.persistence.snapshot-store.azure-blob-store"); | ||
| } | ||
| } | ||
| } |
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