Skip to content
This repository was archived by the owner on Mar 20, 2025. It is now read-only.

Commit 6603fba

Browse files
authored
Move Hosting extension project from Akka.Hosting repo to this repo (#206)
* Move Hosting extension project from Akka.Hosting repo to here * Add Akka.Hosting.TestKit version
1 parent 66a499b commit 6603fba

13 files changed

+995
-0
lines changed
Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
<Project Sdk="Microsoft.NET.Sdk">
2+
<PropertyGroup>
3+
<TargetFramework>$(NetStandardLibVersion)</TargetFramework>
4+
<PackageReadmeFile>README.md</PackageReadmeFile>
5+
<Description>Akka.Persistence.PostgreSql Microsoft.Extensions.Hosting support.</Description>
6+
</PropertyGroup>
7+
8+
<ItemGroup>
9+
<PackageReference Include="Akka.Persistence.Hosting"/>
10+
<PackageReference Include="Akka.Persistence.Query.Sql"/>
11+
</ItemGroup>
12+
13+
<ItemGroup>
14+
<ProjectReference Include="..\Akka.Persistence.PostgreSql\Akka.Persistence.PostgreSql.csproj" />
15+
</ItemGroup>
16+
</Project>
Lines changed: 246 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,246 @@
1+
using System;
2+
using Akka.Actor;
3+
using Akka.Hosting;
4+
using Akka.Persistence.Hosting;
5+
6+
#nullable enable
7+
namespace Akka.Persistence.PostgreSql.Hosting
8+
{
9+
/// <summary>
10+
/// Extension methods for Akka.Persistence.PostgreSql
11+
/// </summary>
12+
public static class AkkaPersistencePostgreSqlHostingExtensions
13+
{
14+
/// <summary>
15+
/// Add Akka.Persistence.PostgreSql support to the <see cref="ActorSystem"/>
16+
/// </summary>
17+
/// <param name="builder">
18+
/// The builder instance being configured.
19+
/// </param>
20+
/// <param name="connectionString">
21+
/// Connection string used for database access.
22+
/// </param>
23+
/// <param name="mode">
24+
/// <para>
25+
/// Determines which settings should be added by this method call.
26+
/// </para>
27+
/// <i>Default</i>: <see cref="PersistenceMode.Both"/>
28+
/// </param>
29+
/// <param name="schemaName">
30+
/// <para>
31+
/// The schema name for the journal and snapshot store table.
32+
/// </para>
33+
/// <i>Default</i>: <c>"public"</c>
34+
/// </param>
35+
/// <param name="autoInitialize">
36+
/// <para>
37+
/// Should the SQL store table be initialized automatically.
38+
/// </para>
39+
/// <i>Default</i>: <c>false</c>
40+
/// </param>
41+
/// <param name="storedAsType">
42+
/// <para>
43+
/// Determines how data are being de/serialized into the table.
44+
/// </para>
45+
/// <i>Default</i>: <see cref="StoredAsType.ByteA"/>
46+
/// </param>
47+
/// <param name="sequentialAccess">
48+
/// <para>
49+
/// Uses the `CommandBehavior.SequentialAccess` when creating SQL commands, providing a performance
50+
/// improvement for reading large BLOBS.
51+
/// </para>
52+
/// <i>Default</i>: <c>false</c>
53+
/// </param>
54+
/// <param name="useBigintIdentityForOrderingColumn">
55+
/// <para>
56+
/// When set to true, persistence will use `BIGINT` and `GENERATED ALWAYS AS IDENTITY` for journal table
57+
/// schema creation.
58+
/// </para>
59+
/// <i>Default</i>: <c>false</c>
60+
/// </param>
61+
/// <param name="journalBuilder">
62+
/// <para>
63+
/// An <see cref="Action{T}"/> used to configure an <see cref="AkkaPersistenceJournalBuilder"/> instance.
64+
/// </para>
65+
/// <i>Default</i>: <c>null</c>
66+
/// </param>
67+
/// <param name="pluginIdentifier">
68+
/// <para>
69+
/// The configuration identifier for the plugins
70+
/// </para>
71+
/// <i>Default</i>: <c>"postgresql"</c>
72+
/// </param>
73+
/// <param name="isDefaultPlugin">
74+
/// <para>
75+
/// A <c>bool</c> flag to set the plugin as the default persistence plugin for the <see cref="ActorSystem"/>
76+
/// </para>
77+
/// <i>Default</i>: <c>true</c>
78+
/// </param>
79+
/// <returns>
80+
/// The same <see cref="AkkaConfigurationBuilder"/> instance originally passed in.
81+
/// </returns>
82+
public static AkkaConfigurationBuilder WithPostgreSqlPersistence(
83+
this AkkaConfigurationBuilder builder,
84+
string connectionString,
85+
PersistenceMode mode = PersistenceMode.Both,
86+
string schemaName = "public",
87+
bool autoInitialize = false,
88+
StoredAsType storedAsType = StoredAsType.ByteA,
89+
bool sequentialAccess = false,
90+
bool useBigintIdentityForOrderingColumn = false,
91+
Action<AkkaPersistenceJournalBuilder>? journalBuilder = null,
92+
string pluginIdentifier = "postgresql",
93+
bool isDefaultPlugin = true)
94+
{
95+
if (mode == PersistenceMode.SnapshotStore && journalBuilder is { })
96+
throw new Exception($"{nameof(journalBuilder)} can only be set when {nameof(mode)} is set to either {PersistenceMode.Both} or {PersistenceMode.Journal}");
97+
98+
var journalOpt = new PostgreSqlJournalOptions(isDefaultPlugin, pluginIdentifier)
99+
{
100+
ConnectionString = connectionString,
101+
SchemaName = schemaName,
102+
AutoInitialize = autoInitialize,
103+
StoredAs = storedAsType,
104+
SequentialAccess = sequentialAccess,
105+
UseBigIntIdentityForOrderingColumn = useBigintIdentityForOrderingColumn
106+
};
107+
108+
var adapters = new AkkaPersistenceJournalBuilder(journalOpt.Identifier, builder);
109+
journalBuilder?.Invoke(adapters);
110+
journalOpt.Adapters = adapters;
111+
112+
var snapshotOpt = new PostgreSqlSnapshotOptions(isDefaultPlugin, pluginIdentifier)
113+
{
114+
ConnectionString = connectionString,
115+
SchemaName = schemaName,
116+
AutoInitialize = autoInitialize,
117+
StoredAs = storedAsType,
118+
SequentialAccess = sequentialAccess
119+
};
120+
121+
return mode switch
122+
{
123+
PersistenceMode.Journal => builder.WithPostgreSqlPersistence(journalOpt, null),
124+
PersistenceMode.SnapshotStore => builder.WithPostgreSqlPersistence(null, snapshotOpt),
125+
PersistenceMode.Both => builder.WithPostgreSqlPersistence(journalOpt, snapshotOpt),
126+
_ => throw new ArgumentOutOfRangeException(nameof(mode), mode, "Invalid PersistenceMode defined.")
127+
};
128+
}
129+
130+
/// <summary>
131+
/// Add Akka.Persistence.PostgreSql support to the <see cref="ActorSystem"/>
132+
/// </summary>
133+
/// <param name="builder">
134+
/// The builder instance being configured.
135+
/// </param>
136+
/// <param name="snapshotOptionConfigurator">
137+
/// <para>
138+
/// An <see cref="Action{T}"/> that modifies an instance of <see cref="PostgreSqlSnapshotOptions"/>,
139+
/// used to configure the snapshot store plugin
140+
/// </para>
141+
/// <i>Default</i>: <c>null</c>
142+
/// </param>
143+
/// <param name="journalOptionConfigurator">
144+
/// <para>
145+
/// An <see cref="Action{T}"/> that modifies an instance of <see cref="PostgreSqlJournalOptions"/>,
146+
/// used to configure the journal plugin
147+
/// </para>
148+
/// <i>Default</i>: <c>null</c>
149+
/// </param>
150+
/// <param name="isDefaultPlugin">
151+
/// <para>
152+
/// A <c>bool</c> flag to set the plugin as the default persistence plugin for the <see cref="ActorSystem"/>
153+
/// </para>
154+
/// <i>Default</i>: <c>true</c>
155+
/// </param>
156+
/// <returns>
157+
/// The same <see cref="AkkaConfigurationBuilder"/> instance originally passed in.
158+
/// </returns>
159+
/// <exception cref="ArgumentException">
160+
/// Thrown when both <param name="journalOptionConfigurator"/> and
161+
/// <param name="snapshotOptionConfigurator"/> are null.
162+
/// </exception>
163+
public static AkkaConfigurationBuilder WithPostgreSqlPersistence(
164+
this AkkaConfigurationBuilder builder,
165+
Action<PostgreSqlJournalOptions>? journalOptionConfigurator = null,
166+
Action<PostgreSqlSnapshotOptions>? snapshotOptionConfigurator = null,
167+
bool isDefaultPlugin = true)
168+
{
169+
if (journalOptionConfigurator is null && snapshotOptionConfigurator is null)
170+
throw new ArgumentException($"{nameof(journalOptionConfigurator)} and {nameof(snapshotOptionConfigurator)} could not both be null");
171+
172+
PostgreSqlJournalOptions? journalOptions = null;
173+
if(journalOptionConfigurator is { })
174+
{
175+
journalOptions = new PostgreSqlJournalOptions(isDefaultPlugin);
176+
journalOptionConfigurator(journalOptions);
177+
}
178+
179+
PostgreSqlSnapshotOptions? snapshotOptions = null;
180+
if (snapshotOptionConfigurator is { })
181+
{
182+
snapshotOptions = new PostgreSqlSnapshotOptions(isDefaultPlugin);
183+
snapshotOptionConfigurator(snapshotOptions);
184+
}
185+
186+
return builder.WithPostgreSqlPersistence(journalOptions, snapshotOptions);
187+
}
188+
189+
/// <summary>
190+
/// Add Akka.Persistence.PostgreSql support to the <see cref="ActorSystem"/>
191+
/// </summary>
192+
/// <param name="builder">
193+
/// The builder instance being configured.
194+
/// </param>
195+
/// <param name="snapshotOptions">
196+
/// <para>
197+
/// An instance of <see cref="PostgreSqlSnapshotOptions"/>, used to configure the snapshot store plugin
198+
/// </para>
199+
/// <i>Default</i>: <c>null</c>
200+
/// </param>
201+
/// <param name="journalOptions">
202+
/// <para>
203+
/// An instance of <see cref="PostgreSqlJournalOptions"/>, used to configure the journal plugin
204+
/// </para>
205+
/// <i>Default</i>: <c>null</c>
206+
/// </param>
207+
/// <returns>
208+
/// The same <see cref="AkkaConfigurationBuilder"/> instance originally passed in.
209+
/// </returns>
210+
/// <exception cref="ArgumentException">
211+
/// Thrown when both <param name="journalOptions"/> and <param name="snapshotOptions"/> are null.
212+
/// </exception>
213+
public static AkkaConfigurationBuilder WithPostgreSqlPersistence(
214+
this AkkaConfigurationBuilder builder,
215+
PostgreSqlJournalOptions? journalOptions = null,
216+
PostgreSqlSnapshotOptions? snapshotOptions = null)
217+
{
218+
if (journalOptions is null && snapshotOptions is null)
219+
throw new ArgumentException($"{nameof(journalOptions)} and {nameof(snapshotOptions)} could not both be null");
220+
221+
return (journalOptions, snapshotOptions) switch
222+
{
223+
(null, null) =>
224+
throw new ArgumentException($"{nameof(journalOptions)} and {nameof(snapshotOptions)} could not both be null"),
225+
226+
(_, null) =>
227+
builder
228+
.AddHocon(journalOptions.ToConfig(), HoconAddMode.Prepend)
229+
.AddHocon(journalOptions.DefaultConfig, HoconAddMode.Append),
230+
231+
(null, _) =>
232+
builder
233+
.AddHocon(snapshotOptions.ToConfig(), HoconAddMode.Prepend)
234+
.AddHocon(snapshotOptions.DefaultConfig, HoconAddMode.Append),
235+
236+
(_, _) =>
237+
builder
238+
.AddHocon(journalOptions.ToConfig(), HoconAddMode.Prepend)
239+
.AddHocon(snapshotOptions.ToConfig(), HoconAddMode.Prepend)
240+
.AddHocon(journalOptions.DefaultConfig, HoconAddMode.Append)
241+
.AddHocon(snapshotOptions.DefaultConfig, HoconAddMode.Append),
242+
};
243+
}
244+
245+
}
246+
}
Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
// -----------------------------------------------------------------------
2+
// <copyright file="PostgreSqlJournalOptions.cs" company="Akka.NET Project">
3+
// Copyright (C) 2013-2022 .NET Foundation <https://github.com/akkadotnet/akka.net>
4+
// </copyright>
5+
// -----------------------------------------------------------------------
6+
7+
using System;
8+
using System.Data;
9+
using System.Text;
10+
using Akka.Configuration;
11+
using Akka.Persistence.Hosting;
12+
13+
#nullable enable
14+
namespace Akka.Persistence.PostgreSql.Hosting
15+
{
16+
/// <summary>
17+
/// Akka.Hosting options class to set up PostgreSql persistence journal.
18+
/// </summary>
19+
public sealed class PostgreSqlJournalOptions: SqlJournalOptions
20+
{
21+
private static readonly Config Default = PostgreSqlPersistence.DefaultConfiguration()
22+
.GetConfig(PostgreSqlJournalSettings.JournalConfigPath);
23+
24+
/// <summary>
25+
/// Create a new instance of <see cref="PostgreSqlJournalOptions"/>
26+
/// </summary>
27+
public PostgreSqlJournalOptions() : this(true)
28+
{
29+
}
30+
31+
/// <summary>
32+
/// Create a new instance of <see cref="PostgreSqlJournalOptions"/>
33+
/// </summary>
34+
/// <param name="isDefaultPlugin">Indicates if this journal configuration should be the default configuration for all persistence</param>
35+
/// <param name="identifier">The journal configuration identifier. <i>Default</i>: "postgresql"</param>
36+
public PostgreSqlJournalOptions(bool isDefaultPlugin, string identifier = "postgresql") : base(isDefaultPlugin)
37+
{
38+
Identifier = identifier;
39+
}
40+
41+
/// <summary>
42+
/// <para>
43+
/// The plugin identifier for this persistence plugin
44+
/// </para>
45+
/// <b>Default</b>: <c>"postgresql"</c>
46+
/// </summary>
47+
public override string Identifier { get; set; }
48+
49+
/// <summary>
50+
/// <para>
51+
/// PostgreSQL schema name to table corresponding with persistent journal.
52+
/// </para>
53+
/// <b>Default</b>: <c>"public"</c>
54+
/// </summary>
55+
public override string SchemaName { get; set; } = "public";
56+
57+
/// <summary>
58+
/// <para>
59+
/// PostgreSQL table corresponding with persistent journal.
60+
/// </para>
61+
/// <b>Default</b>: <c>"event_journal"</c>
62+
/// </summary>
63+
public override string TableName { get; set; } = "event_journal";
64+
65+
/// <summary>
66+
/// <para>
67+
/// PostgreSQL table corresponding with persistent journal metadata.
68+
/// </para>
69+
/// <b>Default</b>: <c>"metadata"</c>
70+
/// </summary>
71+
public override string MetadataTableName { get; set; } = "metadata";
72+
73+
/// <summary>
74+
/// <para>
75+
/// Uses the CommandBehavior.SequentialAccess when creating DB commands, providing a performance
76+
/// improvement for reading large BLOBS.
77+
/// </para>
78+
/// <b>Default</b>: <c>false</c>
79+
/// </summary>
80+
public override bool SequentialAccess { get; set; } = false;
81+
82+
/// <summary>
83+
/// <para>
84+
/// Postgres data type for payload column
85+
/// </para>
86+
/// <b>Default</b>: <see cref="StoredAsType.ByteA"/>
87+
/// </summary>
88+
public StoredAsType StoredAs { get; set; } = StoredAsType.ByteA;
89+
90+
/// <summary>
91+
/// <para>
92+
/// When turned on, persistence will use `BIGINT` and `GENERATED ALWAYS AS IDENTITY` for the ordering
93+
/// column in the journal table during schema creation.
94+
/// </para>
95+
/// <b>Default</b>: <c>false</c>
96+
/// </summary>
97+
public bool UseBigIntIdentityForOrderingColumn { get; set; } = false;
98+
99+
/// <inheritdoc/>
100+
public override IsolationLevel ReadIsolationLevel { get; set; } = IsolationLevel.Unspecified;
101+
102+
/// <inheritdoc/>
103+
public override IsolationLevel WriteIsolationLevel { get; set; } = IsolationLevel.Unspecified;
104+
105+
protected override Config InternalDefaultConfig => Default;
106+
107+
protected override StringBuilder Build(StringBuilder sb)
108+
{
109+
sb.AppendLine($"use-bigint-identity-for-ordering-column = {(UseBigIntIdentityForOrderingColumn ? "on" : "off")}");
110+
sb.AppendLine($"stored-as = {StoredAs.ToHocon()}");
111+
112+
return base.Build(sb);
113+
}
114+
}
115+
}

0 commit comments

Comments
 (0)