Attempt to fix JSON and JSONB payload DbType (failed)

src/Akka.Persistence.PostgreSql.Tests/Serialization/PostgreSqlJournalJsonBSerializationSpec.cs

@@ -0,0 +1,104 @@
+//-----------------------------------------------------------------------
+// <copyright file="PostgreSqlJournalSerializationSpec.cs" company="Akka.NET Project">
+//     Copyright (C) 2009-2021 Lightbend Inc. <http://www.lightbend.com>
+//     Copyright (C) 2013-2021 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+using System.Collections.Generic;
+using Akka.Actor;
+using Akka.Configuration;
+using Akka.Persistence.TCK.Serialization;
+using Akka.Util.Internal;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Akka.Persistence.PostgreSql.Tests.Serialization
+{
+    [Collection("PostgreSqlSpec")]
+    public class PostgreSqlJournalJsonBSerializationSpec : JournalSerializationSpec
+    {
+        public PostgreSqlJournalJsonBSerializationSpec(ITestOutputHelper output, PostgresFixture fixture)
+            : base(CreateSpecConfig(fixture), "PostgreSqlJournalSerializationSpec", output)
+        {
+        }
+
+        private static Config CreateSpecConfig(PostgresFixture fixture)
+        {
+            //need to make sure db is created before the tests start
+            DbUtils.Initialize(fixture);
+
+            return ConfigurationFactory.ParseString($@"
+                akka.persistence {{
+                    publish-plugin-commands = on
+                    journal {{
+                        plugin = ""akka.persistence.journal.postgresql""
+                        postgresql {{
+                            stored-as = jsonb
+                            connection-string = ""{DbUtils.ConnectionString}""
+                            auto-initialize = on
+                        }}
+                    }}
+                    snapshot-store {{
+                        plugin = ""akka.persistence.snapshot-store.postgresql""
+                        postgresql {{
+                            stored-as = jsonb
+                            connection-string = ""{DbUtils.ConnectionString}""
+                            auto-initialize = on
+                        }}
+                    }}
+                }}
+                akka.test.single-expect-default = 10s")
+                .WithFallback(PostgreSqlPersistence.DefaultConfiguration());
+        }
+
+        [Fact(Skip = "Sql plugin does not support EventAdapter.Manifest")]
+        public override void Journal_should_serialize_Persistent_with_EventAdapter_manifest()
+        {
+        }
+
+        [Fact]
+        public override void Journal_should_serialize_Persistent_with_string_manifest()
+        {
+            var probe = CreateTestProbe();
+            var persistentEvent = new Persistent(new TestJournal.MyPayload2("b", 5), 1L, Pid, null, false, null, WriterGuid);
+
+            var messages = new List<AtomicWrite>
+            {
+                new AtomicWrite(persistentEvent)
+            };
+
+            Journal.Tell(new WriteMessages(messages, probe.Ref, ActorInstanceId));
+            probe.ExpectMsg<WriteMessagesSuccessful>();
+            probe.ExpectMsg<WriteMessageSuccess>(m => m.ActorInstanceId == ActorInstanceId && m.Persistent.PersistenceId == Pid);
+
+            Journal.Tell(new ReplayMessages(0, long.MaxValue, long.MaxValue, Pid, probe.Ref));
+            probe.ExpectMsg<ReplayedMessage>(s => s.Persistent.PersistenceId == persistentEvent.PersistenceId
+                                                  && s.Persistent.SequenceNr == persistentEvent.SequenceNr
+                                                  && s.Persistent.Payload.AsInstanceOf<TestJournal.MyPayload2>().Data.Equals("b"));
+            probe.ExpectMsg<RecoverySuccess>();
+        }
+
+        [Fact]
+        public override void Journal_should_serialize_Persistent()
+        {
+            var probe = CreateTestProbe();
+            var persistentEvent = new Persistent(new TestJournal.MyPayload("a"), 1L, Pid, null, false, null, WriterGuid);
+
+            var messages = new List<AtomicWrite>
+            {
+                new AtomicWrite(persistentEvent)
+            };
+
+            Journal.Tell(new WriteMessages(messages, probe.Ref, ActorInstanceId));
+            probe.ExpectMsg<WriteMessagesSuccessful>();
+            probe.ExpectMsg<WriteMessageSuccess>(m => m.ActorInstanceId == ActorInstanceId && m.Persistent.PersistenceId == Pid);
+
+            Journal.Tell(new ReplayMessages(0, long.MaxValue, long.MaxValue, Pid, probe.Ref));
+            probe.ExpectMsg<ReplayedMessage>(s => s.Persistent.PersistenceId == Pid
+                                                  && s.Persistent.SequenceNr == persistentEvent.SequenceNr
+                                                  && s.Persistent.Payload.AsInstanceOf<TestJournal.MyPayload>().Data.Equals("a"));
+            probe.ExpectMsg<RecoverySuccess>();
+        }
+    }
+}

src/Akka.Persistence.PostgreSql.Tests/Serialization/PostgreSqlJournalJsonSerializationSpec.cs

@@ -0,0 +1,106 @@
+//-----------------------------------------------------------------------
+// <copyright file="PostgreSqlJournalSerializationSpec.cs" company="Akka.NET Project">
+//     Copyright (C) 2009-2021 Lightbend Inc. <http://www.lightbend.com>
+//     Copyright (C) 2013-2021 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+using System.Collections.Generic;
+using Akka.Actor;
+using Akka.Configuration;
+using Akka.Persistence.TCK.Serialization;
+using Akka.Util.Internal;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Akka.Persistence.PostgreSql.Tests.Serialization
+{
+    [Collection("PostgreSqlSpec")]
+    public class PostgreSqlJournalJsonSerializationSpec : JournalSerializationSpec
+    {
+        public PostgreSqlJournalJsonSerializationSpec(ITestOutputHelper output, PostgresFixture fixture)
+            : base(CreateSpecConfig(fixture), "PostgreSqlJournalSerializationSpec", output)
+        {
+        }
+
+        private static Config CreateSpecConfig(PostgresFixture fixture)
+        {
+            //need to make sure db is created before the tests start
+            DbUtils.Initialize(fixture);
+
+            return ConfigurationFactory.ParseString($@"
+                akka.persistence {{
+                    publish-plugin-commands = on
+                    journal {{
+                        plugin = ""akka.persistence.journal.postgresql""
+                        postgresql {{
+                            stored-as = json
+                            connection-string = ""{DbUtils.ConnectionString}""
+                            auto-initialize = on
+                        }}
+                    }}
+                    snapshot-store {{
+                        plugin = ""akka.persistence.snapshot-store.postgresql""
+                        postgresql {{
+                            stored-as = json
+                            connection-string = ""{DbUtils.ConnectionString}""
+                            auto-initialize = on
+                        }}
+                    }}
+                }}
+                akka.test.single-expect-default = 10s")
+                .WithFallback(PostgreSqlPersistence.DefaultConfiguration());
+        }
+
+        [Fact(Skip = "Sql plugin does not support EventAdapter.Manifest")]
+        public override void Journal_should_serialize_Persistent_with_EventAdapter_manifest()
+        {
+        }
+
+
+        [Fact]
+        public override void Journal_should_serialize_Persistent_with_string_manifest()
+        {
+            var probe = CreateTestProbe();
+            var persistentEvent = new Persistent(new TestJournal.MyPayload2("b", 5), 1L, Pid, null, false, null, WriterGuid);
+
+            var messages = new List<AtomicWrite>
+            {
+                new AtomicWrite(persistentEvent)
+            };
+
+            Journal.Tell(new WriteMessages(messages, probe.Ref, ActorInstanceId));
+            probe.ExpectMsg<WriteMessagesSuccessful>();
+            probe.ExpectMsg<WriteMessageSuccess>(m => m.ActorInstanceId == ActorInstanceId && m.Persistent.PersistenceId == Pid);
+
+            Journal.Tell(new ReplayMessages(0, long.MaxValue, long.MaxValue, Pid, probe.Ref));
+            probe.ExpectMsg<ReplayedMessage>(s => s.Persistent.PersistenceId == persistentEvent.PersistenceId
+                                                  && s.Persistent.SequenceNr == persistentEvent.SequenceNr
+                                                  && s.Persistent.Payload.AsInstanceOf<TestJournal.MyPayload2>().Data.Equals("b"));
+            probe.ExpectMsg<RecoverySuccess>();
+        }
+
+        [Fact]
+        public override void Journal_should_serialize_Persistent()
+        {
+            var probe = CreateTestProbe();
+            var persistentEvent = new Persistent(new TestJournal.MyPayload("a"), 1L, Pid, null, false, null, WriterGuid);
+
+            var messages = new List<AtomicWrite>
+            {
+                new AtomicWrite(persistentEvent)
+            };
+
+            Journal.Tell(new WriteMessages(messages, probe.Ref, ActorInstanceId));
+            probe.ExpectMsg<WriteMessagesSuccessful>();
+            probe.ExpectMsg<WriteMessageSuccess>(m => m.ActorInstanceId == ActorInstanceId && m.Persistent.PersistenceId == Pid);
+
+            Journal.Tell(new ReplayMessages(0, long.MaxValue, long.MaxValue, Pid, probe.Ref));
+            probe.ExpectMsg<ReplayedMessage>(s => s.Persistent.PersistenceId == Pid
+                                                  && s.Persistent.SequenceNr == persistentEvent.SequenceNr
+                                                  && s.Persistent.Payload.AsInstanceOf<TestJournal.MyPayload>().Data.Equals("a"));
+            probe.ExpectMsg<RecoverySuccess>();
+        }
+
+    }
+}

src/Akka.Persistence.PostgreSql.Tests/Serialization/PostgreSqlJournalSerializationSpec.cs

@@ -25,22 +25,28 @@ private static Config CreateSpecConfig(PostgresFixture fixture)
             //need to make sure db is created before the tests start
             DbUtils.Initialize(fixture);
 
-            return ConfigurationFactory.ParseString(@"
-                akka.persistence {
+            return ConfigurationFactory.ParseString($@"
+                akka.persistence {{
                     publish-plugin-commands = on
-                    journal {
+                    journal {{
                         plugin = ""akka.persistence.journal.postgresql""
-                        postgresql {
-                            class = ""Akka.Persistence.PostgreSql.Journal.PostgreSqlJournal, Akka.Persistence.PostgreSql""
-                            plugin-dispatcher = ""akka.actor.default-dispatcher""
-                            table-name = event_journal
-                            schema-name = public
+                        postgresql {{
+                            stored-as = bytea
+                            connection-string = ""{DbUtils.ConnectionString}""
                             auto-initialize = on
-                            connection-string = """ + DbUtils.ConnectionString + @"""
-                        }
-                    }
-                }
-                akka.test.single-expect-default = 10s");
+                        }}
+                    }}
+                    snapshot-store {{
+                        plugin = ""akka.persistence.snapshot-store.postgresql""
+                        postgresql {{
+                            stored-as = bytea
+                            connection-string = ""{DbUtils.ConnectionString}""
+                            auto-initialize = on
+                        }}
+                    }}
+                }}
+                akka.test.single-expect-default = 10s")
+                .WithFallback(PostgreSqlPersistence.DefaultConfiguration());
         }
 
         [Fact(Skip = "Sql plugin does not support EventAdapter.Manifest")]

src/Akka.Persistence.PostgreSql.Tests/Serialization/PostgreSqlSnapshotStoreJsonBSerializationSpec.cs

@@ -0,0 +1,90 @@
+//-----------------------------------------------------------------------
+// <copyright file="PostgreSqlSnapshotStoreSerializationSpec.cs" company="Akka.NET Project">
+//     Copyright (C) 2009-2021 Lightbend Inc. <http://www.lightbend.com>
+//     Copyright (C) 2013-2021 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+using Akka.Actor;
+using Akka.Configuration;
+using Akka.Persistence.TCK.Serialization;
+using Akka.Util.Internal;
+using FluentAssertions;
+using Newtonsoft.Json;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Akka.Persistence.PostgreSql.Tests.Serialization
+{
+    [Collection("PostgreSqlSpec")]
+    public class PostgreSqlSnapshotStoreJsonBSerializationSpec : SnapshotStoreSerializationSpec
+    {
+        public PostgreSqlSnapshotStoreJsonBSerializationSpec(ITestOutputHelper output, PostgresFixture fixture)
+            : base(CreateSpecConfig(fixture), "PostgreSqlSnapshotStoreSerializationSpec", output)
+        {
+        }
+
+        private static Config CreateSpecConfig(PostgresFixture fixture)
+        {
+            //need to make sure db is created before the tests start
+            DbUtils.Initialize(fixture);
+
+            return ConfigurationFactory.ParseString($@"
+                akka.persistence {{
+                    publish-plugin-commands = on
+                    journal {{
+                        plugin = ""akka.persistence.journal.postgresql""
+                        postgresql {{
+                            stored-as = jsonb
+                            connection-string = ""{DbUtils.ConnectionString}""
+                            auto-initialize = on
+                        }}
+                    }}
+                    snapshot-store {{
+                        plugin = ""akka.persistence.snapshot-store.postgresql""
+                        postgresql {{
+                            stored-as = jsonb
+                            connection-string = ""{DbUtils.ConnectionString}""
+                            auto-initialize = on
+                        }}
+                    }}
+                }}
+                akka.test.single-expect-default = 10s")
+                .WithFallback(PostgreSqlPersistence.DefaultConfiguration());
+        }
+
+        [Fact]
+        public override void SnapshotStore_should_serialize_Payload()
+        {
+            var probe = CreateTestProbe();
+
+            var snapshot = new Test.MySnapshot("a");
+
+            var metadata = new SnapshotMetadata(Pid, 1);
+            SnapshotStore.Tell(new SaveSnapshot(metadata, snapshot), probe.Ref);
+            probe.ExpectMsg<SaveSnapshotSuccess>();
+
+            SnapshotStore.Tell(new LoadSnapshot(Pid, SnapshotSelectionCriteria.Latest, long.MaxValue), probe.Ref);
+            probe.ExpectMsg<LoadSnapshotResult>(s => 
+                s.Snapshot.Snapshot is Test.MySnapshot
+                && s.Snapshot.Snapshot.AsInstanceOf<Test.MySnapshot>().Data.Equals("a"));
+        }
+
+        [Fact]
+        public override void SnapshotStore_should_serialize_Payload_with_string_manifest()
+        {
+            var probe = CreateTestProbe();
+
+            var snapshot = new Test.MySnapshot2("a");
+
+            var metadata = new SnapshotMetadata(Pid, 1);
+            SnapshotStore.Tell(new SaveSnapshot(metadata, snapshot), probe.Ref);
+            probe.ExpectMsg<SaveSnapshotSuccess>();
+
+            SnapshotStore.Tell(new LoadSnapshot(Pid, SnapshotSelectionCriteria.Latest, long.MaxValue), probe.Ref);
+            probe.ExpectMsg<LoadSnapshotResult>(s => 
+                s.Snapshot.Snapshot is Test.MySnapshot2
+                && s.Snapshot.Snapshot.AsInstanceOf<Test.MySnapshot2>().Data.Equals("a"));
+        }
+    }
+}