-
Notifications
You must be signed in to change notification settings - Fork 11
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: use the correct snapshot logic
- Loading branch information
Showing
6 changed files
with
351 additions
and
117 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,51 @@ | ||
// ----------------------------------------------------------------------- | ||
// <copyright file="AkkaDataConnection.cs" company="Akka.NET Project"> | ||
// Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net> | ||
// </copyright> | ||
// ----------------------------------------------------------------------- | ||
|
||
using System; | ||
using System.Threading.Tasks; | ||
using LinqToDB; | ||
using LinqToDB.Data; | ||
using LinqToDB.Data.RetryPolicy; | ||
|
||
namespace Akka.Persistence.Sql.Db | ||
{ | ||
public class AkkaDataConnection : IDisposable, IAsyncDisposable | ||
{ | ||
private readonly string _providerName; | ||
|
||
public AkkaDataConnection( | ||
string providerName, | ||
DataConnection connection) | ||
{ | ||
_providerName = providerName.ToLower(); | ||
Db = connection; | ||
} | ||
|
||
public bool UseDateTime => | ||
!_providerName.Contains("sqlite") && | ||
!_providerName.Contains("postgresql"); | ||
|
||
public DataConnection Db { get; } | ||
|
||
public IRetryPolicy RetryPolicy | ||
{ | ||
get => Db.RetryPolicy; | ||
set => Db.RetryPolicy = value; | ||
} | ||
|
||
public ValueTask DisposeAsync() | ||
=> Db.DisposeAsync(); | ||
|
||
public void Dispose() | ||
=> Db.Dispose(); | ||
|
||
public AkkaDataConnection Clone() | ||
=> new(_providerName, (DataConnection)Db.Clone()); | ||
|
||
public ITable<T> GetTable<T>() where T : class | ||
=> Db.GetTable<T>(); | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
92 changes: 92 additions & 0 deletions
92
src/Akka.Persistence.Sql/Snapshot/ByteArrayLongSnapshotSerializer.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,92 @@ | ||
// ----------------------------------------------------------------------- | ||
// <copyright file="ByteArrayLongSnapshotSerializer.cs" company="Akka.NET Project"> | ||
// Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net> | ||
// </copyright> | ||
// ----------------------------------------------------------------------- | ||
|
||
using System; | ||
using Akka.Persistence.Sql.Config; | ||
using Akka.Serialization; | ||
using Akka.Util; | ||
|
||
namespace Akka.Persistence.Sql.Snapshot | ||
{ | ||
public class ByteArrayLongSnapshotSerializer : ISnapshotSerializer<LongSnapshotRow> | ||
{ | ||
private readonly SnapshotConfig _config; | ||
private readonly Akka.Serialization.Serialization _serialization; | ||
|
||
public ByteArrayLongSnapshotSerializer( | ||
Akka.Serialization.Serialization serialization, | ||
SnapshotConfig config) | ||
{ | ||
_serialization = serialization; | ||
_config = config; | ||
} | ||
|
||
public Try<LongSnapshotRow> Serialize(SnapshotMetadata metadata, object snapshot) | ||
=> Try<LongSnapshotRow>.From(() => ToSnapshotEntry(metadata, snapshot)); | ||
|
||
public Try<SelectedSnapshot> Deserialize(LongSnapshotRow t) | ||
=> Try<SelectedSnapshot>.From(() => ReadSnapshot(t)); | ||
|
||
protected SelectedSnapshot ReadSnapshot(LongSnapshotRow reader) | ||
{ | ||
var metadata = new SnapshotMetadata( | ||
reader.PersistenceId, | ||
reader.SequenceNumber, | ||
new DateTime(reader.Created)); | ||
|
||
var snapshot = GetSnapshot(reader); | ||
|
||
return new SelectedSnapshot(metadata, snapshot); | ||
} | ||
|
||
protected object GetSnapshot(LongSnapshotRow reader) | ||
{ | ||
var manifest = reader.Manifest; | ||
var binary = reader.Payload; | ||
|
||
if (reader.SerializerId is null) | ||
{ | ||
var type = Type.GetType(manifest, true); | ||
|
||
// TODO: hack. Replace when https://github.com/akkadotnet/akka.net/issues/3811 | ||
return Akka.Serialization.Serialization.WithTransport( | ||
system: _serialization.System, | ||
state: (serializer: _serialization.FindSerializerForType(type, _config.DefaultSerializer), binary, type), | ||
action: state => state.serializer.FromBinary(state.binary, state.type)); | ||
} | ||
|
||
var serializerId = reader.SerializerId.Value; | ||
return _serialization.Deserialize(binary, serializerId, manifest); | ||
} | ||
|
||
private LongSnapshotRow ToSnapshotEntry(SnapshotMetadata metadata, object snapshot) | ||
{ | ||
var snapshotType = snapshot.GetType(); | ||
var serializer = _serialization.FindSerializerForType(snapshotType, _config.DefaultSerializer); | ||
var binary = Akka.Serialization.Serialization.WithTransport( | ||
system: _serialization.System, | ||
state: (serializer, snapshot), | ||
action: state => state.serializer.ToBinary(state.snapshot)); | ||
|
||
var manifest = serializer switch | ||
{ | ||
SerializerWithStringManifest stringManifest => stringManifest.Manifest(snapshot), | ||
{ IncludeManifest: true } => snapshotType.TypeQualifiedName(), | ||
_ => string.Empty | ||
}; | ||
|
||
return new LongSnapshotRow | ||
{ | ||
PersistenceId = metadata.PersistenceId, | ||
SequenceNumber = metadata.SequenceNr, | ||
Created = metadata.Timestamp.Ticks, | ||
Manifest = manifest, | ||
Payload = binary, | ||
SerializerId = serializer.Identifier | ||
}; | ||
} | ||
} | ||
} |
Oops, something went wrong.