Persisence serializer setting

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs

@@ -224,6 +224,11 @@ public abstract class BatchingSqlJournalSetup
         public QueryConfiguration NamingConventions { get; }
 
         /// <summary>
+        /// The default serializer used when not type override matching is found
+        /// </summary>
+        public string DefaultSerializer { get; }
+
+    /// <summary>
         /// Initializes a new instance of the <see cref="BatchingSqlJournalSetup" /> class.
         /// </summary>
         /// <param name="config">The configuration used to configure the journal.</param>
@@ -281,6 +286,7 @@ protected BatchingSqlJournalSetup(Config config, QueryConfiguration namingConven
             CircuitBreakerSettings = new CircuitBreakerSettings(config.GetConfig("circuit-breaker"));
             ReplayFilterSettings = new ReplayFilterSettings(config.GetConfig("replay-filter"));
             NamingConventions = namingConventions;
+            DefaultSerializer = config.GetString("serializer");
         }
 
         /// <summary>
@@ -302,7 +308,8 @@ protected BatchingSqlJournalSetup(Config config, QueryConfiguration namingConven
         /// </param>
         /// <param name="replayFilterSettings">The settings used when replaying events from database back to the persistent actors.</param>
         /// <param name="namingConventions">The naming conventions used by the database to construct valid SQL statements.</param>
-        protected BatchingSqlJournalSetup(string connectionString, int maxConcurrentOperations, int maxBatchSize, int maxBufferSize, bool autoInitialize, TimeSpan connectionTimeout, IsolationLevel isolationLevel, CircuitBreakerSettings circuitBreakerSettings, ReplayFilterSettings replayFilterSettings, QueryConfiguration namingConventions)
+        /// <param name="defaultSerializer">The serializer used when no specific type matching can be found.</param>
+        protected BatchingSqlJournalSetup(string connectionString, int maxConcurrentOperations, int maxBatchSize, int maxBufferSize, bool autoInitialize, TimeSpan connectionTimeout, IsolationLevel isolationLevel, CircuitBreakerSettings circuitBreakerSettings, ReplayFilterSettings replayFilterSettings, QueryConfiguration namingConventions, string defaultSerializer)
         {
             ConnectionString = connectionString;
             MaxConcurrentOperations = maxConcurrentOperations;
@@ -314,6 +321,7 @@ protected BatchingSqlJournalSetup(string connectionString, int maxConcurrentOper
             CircuitBreakerSettings = circuitBreakerSettings;
             ReplayFilterSettings = replayFilterSettings;
             NamingConventions = namingConventions;
+            DefaultSerializer = defaultSerializer;
         }
     }
 
@@ -1148,7 +1156,7 @@ protected virtual void WriteEvent(TCommand command, IPersistentRepresentation pe
             var manifest = string.IsNullOrEmpty(persistent.Manifest)
                 ? payloadType.TypeQualifiedName()
                 : persistent.Manifest;
-            var serializer = _serialization.FindSerializerForType(payloadType);
+            var serializer = _serialization.FindSerializerForType(payloadType, Setup.DefaultSerializer);
             var binary = serializer.ToBinary(persistent.Payload);
 
             AddParameter(command, "@PersistenceId", DbType.String, persistent.PersistenceId);
@@ -1174,7 +1182,7 @@ protected virtual IPersistentRepresentation ReadEvent(DbDataReader reader)
             var payload = reader[PayloadIndex];
 
             var type = Type.GetType(manifest, true);
-            var deserializer = _serialization.FindSerializerForType(type);
+            var deserializer = _serialization.FindSerializerForType(type, Setup.DefaultSerializer);
             var deserialized = deserializer.FromBinary((byte[])payload, type);
 
             var persistent = new Persistent(deserialized, sequenceNr, persistenceId, manifest, isDeleted, ActorRefs.NoSender, null);

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/QueryExecutor.cs

@@ -162,6 +162,11 @@ public class QueryConfiguration
         /// </summary>
         public readonly TimeSpan Timeout;
 
+        /// <summary>
+        /// The default serializer used when not type override matching is found
+        /// </summary>
+        public string DefaultSerializer { get; }
+
         /// <summary>
         /// TBD
         /// </summary>
@@ -177,6 +182,7 @@ public class QueryConfiguration
         /// <param name="tagsColumnName">TBD</param>
         /// <param name="orderingColumnName">TBD</param>
         /// <param name="timeout">TBD</param>
+        /// <param name="defaultSerializer">The default serializer used when not type override matching is found</param>
         public QueryConfiguration(
             string schemaName,
             string journalEventsTableName,
@@ -189,7 +195,8 @@ public QueryConfiguration(
             string isDeletedColumnName,
             string tagsColumnName,
             string orderingColumnName,
-            TimeSpan timeout)
+            TimeSpan timeout,
+            string defaultSerializer)
         {
             SchemaName = schemaName;
             JournalEventsTableName = journalEventsTableName;
@@ -203,6 +210,7 @@ public QueryConfiguration(
             Timeout = timeout;
             TagsColumnName = tagsColumnName;
             OrderingColumnName = orderingColumnName;
+            DefaultSerializer = defaultSerializer;
         }
 
         /// <summary>
@@ -658,7 +666,7 @@ protected virtual IPersistentRepresentation ReadEvent(DbDataReader reader)
             var payload = reader[PayloadIndex];
 
             var type = Type.GetType(manifest, true);
-            var deserializer = Serialization.FindSerializerForType(type);
+            var deserializer = Serialization.FindSerializerForType(type, Configuration.DefaultSerializer);
             var deserialized = deserializer.FromBinary((byte[])payload, type);
 
             return new Persistent(deserialized, sequenceNr, persistenceId, manifest, isDeleted, ActorRefs.NoSender, null);

src/contrib/persistence/Akka.Persistence.Sql.Common/Settings.cs

@@ -113,6 +113,11 @@ public class SnapshotStoreSettings
         /// </summary>
         public bool AutoInitialize { get; private set; }
 
+        /// <summary>
+        /// The default serializer being used if no type match override is specified
+        /// </summary>
+        public string DefaultSerializer { get; private set; }
+
         /// <summary>
         /// Initializes a new instance of the <see cref="SnapshotStoreSettings"/> class.
         /// </summary>
@@ -130,6 +135,7 @@ public SnapshotStoreSettings(Config config)
             SchemaName = config.GetString("schema-name");
             TableName = config.GetString("table-name");
             AutoInitialize = config.GetBoolean("auto-initialize");
+            DefaultSerializer = config.GetString("serializer");
         }
 
         /// <summary>

src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/QueryExecutor.cs

@@ -102,6 +102,11 @@ public class QueryConfiguration
         /// </summary>
         public readonly TimeSpan Timeout;
 
+        /// <summary>
+        /// The default serializer used when not type override matching is found
+        /// </summary>
+        public readonly string DefaultSerializer;
+
         /// <summary>
         /// TBD
         /// </summary>
@@ -113,6 +118,7 @@ public class QueryConfiguration
         /// <param name="manifestColumnName">TBD</param>
         /// <param name="timestampColumnName">TBD</param>
         /// <param name="timeout">TBD</param>
+        /// <param name="defaultSerializer">The default serializer used when not type override matching is found</param>
         public QueryConfiguration(
             string schemaName,
             string snapshotTableName,
@@ -121,7 +127,8 @@ public QueryConfiguration(
             string payloadColumnName,
             string manifestColumnName,
             string timestampColumnName,
-            TimeSpan timeout)
+            TimeSpan timeout, 
+            string defaultSerializer)
         {
             SchemaName = schemaName;
             SnapshotTableName = snapshotTableName;
@@ -131,6 +138,7 @@ public QueryConfiguration(
             ManifestColumnName = manifestColumnName;
             TimestampColumnName = timestampColumnName;
             Timeout = timeout;
+            DefaultSerializer = defaultSerializer;
         }
 
         /// <summary>
@@ -309,7 +317,7 @@ DELETE FROM {Configuration.FullSnapshotTableName}
         protected virtual void SetPayloadParameter(object snapshot, DbCommand command)
         {
             var snapshotType = snapshot.GetType();
-            var serializer = Serialization.FindSerializerForType(snapshotType);
+            var serializer = Serialization.FindSerializerForType(snapshotType, Configuration.DefaultSerializer);
 
             var binary = serializer.ToBinary(snapshot);
             AddParameter(command, "@Payload", DbType.Binary, binary);
@@ -521,7 +529,7 @@ protected virtual SelectedSnapshot ReadSnapshot(DbDataReader reader)
         protected object GetSnapshot(DbDataReader reader)
         {
             var type = Type.GetType(reader.GetString(3), true);
-            var serializer = Serialization.FindSerializerForType(type);
+            var serializer = Serialization.FindSerializerForType(type, Configuration.DefaultSerializer);
             var binary = (byte[])reader[4];
 
             var obj = serializer.FromBinary(binary, type);

src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/SqlSnapshotStore.cs

@@ -217,7 +217,7 @@ protected override async Task DeleteAsync(string persistenceId, SnapshotSelectio
         private SnapshotEntry ToSnapshotEntry(SnapshotMetadata metadata, object snapshot)
         {
             var snapshotType = snapshot.GetType();
-            var serializer = Context.System.Serialization.FindSerializerForType(snapshotType);
+            var serializer = Context.System.Serialization.FindSerializerForType(snapshotType, _settings.DefaultSerializer);
 
             var binary = serializer.ToBinary(snapshot);
 

src/contrib/persistence/Akka.Persistence.Sqlite/Journal/BatchingSqliteJournal.cs

@@ -12,6 +12,7 @@
 using System.Data.Common;
 using Microsoft.Data.Sqlite;
 using Akka.Configuration;
+using Akka.Pattern;
 using Akka.Persistence.Sql.Common.Journal;
 
 namespace Akka.Persistence.Sqlite.Journal
@@ -22,9 +23,9 @@ namespace Akka.Persistence.Sqlite.Journal
     public sealed class BatchingSqliteJournalSetup : BatchingSqlJournalSetup
     {
         /// <summary>
-        /// TBD
+        /// Initializes a new instance of the <see cref="BatchingSqliteJournalSetup" /> class.
         /// </summary>
-        /// <param name="config">TBD</param>
+        /// <param name="config">Config object used to obtain Journal settings</param>
         public BatchingSqliteJournalSetup(Config config) : base(config, new QueryConfiguration(
                     schemaName: null,
                     journalEventsTableName: config.GetString("table-name"),
@@ -37,26 +38,34 @@ public sealed class BatchingSqliteJournalSetup : BatchingSqlJournalSetup
                     isDeletedColumnName: "is_deleted",
                     tagsColumnName: "tags",
                     orderingColumnName: "ordering",
-                    timeout: config.GetTimeSpan("connection-timeout")))
+                    timeout: config.GetTimeSpan("connection-timeout"),
+                    defaultSerializer: config.GetString("serializer")))
         {
         }
 
         /// <summary>
-        /// TBD
+        /// Initializes a new instance of the <see cref="BatchingSqliteJournalSetup" /> class.
         /// </summary>
-        /// <param name="connectionString">TBD</param>
-        /// <param name="maxConcurrentOperations">TBD</param>
-        /// <param name="maxBatchSize">TBD</param>
-        /// <param name="maxBufferSize">TBD</param>
-        /// <param name="autoInitialize">TBD</param>
-        /// <param name="connectionTimeout">TBD</param>
-        /// <param name="isolationLevel">TBD</param>
-        /// <param name="circuitBreakerSettings">TBD</param>
-        /// <param name="replayFilterSettings">TBD</param>
-        /// <param name="namingConventions">TBD</param>
+        /// <param name="connectionString">The connection string used to connect to the database.</param>
+        /// <param name="maxConcurrentOperations">The maximum number of batch operations allowed to be executed at the same time.</param>
+        /// <param name="maxBatchSize">The maximum size of single batch of operations to be executed over a single <see cref="DbConnection"/>.</param>
+        /// <param name="maxBufferSize">The maximum size of requests stored in journal buffer.</param>
+        /// <param name="autoInitialize">
+        /// If set to <c>true</c>, the journal executes all SQL scripts stored under the
+        /// <see cref="BatchingSqlJournal{TConnection,TCommand}.Initializers"/> collection prior
+        /// to starting executing any requests.
+        /// </param>
+        /// <param name="connectionTimeout">The maximum time given for executed <see cref="DbCommand"/> to complete.</param>
+        /// <param name="isolationLevel">The isolation level of transactions used during query execution.</param>
+        /// <param name="circuitBreakerSettings">
+        /// The settings used by the <see cref="CircuitBreaker"/> when for executing request batches.
+        /// </param>
+        /// <param name="replayFilterSettings">The settings used when replaying events from database back to the persistent actors.</param>
+        /// <param name="namingConventions">The naming conventions used by the database to construct valid SQL statements.</param>
+        /// <param name="defaultSerializer">The serializer used when no specific type matching can be found.</param>
         public BatchingSqliteJournalSetup(string connectionString, int maxConcurrentOperations, int maxBatchSize, int maxBufferSize, bool autoInitialize, 
-            TimeSpan connectionTimeout, IsolationLevel isolationLevel, CircuitBreakerSettings circuitBreakerSettings, ReplayFilterSettings replayFilterSettings, QueryConfiguration namingConventions) 
-            : base(connectionString, maxConcurrentOperations, maxBatchSize, maxBufferSize, autoInitialize, connectionTimeout, isolationLevel, circuitBreakerSettings, replayFilterSettings, namingConventions)
+            TimeSpan connectionTimeout, IsolationLevel isolationLevel, CircuitBreakerSettings circuitBreakerSettings, ReplayFilterSettings replayFilterSettings, QueryConfiguration namingConventions, string defaultSerializer) 
+            : base(connectionString, maxConcurrentOperations, maxBatchSize, maxBufferSize, autoInitialize, connectionTimeout, isolationLevel, circuitBreakerSettings, replayFilterSettings, namingConventions, defaultSerializer)
         {
         }
     }

src/contrib/persistence/Akka.Persistence.Sqlite/Journal/SqliteJournal.cs

@@ -41,7 +41,8 @@ public SqliteJournal(Config journalConfig) : base(journalConfig.WithFallback(Ext
                 isDeletedColumnName: "is_deleted",
                 tagsColumnName: "tags",
                 orderingColumnName: "ordering",
-                timeout: config.GetTimeSpan("connection-timeout")), 
+                timeout: config.GetTimeSpan("connection-timeout"),
+                defaultSerializer: config.GetString("serializer")), 
                     Context.System.Serialization, 
                     GetTimestampProvider(config.GetString("timestamp-provider")));
         }

src/contrib/persistence/Akka.Persistence.Sqlite/Snapshot/SqliteSnapshotStore.cs

@@ -115,7 +115,8 @@ public SqliteSnapshotStore(Config snapshotConfig) : base(snapshotConfig)
                 payloadColumnName: "payload",
                 manifestColumnName: "manifest",
                 timestampColumnName: "created_at",
-                timeout: config.GetTimeSpan("connection-timeout")), 
+                timeout: config.GetTimeSpan("connection-timeout"),
+                defaultSerializer: config.GetString("serializer")), 
                 Context.System.Serialization);
         }
 

