Skip to content
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.

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

Fix missing default query configuration #317

Merged
merged 2 commits into from
Sep 14, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
9 changes: 9 additions & 0 deletions Akka.Persistence.Sql.sln
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,10 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Akka.Persistence.Sql.Hostin
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Akka.Persistence.Sql.Benchmarks", "src\Akka.Persistence.Sql.Benchmarks\Akka.Persistence.Sql.Benchmarks.csproj", "{1BE5B28E-0AB1-44BC-A5E8-5D352B42ACCF}"
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Examples", "Examples", "{42F8143A-03CA-41DA-B24D-33193ABB252C}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "HoconConfiguration", "src\Examples\HoconConfiguration\HoconConfiguration.csproj", "{043A8917-5931-40FC-A093-21BB0AB56875}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
Expand Down Expand Up @@ -130,6 +134,10 @@ Global
{1BE5B28E-0AB1-44BC-A5E8-5D352B42ACCF}.Debug|Any CPU.Build.0 = Debug|Any CPU
{1BE5B28E-0AB1-44BC-A5E8-5D352B42ACCF}.Release|Any CPU.ActiveCfg = Release|Any CPU
{1BE5B28E-0AB1-44BC-A5E8-5D352B42ACCF}.Release|Any CPU.Build.0 = Release|Any CPU
{043A8917-5931-40FC-A093-21BB0AB56875}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{043A8917-5931-40FC-A093-21BB0AB56875}.Debug|Any CPU.Build.0 = Debug|Any CPU
{043A8917-5931-40FC-A093-21BB0AB56875}.Release|Any CPU.ActiveCfg = Release|Any CPU
{043A8917-5931-40FC-A093-21BB0AB56875}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
Expand All @@ -144,6 +152,7 @@ Global
{74AFECF0-CC70-47C9-9393-2AB952B23BBD} = {C4AC49FF-9ECF-4D38-A201-1105BBF7E628}
{4ACEF3FD-F5B4-4C07-B85A-AD4D6AB68F4D} = {C59D2898-1994-4C40-B152-1798060051ED}
{1BE5B28E-0AB1-44BC-A5E8-5D352B42ACCF} = {3E392F73-EB03-4148-BC60-A1F734486190}
{043A8917-5931-40FC-A093-21BB0AB56875} = {42F8143A-03CA-41DA-B24D-33193ABB252C}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {B99E6BB8-642A-4A68-86DF-69567CBA700A}
Expand Down
3 changes: 3 additions & 0 deletions build.fsx
Original file line number Diff line number Diff line change
Expand Up @@ -89,8 +89,10 @@ Target "RunTests" (fun _ ->
| true -> !! "./src/**/*.Tests.csproj"
-- "./src/**/*.Benchmark.*.csproj"
-- "./src/**/*.Data.Compatibility.Tests.csproj" // All of the data docker images are Linux only
-- "./src/Examples/**/*.csproj" // skip example projects
| _ -> !! "./src/**/*.Tests.csproj"
-- "./src/**/*.Benchmark.*.csproj"
-- "./src/Examples/**/*.csproj" // skip example projects
++ "./src/**/*.DockerTests.csproj" // if you need to filter specs for Linux vs. Windows, do it here

let runSingleProject project =
Expand Down Expand Up @@ -144,6 +146,7 @@ Target "CreateNuget" (fun _ ->
-- "src/**/*Tests.csproj" // Don't publish unit tests
-- "src/**/*Tests*.csproj"
-- "src/**/*.Benchmark.*.csproj"
-- "./src/Examples/**/*.csproj" // skip example projects

let runSingleProject project =
DotNetCli.Pack
Expand Down
2 changes: 1 addition & 1 deletion src/Akka.Persistence.Sql/Query/JournalSequenceActor.cs
Original file line number Diff line number Diff line change
Expand Up @@ -205,7 +205,7 @@ protected override void PreStart()
{
if (t.IsFaulted)
{
_log.Info(t.Exception, "Failed to recover fast, using event-by-event recovery instead");
_log.Debug($"Failed to recover fast, using event-by-event recovery instead. Message: [{t.Exception?.Message ?? "nothing"}]");
}
else if (t.IsCompleted)
{
Expand Down
5 changes: 2 additions & 3 deletions src/Akka.Persistence.Sql/Query/SqlReadJournal.cs
Original file line number Diff line number Diff line change
Expand Up @@ -50,8 +50,7 @@ public class SqlReadJournal :

public SqlReadJournal(
ExtendedActorSystem system,
Configuration.Config config,
string configPath)
Configuration.Config config)
{
var writePluginId = config.GetString("write-plugin");

Expand All @@ -68,7 +67,7 @@ public SqlReadJournal(
_mat = Materializer.CreateSystemMaterializer(
context: system,
settings: ActorMaterializerSettings.Create(system),
namePrefix: $"l2db-query-mat{configPath}");
namePrefix: $"l2db-query-mat-{Guid.NewGuid():N}");
Copy link
Member

Choose a reason for hiding this comment

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

What was the fix here?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

That was not the fix, the SqlReadJournal is a bit different in that it creates a materializer, which none of our other read journal does.

It tries to create a single materializer by appending the "config path".

The problem is that "config path" were never being passed into the provider, the provider was created through reflection magic and none of the activator code passes in the "config path".

So I just remove that ambiguous code and replace it with a GUID


_readJournalDao = new ByteArrayReadJournalDao(
scheduler: system.Scheduler.Advanced,
Expand Down
16 changes: 2 additions & 14 deletions src/Akka.Persistence.Sql/Query/SqlReadJournalProvider.cs
Original file line number Diff line number Diff line change
Expand Up @@ -12,29 +12,17 @@ namespace Akka.Persistence.Sql.Query
public class SqlReadJournalProvider : IReadJournalProvider
{
private readonly Configuration.Config _config;
private readonly string _configPath;
private readonly ExtendedActorSystem _system;

public SqlReadJournalProvider(
ExtendedActorSystem system,
Configuration.Config config)
{
_system = system;
_config = config;
_configPath = "sql";
}

public SqlReadJournalProvider(
ExtendedActorSystem system,
Configuration.Config config,
string configPath)
{
_system = system;
_config = config;
_configPath = configPath;
_config = config.WithFallback(SqlPersistence.DefaultQueryConfiguration);
Copy link
Contributor Author

Choose a reason for hiding this comment

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

This is the actual fix

Copy link
Member

Choose a reason for hiding this comment

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

Got it, ok thank you.

}

public IReadJournal GetReadJournal()
=> new SqlReadJournal(_system, _config, _configPath);
=> new SqlReadJournal(_system, _config);
}
}
1 change: 1 addition & 0 deletions src/Directory.Packages.props
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
<PackageVersion Include="Akka.Persistence.Query" Version="$(AkkaVersion)" />
<PackageVersion Include="Akka.Serialization.Hyperion" Version="$(AkkaVersion)" />
<PackageVersion Include="Akka.Streams" Version="$(AkkaVersion)" />
<PackageVersion Include="Microsoft.Data.Sqlite" Version="7.0.11" />
</ItemGroup>
<!-- App dependencies -->
<ItemGroup>
Expand Down
25 changes: 25 additions & 0 deletions src/Examples/HoconConfiguration/HoconConfiguration.csproj
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
<Project Sdk="Microsoft.NET.Sdk">

<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net6.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<IsPackable>false</IsPackable>
</PropertyGroup>

<ItemGroup>
<ProjectReference Include="..\..\Akka.Persistence.Sql\Akka.Persistence.Sql.csproj" />
</ItemGroup>

<ItemGroup>
<PackageReference Include="Microsoft.Data.Sqlite" />
</ItemGroup>

<ItemGroup>
<None Update="db\test.db">
<CopyToOutputDirectory>Always</CopyToOutputDirectory>
</None>
</ItemGroup>

</Project>
58 changes: 58 additions & 0 deletions src/Examples/HoconConfiguration/Program.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
// See https://aka.ms/new-console-template for more information

using Akka.Actor;
using Akka.Persistence;
using Akka.Persistence.Query;
using Akka.Persistence.Sql.Query;
using Akka.Streams;
using Akka.Streams.Dsl;
using HoconConfiguration;

const string config =
"""
akka.persistence {
journal {
plugin = "akka.persistence.journal.sql"
sql {
class = "Akka.Persistence.Sql.Journal.SqlWriteJournal, Akka.Persistence.Sql"
connection-string = "DataSource=db/test.db;"
provider-name = "SQLite.MS"
}
}
query.journal.sql {
class = "Akka.Persistence.Sql.Query.SqlReadJournalProvider, Akka.Persistence.Sql"
connection-string = "DataSource=db/test.db;"
provider-name = "SQLite.MS"
}
snapshot-store {
plugin = "akka.persistence.snapshot-store.sql"
sql {
class = "Akka.Persistence.Sql.Snapshot.SqlSnapshotStore, Akka.Persistence.Sql"
connection-string = "DataSource=db/test.db;"
provider-name = "SQLite.MS"
}
}
}
""";

var sys = ActorSystem.Create("my-system", config);
var actor = sys.ActorOf(TestPersistenceActor.Props("test"));
var reader = sys.ReadJournalFor<SqlReadJournal>("akka.persistence.query.journal.sql");

foreach (var i in Enumerable.Range(0, 200))
{
var chr = (char)(65 + i % 26);
actor.Tell(new string(new []{chr}));
await Task.Delay(TimeSpan.FromSeconds(0.05));
}

_ = reader.CurrentAllEvents(Offset.NoOffset())
.Select(e =>
{
Console.WriteLine($"New event: {e.Event}");
return e;
}).RunWith(Sink.Ignore<EventEnvelope>(), sys.Materializer());

Console.ReadKey();

await sys.Terminate();
50 changes: 50 additions & 0 deletions src/Examples/HoconConfiguration/TestPersistenceActor.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
// -----------------------------------------------------------------------
// <copyright file="PersistenceActor.cs" company="Akka.NET Project">
// Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
// </copyright>
// -----------------------------------------------------------------------

using Akka.Actor;
using Akka.Event;
using Akka.Persistence;

namespace HoconConfiguration
{
public class TestPersistenceActor: ReceivePersistentActor
{
public static Props Props(string id) => Akka.Actor.Props.Create(() => new TestPersistenceActor(id));

public override string PersistenceId { get; }

private int _counter;
private string _state = string.Empty;

public TestPersistenceActor(string persistenceId)
{
PersistenceId = persistenceId;
var log = Context.GetLogger();

Recover<SnapshotOffer>(s => _state = (string)s.Snapshot);
Recover<string>(s => _state += s);

Command<string>(
msg =>
{
Persist(msg,
s =>
{
_state += s;
_counter++;
if(_counter % 25 == 0)
SaveSnapshot(_state);
log.Info($"Persisted message: {s}");
});
});
Command<SaveSnapshotSuccess>(
_ =>
{
log.Info($"Snapshot persisted. State: {_state}");
});
}
}
}
Empty file.