Converted SqlSharding sample to use PostgreSql

infrastructure/postgres/docker-compose.yml

@@ -0,0 +1,20 @@
+# Use postgres/example user/password credentials
+version: '3.1'
+
+services:
+
+  db:
+    image: postgres
+    restart: always
+    environment:
+      POSTGRES_USER: akka-user
+      POSTGRES_PASSWORD: example
+      POSTGRES_DB: Akka
+    ports:
+      - 5432:5432 
+
+  adminer:
+    image: adminer
+    restart: always
+    ports:
+      - 8080:8080

src/clustering/sharding-sqlserver/SqlSharding.Host/Actors/ProductTotalsActor.cs

@@ -26,6 +26,11 @@ public static Props GetProps(string persistenceId)
     public const string TotalsEntityNameConstant = "totals";
 
     private readonly ILoggingAdapter _log = Context.GetLogger();
+
+
+    public override string PersistenceId { get; }
+
+    public ProductState State { get; set; }
 
     public ProductTotalsActor(string persistenceId)
     {
@@ -48,7 +53,16 @@ public ProductTotalsActor(string persistenceId)
         Command<IProductCommand>(cmd =>
         {
             var response = State.ProcessCommand(cmd);
+
+            if (!response.Success)
+            {
+                Sender.Tell(response);
+                // bail out
+                return;
+            }
+
             var sentResponse = false;
+
             PersistAllAsync(response.ResponseEvents, productEvent =>
             {
                 _log.Info("Processed: {0}", productEvent);
@@ -86,8 +100,4 @@ public ProductTotalsActor(string persistenceId)
             }
         });
     }
-
-    public override string PersistenceId { get; }
-
-    public ProductState State { get; set; }
 }

src/clustering/sharding-sqlserver/SqlSharding.Host/Program.cs

@@ -3,7 +3,9 @@
 using Akka.Actor;
 using Akka.Cluster.Hosting;
 using Akka.Cluster.Sharding;
+using Akka.DependencyInjection;
 using Akka.Hosting;
+using Akka.Persistence.PostgreSql.Hosting;
 using Akka.Persistence.SqlServer.Hosting;
 using Akka.Remote.Hosting;
 using Microsoft.Extensions.Configuration;
@@ -21,7 +23,7 @@
     .ConfigureServices((context, services) =>
     {
         // maps to environment variable ConnectionStrings__AkkaSqlConnection
-        var connectionString = context.Configuration.GetConnectionString("AkkaSqlConnection");
+        var connectionString = context.Configuration.GetConnectionString("AkkaPostgresConnection");
 
 
         var akkaSection = context.Configuration.GetSection("Akka");
@@ -42,7 +44,7 @@
                 .AddAppSerialization()
                 .WithClustering(new ClusterOptions()
                     { Roles = new[] { ProductActorProps.SingletonActorRole }, SeedNodes = seeds })
-                .WithSqlServerPersistence(connectionString)
+                .WithPostgreSqlPersistence(connectionString, autoInitialize:true)
                 .WithShardRegion<ProductMarker>("products", s => ProductTotalsActor.GetProps(s),
                     new ProductMessageRouter(),
                     new ShardOptions()
@@ -54,7 +56,8 @@
                 {
                     var shardRegion = registry.Get<ProductMarker>();
 
-                    var indexProps = Props.Create(() => new ProductIndexActor(shardRegion));
+                    var depR = DependencyResolver.For(system);
+                    var indexProps = depR.Props<ProductIndexActor>(shardRegion);
                     var singletonProps = system.ProductSingletonProps(indexProps);
                     registry.TryRegister<ProductIndexActor>(system.ActorOf(singletonProps,
                         ProductActorProps.SingletonActorName));

src/clustering/sharding-sqlserver/SqlSharding.Host/SqlSharding.Host.csproj

@@ -13,6 +13,7 @@
 
     <ItemGroup>
         <PackageReference Include="Akka.Persistence.SqlServer.Hosting" Version="$(AkkaHostingVersion)" />
+        <PackageReference Include="Akka.Persistence.PostgreSql.Hosting" Version="$(AkkaHostingVersion)" />
         <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="$(MicrosoftExtensionsVersion)" />
         <PackageReference Include="Microsoft.Extensions.Configuration.EnvironmentVariables" Version="$(MicrosoftExtensionsVersion)" />
     </ItemGroup>

src/clustering/sharding-sqlserver/SqlSharding.Host/appsettings.Development.json

@@ -7,6 +7,7 @@
         }
     },
     "ConnectionStrings": {
-        "AkkaSqlConnection": "Server=localhost,1533; Database=Akka; User Id=sa; Password=yourStrong(!)Password;"
+        "AkkaSqlConnection": "Server=localhost,1533; Database=Akka; User Id=sa; Password=yourStrong(!)Password;",
+        "AkkaPostgresConnection": "User ID=akka-user; Password=example; Host=localhost;Database=Akka;"
     }
 }

src/clustering/sharding-sqlserver/SqlSharding.Shared/Sharding/ProductActorProps.cs

@@ -1,5 +1,6 @@
 using Akka.Actor;
 using Akka.Cluster.Tools.Singleton;
+using Akka.DependencyInjection;
 
 namespace SqlSharding.Shared.Sharding;
 

src/clustering/sharding-sqlserver/SqlSharding.WebApp/Program.cs

@@ -1,5 +1,6 @@
 using Akka.Actor;
 using Akka.Cluster.Hosting;
+using Akka.DependencyInjection;
 using Akka.Hosting;
 using Akka.Remote.Hosting;
 using SqlSharding.Shared.Serialization;
@@ -24,15 +25,15 @@
 builder.Services.AddAkka("SqlSharding", (configurationBuilder, provider) =>
 {
     configurationBuilder.WithRemoting(hostName, port)
-        .AddAppSerialization()
-        .WithClustering(new ClusterOptions()
-            { Roles = new[] { "Web" }, SeedNodes = seeds })
-        .WithShardRegionProxy<ProductMarker>("products", ProductActorProps.SingletonActorRole,
-            new ProductMessageRouter())
-        .WithActors((system, registry) =>
-        {
-            var proxyProps = system.ProductIndexProxyProps();
-            registry.TryRegister<ProductIndexMarker>(system.ActorOf(proxyProps, "product-proxy"));
+              .AddAppSerialization()
+              .WithClustering(new ClusterOptions()
+                  { Roles = new[] { "Web" }, SeedNodes = seeds })
+              .WithShardRegionProxy<ProductMarker>("products", ProductActorProps.SingletonActorRole,
+                  new ProductMessageRouter())
+              .WithActors((system, registry) =>
+              {
+                  var proxyProps = system.ProductIndexProxyProps();
+                  registry.TryRegister<ProductIndexMarker>(system.ActorOf(proxyProps, "product-proxy"));
         });
 });