src/core/Akka.API.Tests/CoreAPISpec.ApproveCore.approved.txt

@@ -4446,8 +4446,8 @@ namespace Akka.Serialization
         public void AddSerializer(Akka.Serialization.Serializer serializer) { }
         public object Deserialize(byte[] bytes, int serializerId, System.Type type) { }
         public object Deserialize(byte[] bytes, int serializerId, string manifest) { }
-        public Akka.Serialization.Serializer FindSerializerFor(object obj) { }
-        public Akka.Serialization.Serializer FindSerializerForType(System.Type objectType) { }
+        public Akka.Serialization.Serializer FindSerializerFor(object obj, string defaultSerializerName = null) { }
+        public Akka.Serialization.Serializer FindSerializerForType(System.Type objectType, string defaultSerializerName = null) { }
         public static string SerializedActorPath(Akka.Actor.IActorRef actorRef) { }
         public static T SerializeWithTransport<T>(Akka.Actor.ActorSystem system, Akka.Actor.Address address, System.Func<T> action) { }
     }

src/core/Akka.Persistence.Tests/Akka.Persistence.Tests.csproj

@@ -7,6 +7,7 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <ProjectReference Include="..\..\contrib\serializers\Akka.Serialization.Hyperion\Akka.Serialization.Hyperion.csproj" />
     <ProjectReference Include="..\Akka.Persistence\Akka.Persistence.csproj" />
     <ProjectReference Include="..\Akka.Remote\Akka.Remote.csproj" />
     <ProjectReference Include="..\Akka.Tests.Shared.Internals\Akka.Tests.Shared.Internals.csproj" />

