1+ // -----------------------------------------------------------------------
2+ // <copyright file="CustomObjectSerializerSpec.cs" company="Akka.NET Project">
3+ // Copyright (C) 2009-2022 Lightbend Inc. <http://www.lightbend.com>
4+ // Copyright (C) 2013-2022 .NET Foundation <https://github.com/akkadotnet/akka.net>
5+ // </copyright>
6+ // -----------------------------------------------------------------------
7+
8+ using System ;
9+ using System . IO ;
10+ using System . Text ;
11+ using System . Threading . Tasks ;
12+ using Akka . Actor ;
13+ using Akka . Configuration ;
14+ using Akka . Serialization ;
15+ using FluentAssertions ;
16+ using Microsoft . Data . Sqlite ;
17+ using Xunit ;
18+ using Xunit . Abstractions ;
19+
20+ namespace Akka . Persistence . Sqlite . Tests
21+ {
22+ public class CustomObjectSerializerSpec : Akka . TestKit . Xunit2 . TestKit , IAsyncLifetime
23+ {
24+ private static readonly string ConnectionString ;
25+ private static readonly Config Config ;
26+ static CustomObjectSerializerSpec ( )
27+ {
28+ var filename = $ "AkkaSqlite-{ Guid . NewGuid ( ) } .db";
29+ File . Copy ( "./data/Sqlite.CustomObject.db" , $ "{ filename } .db") ;
30+
31+ ConnectionString = $ "DataSource={ filename } .db";
32+ Config = ConfigurationFactory . ParseString ( $@ "
33+ akka.actor {{
34+ serializers {{
35+ mySerializer = ""{ typeof ( MySerializer ) . AssemblyQualifiedName } ""
36+ }}
37+ serialization-bindings {{
38+ ""System.Object"" = mySerializer
39+ }}
40+ }}
41+
42+ akka.persistence {{
43+ journal {{
44+ plugin = ""akka.persistence.journal.sqlite""
45+ sqlite {{
46+ connection-string = ""{ ConnectionString } ""
47+ auto-initialize = on
48+ }}
49+ }}
50+ snapshot-store {{
51+ plugin = ""akka.persistence.snapshot-store.sqlite""
52+ sqlite {{
53+ connection-string = ""{ ConnectionString } ""
54+ auto-initialize = on
55+ }}
56+ }}
57+ }}" ) . WithFallback ( SqlitePersistence . DefaultConfiguration ( ) ) ;
58+ }
59+
60+ public CustomObjectSerializerSpec ( ITestOutputHelper helper )
61+ : base ( Config , nameof ( CustomObjectSerializerSpec ) , helper )
62+ {
63+ }
64+
65+ [ Fact ( DisplayName = "Persistence.Sql should use custom serializer for object type" ) ]
66+ public async Task CustomSerializerTest ( )
67+ {
68+ var probe = CreateTestProbe ( ) ;
69+
70+ // Sanity check to see that the system should serialize object type using MySerializer
71+ var serializer = Sys . Serialization . FindSerializerForType ( typeof ( Persisted ) ) ;
72+ serializer . Should ( ) . BeOfType < MySerializer > ( ) ;
73+
74+ var actor = Sys . ActorOf ( Props . Create ( ( ) => new PersistedActor ( "a" , probe ) ) ) ;
75+ probe . ExpectMsg ( "recovered" ) ;
76+ actor . Tell ( new Persisted ( "a" ) , probe ) ;
77+ probe . ExpectMsg ( new Persisted ( "a" ) ) ;
78+
79+ // Read the database directly, make sure that we're using the correct object type serializer
80+ var conn = new SqliteConnection ( ConnectionString ) ;
81+ conn . Open ( ) ;
82+ const string sql = "SELECT ej.serializer_id FROM event_journal ej WHERE ej.persistence_id = 'a'" ;
83+ await using var cmd = new SqliteCommand ( sql , conn ) ;
84+ var record = await cmd . ExecuteReaderAsync ( ) ;
85+ await record . ReadAsync ( ) ;
86+
87+ // In the bug this fails, the serializer id is JSON id instead of MySerializer id
88+ record [ 0 ] . Should ( ) . Be ( 9999 ) ;
89+ }
90+
91+ [ Fact ( DisplayName = "Persistence.Sql should be able to read legacy data" ) ]
92+ public void LegacyDataTest ( )
93+ {
94+ var probe = CreateTestProbe ( ) ;
95+ var actor = Sys . ActorOf ( Props . Create ( ( ) => new PersistedActor ( "old" , probe ) ) ) ;
96+ probe . ExpectMsg ( new Persisted ( "old" ) ) ;
97+ probe . ExpectMsg ( "recovered" ) ;
98+ }
99+
100+ public Task InitializeAsync ( )
101+ {
102+ if ( File . Exists ( "AkkaSqlite.db" ) )
103+ File . Delete ( "AkkaSqlite.db" ) ;
104+ return Task . CompletedTask ;
105+ }
106+
107+ public Task DisposeAsync ( )
108+ {
109+ return Task . CompletedTask ;
110+ }
111+ }
112+
113+ internal sealed class Persisted : IEquatable < Persisted >
114+ {
115+ public Persisted ( string payload )
116+ {
117+ Payload = payload ;
118+ }
119+
120+ public string Payload { get ; }
121+
122+ public bool Equals ( Persisted other )
123+ {
124+ if ( ReferenceEquals ( null , other ) ) return false ;
125+ if ( ReferenceEquals ( this , other ) ) return true ;
126+ return Payload == other . Payload ;
127+ }
128+
129+ public override bool Equals ( object obj )
130+ {
131+ return ReferenceEquals ( this , obj ) || obj is Persisted other && Equals ( other ) ;
132+ }
133+
134+ public override int GetHashCode ( )
135+ {
136+ return ( Payload != null ? Payload . GetHashCode ( ) : 0 ) ;
137+ }
138+ }
139+
140+ internal class MySerializer : Serializer
141+ {
142+ public MySerializer ( ExtendedActorSystem system ) : base ( system )
143+ {
144+ }
145+
146+ public override bool IncludeManifest { get { return true ; } }
147+ public override int Identifier { get { return 9999 ; } }
148+
149+ public override byte [ ] ToBinary ( object obj )
150+ {
151+ return Encoding . UTF8 . GetBytes ( obj . ToString ( ) ) ;
152+ }
153+
154+ public override object FromBinary ( byte [ ] bytes , Type type )
155+ {
156+ return Encoding . UTF8 . GetString ( bytes ) ;
157+ }
158+ }
159+
160+ internal sealed class PersistedActor : UntypedPersistentActor
161+ {
162+ private readonly IActorRef _probe ;
163+
164+ public PersistedActor ( string persistenceId , IActorRef probe )
165+ {
166+ PersistenceId = persistenceId ;
167+ _probe = probe ;
168+ }
169+
170+ public override string PersistenceId { get ; }
171+
172+ protected override void OnCommand ( object message )
173+ {
174+ var sender = Sender ;
175+ Persist ( message , _ =>
176+ {
177+ sender . Tell ( message ) ;
178+ } ) ;
179+ }
180+
181+ protected override void OnRecover ( object message )
182+ {
183+ switch ( message )
184+ {
185+ case Persisted msg :
186+ _probe . Tell ( msg ) ;
187+ break ;
188+ case RecoveryCompleted _:
189+ _probe . Tell ( "recovered" ) ;
190+ break ;
191+ }
192+ }
193+ }
194+ }
0 commit comments