Merge pull request #54 from AkkaNetContrib/dev

Akka.Persistence.PostgreSql v1.3.8 release

Author: Aaronontheweb

README.md

@@ -91,6 +91,7 @@ CREATE TABLE {your_journal_table_name} (
     manifest VARCHAR(500) NOT NULL,
     payload BYTEA NOT NULL,
     tags VARCHAR(100) NULL,
+    serializer_id INTEGER NULL,
     CONSTRAINT {your_journal_table_name}_uq UNIQUE (persistence_id, sequence_nr)
 );
 
@@ -99,7 +100,8 @@ CREATE TABLE {your_snapshot_table_name} (
     sequence_nr BIGINT NOT NULL,
     created_at BIGINT NOT NULL,
     manifest VARCHAR(500) NOT NULL,
-    snapshot BYTEA NOT NULL,
+    payload BYTEA NOT NULL,
+    serializer_id INTEGER NULL,
     CONSTRAINT {your_snapshot_table_name}_pk PRIMARY KEY (persistence_id, sequence_nr)
 );
 
@@ -155,4 +157,4 @@ or run postgres in docker
 
 ```
 docker run -d --rm --name=akka-postgres-db -p 5432:5432 -l deployer=akkadotnet -e POSTGRES_USER=postgres -e POSTGRES_PASSWORD=postgres postgres:9.6
-```
+```

RELEASE_NOTES.md

@@ -1,4 +1,17 @@
+#### 1.3.8 July 6 2018 ####
+Upgraded to support Akka.NET 1.3.8 and to take advantage of some performance improvements that have been added to Akka.Persistence for loading large snapshots, which you can read more about here: https://github.com/akkadotnet/akka.net/issues/3422
+
+Note that this feature is currently disabled by default in Akka.Persistence.PostgreSql due to https://github.com/AkkaNetContrib/Akka.Persistence.PostgreSql/issues/53
+
 #### 1.3.1 September 11 2017 ####
+Support for Akka.NET 1.3, .NET Standard 1.6, and the first stable RTM release of Akka.Persistence.
+
+Migration from 1.1.0-beta Up**
+The event journal and snapshot store schema has changed with this release.  In order to keep existing stores compatible with this release, you **must** add a column to both stores for `SerializerId` like so:
+```sql
+ALTER TABLE {your_journal_table_name} ADD COLUMN SerializerId INTEGER NULL
+ALTER TABLE {your_snapshot_table_name} ADD COLUMN SerializerId INTEGER NULL
+```
 
 #### 1.1.2 January 2017 ####
 

src/Akka.Persistence.PostgreSql.Tests/Akka.Persistence.PostgreSql.Tests.csproj

@@ -7,12 +7,9 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.3.0-preview-20170425-07" />
-    <PackageReference Include="Akka" Version="1.3.1" />
-    <PackageReference Include="Akka.Persistence" Version="1.3.1" />
-    <PackageReference Include="Akka.Persistence.Sql.Common" Version="1.3.1" />
-    <PackageReference Include="Akka.Persistence.Sql.TestKit" Version="1.3.1" />
-    <PackageReference Include="Akka.TestKit" Version="1.3.1" />
+    <PackageReference Include="Akka.Persistence.Sql.Common" Version="1.3.8" />
+    <PackageReference Include="Akka.Persistence.Sql.TestKit" Version="1.3.8" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.3.0" />
     <PackageReference Include="Microsoft.Extensions.Configuration" Version="1.1.2" />
     <PackageReference Include="Microsoft.Extensions.Configuration.Xml" Version="1.1.2" />
     <PackageReference Include="Npgsql" Version="3.2.5" />

src/Akka.Persistence.PostgreSql.Tests/PostgreSqlConfigSpec.cs

@@ -31,6 +31,7 @@ public void Should_PostgreSql_journal_has_default_config()
             Assert.Equal("metadata", config.GetString("metadata-table-name"));
             Assert.False(config.GetBoolean("auto-initialize"));
             Assert.Equal("Akka.Persistence.Sql.Common.Journal.DefaultTimestampProvider, Akka.Persistence.Sql.Common", config.GetString("timestamp-provider"));
+            Assert.False(config.GetBoolean("sequential-access"));
         }
 
         [Fact]
@@ -49,6 +50,7 @@ public void Should_PostgreSql_snapshot_has_default_config()
             Assert.Equal("public", config.GetString("schema-name"));
             Assert.Equal("snapshot_store", config.GetString("table-name"));
             Assert.False(config.GetBoolean("auto-initialize"));
+            Assert.False(config.GetBoolean("sequential-access"));
         }
     }
 }

src/Akka.Persistence.PostgreSql.Tests/app.xml

@@ -2,5 +2,6 @@
 <configuration>
   <connectionStrings>
      <add name="TestDb" connectionString="Server=db;Port=5432;Database=postgres;User Id=postgres;Password=postgres;" />
+     <!--<add name="TestDb" connectionString="Server=127.0.0.1;Port=5432;Database=postgres;User Id=postgres;Password=postgres;" />-->
     </connectionStrings>
 </configuration>

src/Akka.Persistence.PostgreSql/Akka.Persistence.PostgreSql.csproj

@@ -20,10 +20,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Akka" Version="1.3.1" />
-    <PackageReference Include="Akka.Persistence" Version="1.3.1" />
-    <PackageReference Include="Akka.Persistence.Sql.Common" Version="1.3.1" />
-    <PackageReference Include="Newtonsoft.Json" Version="9.0.1" />
+    <PackageReference Include="Akka.Persistence.Sql.Common" Version="1.3.8" />
     <PackageReference Include="Npgsql" Version="3.2.5" />
   </ItemGroup>
 

src/Akka.Persistence.PostgreSql/Journal/PostgreSqlJournal.cs

@@ -47,7 +47,8 @@ public PostgreSqlJournal(Config journalConfig) : base(journalConfig)
                 serializerIdColumnName: "serializer_id",
                 timeout: config.GetTimeSpan("connection-timeout"),
                 storedAs: storedAs,
-                defaultSerializer: config.GetString("serializer")),
+                defaultSerializer: config.GetString("serializer"),
+                useSequentialAccess: config.GetBoolean("sequential-access")),
                     Context.System.Serialization,
                     GetTimestampProvider(config.GetString("timestamp-provider")));
 

src/Akka.Persistence.PostgreSql/Journal/PostgreSqlQueryExecutor.cs

@@ -191,9 +191,11 @@ public PostgreSqlQueryConfiguration(
             TimeSpan timeout,
             StoredAsType storedAs,
             string defaultSerializer,
-            JsonSerializerSettings jsonSerializerSettings = null)
+            JsonSerializerSettings jsonSerializerSettings = null, 
+            bool useSequentialAccess = true)
             : base(schemaName, journalEventsTableName, metaTableName, persistenceIdColumnName, sequenceNrColumnName,
-                  payloadColumnName, manifestColumnName, timestampColumnName, isDeletedColumnName, tagsColumnName, orderingColumn, serializerIdColumnName, timeout, defaultSerializer)
+                  payloadColumnName, manifestColumnName, timestampColumnName, isDeletedColumnName, tagsColumnName, orderingColumn, 
+                serializerIdColumnName, timeout, defaultSerializer, useSequentialAccess)
         {
             StoredAs = storedAs;
             JsonSerializerSettings = jsonSerializerSettings ?? new JsonSerializerSettings

src/Akka.Persistence.PostgreSql/Snapshot/PostgreSqlQueryExecutor.cs

@@ -6,6 +6,8 @@
 //-----------------------------------------------------------------------
 
 using Akka.Persistence.Sql.Common.Snapshot;
+using Akka.Serialization;
+using Akka.Util;
 using Newtonsoft.Json;
 using Npgsql;
 using NpgsqlTypes;
@@ -99,7 +101,7 @@ WITH upsert AS (
 
         protected override DbCommand CreateCommand(DbConnection connection)
         {
-            return ((NpgsqlConnection) connection).CreateCommand();
+            return ((NpgsqlConnection)connection).CreateCommand();
         }
 
         protected override void SetTimestampParameter(DateTime timestamp, DbCommand command) => AddParameter(command, "@Timestamp", DbType.Int64, timestamp.Ticks);
@@ -109,6 +111,25 @@ protected override void SetPayloadParameter(object snapshot, DbCommand command)
             command.Parameters.Add(new NpgsqlParameter("@Payload", serializationResult.DbType) { Value = serializationResult.Payload });
         }
 
+        protected override void SetManifestParameters(object snapshot, DbCommand command)
+        {
+            var snapshotType = snapshot.GetType();
+            var serializer = Serialization.FindSerializerForType(snapshotType, Configuration.DefaultSerializer);
+
+            string manifest = "";
+            if (serializer is SerializerWithStringManifest)
+            {
+                manifest = ((SerializerWithStringManifest)serializer).Manifest(snapshot);
+            }
+            else if (!serializer.IncludeManifest)
+            {
+                manifest = snapshotType.TypeQualifiedName();
+            }
+
+            AddParameter(command, "@Manifest", DbType.String, manifest);
+            AddParameter(command, "@SerializerId", DbType.Int32, serializer.Identifier);
+        }
+
         protected override SelectedSnapshot ReadSnapshot(DbDataReader reader)
         {
             var persistenceId = reader.GetString(0);
@@ -118,14 +139,12 @@ protected override SelectedSnapshot ReadSnapshot(DbDataReader reader)
 
             int? serializerId = null;
             Type type = null;
-            if (reader.IsDBNull(5))
-            {
-                type = Type.GetType(manifest, true);
-            }
-            else
-            {
+
+            if (!string.IsNullOrEmpty(manifest))
+                type = Type.GetType(manifest, throwOnError: true);
+
+            if (!reader.IsDBNull(5))
                 serializerId = reader.GetInt32(5);
-            }
 
             var snapshot = _deserialize(type, reader[4], manifest, serializerId);
 
@@ -143,19 +162,21 @@ public class PostgreSqlQueryConfiguration : QueryConfiguration
         public readonly JsonSerializerSettings JsonSerializerSettings;
 
         public PostgreSqlQueryConfiguration(
-            string schemaName, 
-            string snapshotTableName, 
-            string persistenceIdColumnName, 
-            string sequenceNrColumnName, 
-            string payloadColumnName, 
-            string manifestColumnName, 
-            string timestampColumnName, 
+            string schemaName,
+            string snapshotTableName,
+            string persistenceIdColumnName,
+            string sequenceNrColumnName,
+            string payloadColumnName,
+            string manifestColumnName,
+            string timestampColumnName,
             string serializerIdColumnName,
-            TimeSpan timeout, 
+            TimeSpan timeout,
             StoredAsType storedAs,
             string defaultSerializer,
-            JsonSerializerSettings jsonSerializerSettings = null) 
-            : base(schemaName, snapshotTableName, persistenceIdColumnName, sequenceNrColumnName, payloadColumnName, manifestColumnName, timestampColumnName, serializerIdColumnName, timeout, defaultSerializer)
+            JsonSerializerSettings jsonSerializerSettings = null,
+            bool useSequentialAccess = true)
+            : base(schemaName, snapshotTableName, persistenceIdColumnName, sequenceNrColumnName, payloadColumnName, 
+                manifestColumnName, timestampColumnName, serializerIdColumnName, timeout, defaultSerializer, useSequentialAccess)
         {
             StoredAs = storedAs;
             JsonSerializerSettings = jsonSerializerSettings ?? new JsonSerializerSettings

src/Akka.Persistence.PostgreSql/Snapshot/PostgreSqlSnapshotStore.cs

@@ -41,7 +41,8 @@ public PostgreSqlSnapshotStore(Config snapshotConfig) : base(snapshotConfig)
                 serializerIdColumnName: "serializer_id",
                 timeout: config.GetTimeSpan("connection-timeout"),
                 storedAs: storedAs,
-                defaultSerializer: config.GetString("serializer")),
+                defaultSerializer: config.GetString("serializer"),
+                useSequentialAccess: config.GetBoolean("sequential-access")),
                     Context.System.Serialization);
 
             SnapshotSettings = new PostgreSqlSnapshotStoreSettings(config);