src/core/Akka.Persistence.Tests/Serialization/SerializerSpec.cs → src/core/Akka.Persistence.Tests/Serialization/MessageSerializerRemotingSpec.cs

@@ -12,6 +12,7 @@
 using Akka.Actor;
 using Akka.Configuration;
 using Akka.TestKit;
+using FluentAssertions;
 using Xunit;
 
 namespace Akka.Persistence.Tests.Serialization
@@ -22,12 +23,12 @@ internal class Configs
 akka.actor {
   serializers {
     my-payload = ""Akka.Persistence.Tests.Serialization.MyPayloadSerializer, Akka.Persistence.Tests""
-    my-payload2 = ""Akka.Persistence.Tests.Serialization.MyPayload2Serializer, Akka.Persistence.Tests""
     old-payload = ""Akka.Persistence.Tests.Serialization.OldPayloadSerializer, Akka.Persistence.Tests""
+    testserializer = ""Akka.Serialization.HyperionSerializer, Akka.Serialization.Hyperion""
   }
   serialization-bindings {
     ""Akka.Persistence.Tests.Serialization.MyPayload, Akka.Persistence.Tests"" = my-payload
-    ""Akka.Persistence.Tests.Serialization.MyPayload2, Akka.Persistence.Tests"" = my-payload2
+    ""System.Object"" = testserializer
     # this entry was used when creating the data for the test
     # ""deserialize data when class is removed""
     #""Akka.Persistence.Tests.Serialization.OldPayload, Akka.Persistence.Tests"" = old-payload
@@ -94,7 +95,7 @@ public override int GetHashCode()
     }
 
 	// TODO: temporary disabled
-    public abstract class MessageSerializerRemotingSpec : AkkaSpec
+    public class MessageSerializerRemotingSpec : AkkaSpec
     {
         internal class LocalActor : ActorBase
         {
@@ -186,7 +187,7 @@ public void MessageSerializer_should_custom_serialize_Persistent_messages_during
             // this also verifies serialization of Persistent.Sender,
             // because the RemoteActor will reply to the Persistent.Sender
             _localActor.Tell(new Persistent(new MyPayload("a"), sender: TestActor));
-            ExpectMsg("p.a.");
+            ExpectMsg("pa");
         }
 
         [Fact]
@@ -195,8 +196,10 @@ public void MessageSerializer_should_custom_serialize_AtomicWrite_messages_durin
             var p1 = new Persistent(new MyPayload("a"), sender: TestActor);
             var p2 = new Persistent(new MyPayload("b"), sender: TestActor);
             _localActor.Tell(new AtomicWrite(ImmutableList.Create(new IPersistentRepresentation[] {p1, p2})));
-            ExpectMsg("p.a.");
-            ExpectMsg("p.b.");
+            Within(5.Seconds(), () => { 
+                ExpectMsg("pa");
+                ExpectMsg("pb");
+            });
         }
 
         [Fact]